CheckpointStatsTracker { public CheckpointStatsSnapshot createSnapshot() { CheckpointStatsSnapshot snapshot = latestSnapshot; if (dirty && statsReadWriteLock.tryLock()) { try { snapshot = new CheckpointStatsSnapshot( counts.createSnapshot(), summary.createSnapshot(), history.createSnapshot(), latestRestoredCheckpoint); latestSnapshot = snapshot; dirty = false; } finally { statsReadWriteLock.unlock(); } } return snapshot; } CheckpointStatsTracker(
		int numRememberedCheckpoints,
		List<ExecutionJobVertex> jobVertices,
		CheckpointCoordinatorConfiguration jobCheckpointingConfiguration,
		MetricGroup metricGroup); CheckpointCoordinatorConfiguration getJobCheckpointingConfiguration(); CheckpointStatsSnapshot createSnapshot();  }
@Test public void testCreateSnapshot() throws Exception { ExecutionJobVertex jobVertex = mock(ExecutionJobVertex.class); when(jobVertex.getJobVertexId()).thenReturn(new JobVertexID()); when(jobVertex.getParallelism()).thenReturn(1); CheckpointStatsTracker tracker = new CheckpointStatsTracker( 10, Collections.singletonList(jobVertex), mock(CheckpointCoordinatorConfiguration.class), new UnregisteredMetricsGroup()); CheckpointStatsSnapshot snapshot1 = tracker.createSnapshot(); PendingCheckpointStats pending = tracker.reportPendingCheckpoint( 0, 1, CheckpointProperties.forStandardCheckpoint()); pending.reportSubtaskStats(jobVertex.getJobVertexId(), createSubtaskStats(0)); CheckpointStatsSnapshot snapshot2 = tracker.createSnapshot(); assertNotEquals(snapshot1, snapshot2); assertEquals(snapshot2, tracker.createSnapshot()); pending.reportCompletedCheckpoint(null); CheckpointStatsSnapshot snapshot3 = tracker.createSnapshot(); assertNotEquals(snapshot2, snapshot3); tracker.reportRestoredCheckpoint(new RestoredCheckpointStats(12, CheckpointProperties.forStandardCheckpoint(), 12, null)); CheckpointStatsSnapshot snapshot4 = tracker.createSnapshot(); assertNotEquals(snapshot3, snapshot4); assertEquals(snapshot4, tracker.createSnapshot()); }