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 checkpoint does not exist in the store when we fail to add
	 * it into the store (i.e., there exists an exception thrown by the method).
	 */
Tests that the checkpoint does not exist in the store when we fail to add
it into the store .
Tests that the checkpoint does not exist in the store when we fail to add it into the store .
@Test
	public void testAddCheckpointWithFailedRemove() throws Exception {
		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);
		doAnswer(new Answer<RetrievableStateHandle<CompletedCheckpoint>>() {
			@Override
			public RetrievableStateHandle<CompletedCheckpoint> answer(InvocationOnMock invocationOnMock) throws Throwable {
				CompletedCheckpoint checkpoint = (CompletedCheckpoint)invocationOnMock.getArguments()[1];
				RetrievableStateHandle<CompletedCheckpoint> retrievableStateHandle = mock(RetrievableStateHandle.class);
				when(retrievableStateHandle.retrieveState()).thenReturn(checkpoint);
				return retrievableStateHandle;
			}
		}).when(zookeeperStateHandleStoreMock).addAndLock(anyString(), any(CompletedCheckpoint.class));
		doThrow(new Exception()).when(zookeeperStateHandleStoreMock).releaseAndTryRemove(anyString(), any(ZooKeeperStateHandleStore.RemoveCallback.class));
		final int numCheckpointsToRetain = 1;
		final String checkpointsPath = "foobar";
		final RetrievableStateStorageHelper<CompletedCheckpoint> stateSotrage = mock(RetrievableStateStorageHelper.class);
		ZooKeeperCompletedCheckpointStore zooKeeperCompletedCheckpointStore = new ZooKeeperCompletedCheckpointStore(
			numCheckpointsToRetain,
			client,
			checkpointsPath,
			stateSotrage,
			Executors.directExecutor());
		for (long i = 0; i <= numCheckpointsToRetain; ++i) {
			CompletedCheckpoint checkpointToAdd = mock(CompletedCheckpoint.class);
			doReturn(i).when(checkpointToAdd).getCheckpointID();
			doReturn(Collections.emptyMap()).when(checkpointToAdd).getOperatorStates();
			try {
				zooKeeperCompletedCheckpointStore.addCheckpoint(checkpointToAdd);
				List<CompletedCheckpoint> addedCheckpoints = zooKeeperCompletedCheckpointStore.getAllCheckpoints();
				assertTrue(addedCheckpoints.contains(checkpointToAdd));
			} catch (Exception e) {
				List<CompletedCheckpoint> addedCheckpoints = zooKeeperCompletedCheckpointStore.getAllCheckpoints();
				assertFalse(addedCheckpoints.contains(checkpointToAdd));
			}
		}
	}
@ Test public void testAddCheckpointWithFailedRemove ( ) throws Exception { 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 ) ; doAnswer ( new Answer < RetrievableStateHandle < CompletedCheckpoint > > ( ) { @ Override public RetrievableStateHandle < CompletedCheckpoint > answer ( InvocationOnMock invocationOnMock ) throws Throwable { CompletedCheckpoint checkpoint = ( CompletedCheckpoint ) invocationOnMock . getArguments ( ) [ 1 ] ; RetrievableStateHandle < CompletedCheckpoint > retrievableStateHandle = mock ( RetrievableStateHandle . class ) ; when ( retrievableStateHandle . retrieveState ( ) ) . thenReturn ( checkpoint ) ; return retrievableStateHandle ; } } ) . when ( zookeeperStateHandleStoreMock ) . addAndLock ( anyString ( ) , any ( CompletedCheckpoint . class ) ) ; doThrow ( new Exception ( ) ) . when ( zookeeperStateHandleStoreMock ) . releaseAndTryRemove ( anyString ( ) , any ( ZooKeeperStateHandleStore . RemoveCallback . class ) ) ; final int numCheckpointsToRetain = 1 ; final String checkpointsPath = "foobar" ; final RetrievableStateStorageHelper < CompletedCheckpoint > stateSotrage = mock ( RetrievableStateStorageHelper . class ) ; ZooKeeperCompletedCheckpointStore zooKeeperCompletedCheckpointStore = new ZooKeeperCompletedCheckpointStore ( numCheckpointsToRetain , client , checkpointsPath , stateSotrage , Executors . directExecutor ( ) ) ; for ( long i = 0 ; i <= numCheckpointsToRetain ; ++ i ) { CompletedCheckpoint checkpointToAdd = mock ( CompletedCheckpoint . class ) ; doReturn ( i ) . when ( checkpointToAdd ) . getCheckpointID ( ) ; doReturn ( Collections . emptyMap ( ) ) . when ( checkpointToAdd ) . getOperatorStates ( ) ; try { zooKeeperCompletedCheckpointStore . addCheckpoint ( checkpointToAdd ) ; List < CompletedCheckpoint > addedCheckpoints = zooKeeperCompletedCheckpointStore . getAllCheckpoints ( ) ; assertTrue ( addedCheckpoints . contains ( checkpointToAdd ) ) ; } catch ( Exception e ) { List < CompletedCheckpoint > addedCheckpoints = zooKeeperCompletedCheckpointStore . getAllCheckpoints ( ) ; assertFalse ( addedCheckpoints . contains ( checkpointToAdd ) ) ; } } }
Tests that the checkpoint does not exist in the store when we fail to add
 it into the store (i.e., there exists an exception thrown by the method).
Tests that the checkpoint does not exist in the store when we fail to add it into the store ( i . e . there exists an exception thrown by the method ) .
// The checkpoint should be in the store if we successfully add it into the store. // The checkpoint should not be in the store if any exception is thrown.