flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
/**
	 * Tests that the completed checkpoint store can retrieve all checkpoints stored in ZooKeeper
	 * and ignores those which cannot be retrieved via their state handles.
	 */
Tests that the completed checkpoint store can retrieve all checkpoints stored in ZooKeeper
and ignores those which cannot be retrieved via their state handles.
Tests that the completed checkpoint store can retrieve all checkpoints stored in ZooKeeper and ignores those which cannot be retrieved via their state handles .
@Test
	public void testCheckpointRecovery() throws Exception {
		final List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>> checkpointsInZooKeeper = new ArrayList<>(4);
		final CompletedCheckpoint completedCheckpoint1 = new CompletedCheckpoint(
			new JobID(),
			1L,
			1L,
			1L,
			new HashMap<>(),
			null,
			CheckpointProperties.forStandardCheckpoint(),
			null, null);
		final CompletedCheckpoint completedCheckpoint2 = new CompletedCheckpoint(
			new JobID(),
			2L,
			2L,
			2L,
			new HashMap<>(),
			null,
			CheckpointProperties.forStandardCheckpoint(),
			null, null);
		final Collection<Long> expectedCheckpointIds = new HashSet<>(2);
		expectedCheckpointIds.add(1L);
		expectedCheckpointIds.add(2L);
		final RetrievableStateHandle<CompletedCheckpoint> failingRetrievableStateHandle = mock(RetrievableStateHandle.class);
		when(failingRetrievableStateHandle.retrieveState()).thenThrow(new IOException("Test exception"));
		final RetrievableStateHandle<CompletedCheckpoint> retrievableStateHandle1 = mock(RetrievableStateHandle.class);
		when(retrievableStateHandle1.retrieveState()).thenReturn(completedCheckpoint1);
		final RetrievableStateHandle<CompletedCheckpoint> retrievableStateHandle2 = mock(RetrievableStateHandle.class);
		when(retrievableStateHandle2.retrieveState()).thenReturn(completedCheckpoint2);
		checkpointsInZooKeeper.add(Tuple2.of(retrievableStateHandle1, "/foobar1"));
		checkpointsInZooKeeper.add(Tuple2.of(failingRetrievableStateHandle, "/failing1"));
		checkpointsInZooKeeper.add(Tuple2.of(retrievableStateHandle2, "/foobar2"));
		checkpointsInZooKeeper.add(Tuple2.of(failingRetrievableStateHandle, "/failing2"));
		final CuratorFramework client = mock(CuratorFramework.class, Mockito.RETURNS_DEEP_STUBS);
		final RetrievableStateStorageHelper<CompletedCheckpoint> storageHelperMock = mock(RetrievableStateStorageHelper.class);
		ZooKeeperStateHandleStore<CompletedCheckpoint> zooKeeperStateHandleStoreMock = spy(new ZooKeeperStateHandleStore<>(client, storageHelperMock, Executors.directExecutor()));
		whenNew(ZooKeeperStateHandleStore.class).withAnyArguments().thenReturn(zooKeeperStateHandleStoreMock);
		doReturn(checkpointsInZooKeeper).when(zooKeeperStateHandleStoreMock).getAllSortedByNameAndLock();
		final int numCheckpointsToRetain = 1;
		final EnsurePath ensurePathMock = mock(EnsurePath.class);
		final CuratorEvent curatorEventMock = mock(CuratorEvent.class);
		when(curatorEventMock.getType()).thenReturn(CuratorEventType.DELETE);
		when(curatorEventMock.getResultCode()).thenReturn(0);
		when(client.newNamespaceAwareEnsurePath(anyString())).thenReturn(ensurePathMock);
		when(
			client
				.delete()
				.inBackground(any(BackgroundCallback.class), any(Executor.class))
		).thenAnswer(new Answer<ErrorListenerPathable<Void>>() {
			@Override
			public ErrorListenerPathable<Void> answer(InvocationOnMock invocation) throws Throwable {
				final BackgroundCallback callback = (BackgroundCallback) invocation.getArguments()[0];
				ErrorListenerPathable<Void> result = mock(ErrorListenerPathable.class);
				when(result.forPath(anyString())).thenAnswer(new Answer<Void>() {
					@Override
					public Void answer(InvocationOnMock invocation) throws Throwable {
						callback.processResult(client, curatorEventMock);
						return null;
					}
				});
				return result;
			}
		});
		final String checkpointsPath = "foobar";
		final RetrievableStateStorageHelper<CompletedCheckpoint> stateStorage = mock(RetrievableStateStorageHelper.class);
		ZooKeeperCompletedCheckpointStore zooKeeperCompletedCheckpointStore = new ZooKeeperCompletedCheckpointStore(
			numCheckpointsToRetain,
			client,
			checkpointsPath,
			stateStorage,
			Executors.directExecutor());
		zooKeeperCompletedCheckpointStore.recover();
		CompletedCheckpoint latestCompletedCheckpoint = zooKeeperCompletedCheckpointStore.getLatestCheckpoint();
		assertEquals(completedCheckpoint2.getCheckpointID(), latestCompletedCheckpoint.getCheckpointID());
		List<CompletedCheckpoint> completedCheckpoints = zooKeeperCompletedCheckpointStore.getAllCheckpoints();
		Collection<Long> actualCheckpointIds = new HashSet<>(completedCheckpoints.size());
		for (CompletedCheckpoint completedCheckpoint : completedCheckpoints) {
			actualCheckpointIds.add(completedCheckpoint.getCheckpointID());
		}
		assertEquals(expectedCheckpointIds, actualCheckpointIds);
		verify(retrievableStateHandle1, never()).discardState();
		verify(retrievableStateHandle2, never()).discardState();
		verify(failingRetrievableStateHandle, never()).discardState();
	}
@ Test public void testCheckpointRecovery ( ) throws Exception { final List < Tuple2 < RetrievableStateHandle < CompletedCheckpoint > , String > > checkpointsInZooKeeper = new ArrayList < > ( 4 ) ; final CompletedCheckpoint completedCheckpoint1 = new CompletedCheckpoint ( new JobID ( ) , 1L , 1L , 1L , new HashMap < > ( ) , null , CheckpointProperties . forStandardCheckpoint ( ) , null , null ) ; final CompletedCheckpoint completedCheckpoint2 = new CompletedCheckpoint ( new JobID ( ) , 2L , 2L , 2L , new HashMap < > ( ) , null , CheckpointProperties . forStandardCheckpoint ( ) , null , null ) ; final Collection < Long > expectedCheckpointIds = new HashSet < > ( 2 ) ; expectedCheckpointIds . add ( 1L ) ; expectedCheckpointIds . add ( 2L ) ; final RetrievableStateHandle < CompletedCheckpoint > failingRetrievableStateHandle = mock ( RetrievableStateHandle . class ) ; when ( failingRetrievableStateHandle . retrieveState ( ) ) . thenThrow ( new IOException ( "Test exception" ) ) ; final RetrievableStateHandle < CompletedCheckpoint > retrievableStateHandle1 = mock ( RetrievableStateHandle . class ) ; when ( retrievableStateHandle1 . retrieveState ( ) ) . thenReturn ( completedCheckpoint1 ) ; final RetrievableStateHandle < CompletedCheckpoint > retrievableStateHandle2 = mock ( RetrievableStateHandle . class ) ; when ( retrievableStateHandle2 . retrieveState ( ) ) . thenReturn ( completedCheckpoint2 ) ; checkpointsInZooKeeper . add ( Tuple2 . of ( retrievableStateHandle1 , "/foobar1" ) ) ; checkpointsInZooKeeper . add ( Tuple2 . of ( failingRetrievableStateHandle , "/failing1" ) ) ; checkpointsInZooKeeper . add ( Tuple2 . of ( retrievableStateHandle2 , "/foobar2" ) ) ; checkpointsInZooKeeper . add ( Tuple2 . of ( failingRetrievableStateHandle , "/failing2" ) ) ; final CuratorFramework client = mock ( CuratorFramework . class , Mockito . RETURNS_DEEP_STUBS ) ; final RetrievableStateStorageHelper < CompletedCheckpoint > storageHelperMock = mock ( RetrievableStateStorageHelper . class ) ; ZooKeeperStateHandleStore < CompletedCheckpoint > zooKeeperStateHandleStoreMock = spy ( new ZooKeeperStateHandleStore < > ( client , storageHelperMock , Executors . directExecutor ( ) ) ) ; whenNew ( ZooKeeperStateHandleStore . class ) . withAnyArguments ( ) . thenReturn ( zooKeeperStateHandleStoreMock ) ; doReturn ( checkpointsInZooKeeper ) . when ( zooKeeperStateHandleStoreMock ) . getAllSortedByNameAndLock ( ) ; final int numCheckpointsToRetain = 1 ; final EnsurePath ensurePathMock = mock ( EnsurePath . class ) ; final CuratorEvent curatorEventMock = mock ( CuratorEvent . class ) ; when ( curatorEventMock . getType ( ) ) . thenReturn ( CuratorEventType . DELETE ) ; when ( curatorEventMock . getResultCode ( ) ) . thenReturn ( 0 ) ; when ( client . newNamespaceAwareEnsurePath ( anyString ( ) ) ) . thenReturn ( ensurePathMock ) ; when ( client . delete ( ) . inBackground ( any ( BackgroundCallback . class ) , any ( Executor . class ) ) ) . thenAnswer ( new Answer < ErrorListenerPathable < Void > > ( ) { @ Override public ErrorListenerPathable < Void > answer ( InvocationOnMock invocation ) throws Throwable { final BackgroundCallback callback = ( BackgroundCallback ) invocation . getArguments ( ) [ 0 ] ; ErrorListenerPathable < Void > result = mock ( ErrorListenerPathable . class ) ; when ( result . forPath ( anyString ( ) ) ) . thenAnswer ( new Answer < Void > ( ) { @ Override public Void answer ( InvocationOnMock invocation ) throws Throwable { callback . processResult ( client , curatorEventMock ) ; return null ; } } ) ; return result ; } } ) ; final String checkpointsPath = "foobar" ; final RetrievableStateStorageHelper < CompletedCheckpoint > stateStorage = mock ( RetrievableStateStorageHelper . class ) ; ZooKeeperCompletedCheckpointStore zooKeeperCompletedCheckpointStore = new ZooKeeperCompletedCheckpointStore ( numCheckpointsToRetain , client , checkpointsPath , stateStorage , Executors . directExecutor ( ) ) ; zooKeeperCompletedCheckpointStore . recover ( ) ; CompletedCheckpoint latestCompletedCheckpoint = zooKeeperCompletedCheckpointStore . getLatestCheckpoint ( ) ; assertEquals ( completedCheckpoint2 . getCheckpointID ( ) , latestCompletedCheckpoint . getCheckpointID ( ) ) ; List < CompletedCheckpoint > completedCheckpoints = zooKeeperCompletedCheckpointStore . getAllCheckpoints ( ) ; Collection < Long > actualCheckpointIds = new HashSet < > ( completedCheckpoints . size ( ) ) ; for ( CompletedCheckpoint completedCheckpoint : completedCheckpoints ) { actualCheckpointIds . add ( completedCheckpoint . getCheckpointID ( ) ) ; } assertEquals ( expectedCheckpointIds , actualCheckpointIds ) ; verify ( retrievableStateHandle1 , never ( ) ) . discardState ( ) ; verify ( retrievableStateHandle2 , never ( ) ) . discardState ( ) ; verify ( failingRetrievableStateHandle , never ( ) ) . discardState ( ) ; }
Tests that the completed checkpoint store can retrieve all checkpoints stored in ZooKeeper
 and ignores those which cannot be retrieved via their state handles.
Tests that the completed checkpoint store can retrieve all checkpoints stored in ZooKeeper and ignores those which cannot be retrieved via their state handles .
// Mocking for the delete operation on the CuratorFramework client // It assures that the callback is executed synchronously // check that we return the latest retrievable checkpoint // this should remove the latest checkpoint because it is broken // this should remove the second broken checkpoint because we're iterating over all checkpoints // check that we did not discard any of the state handles // Make sure that we also didn't discard any of the broken handles. Only when checkpoints // are subsumed should they be discarded.