flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java
/**
	 * Tests that a failure while storing a completed checkpoint in the completed checkpoint store
	 * will properly fail the originating pending checkpoint and clean upt the completed checkpoint.
	 */
Tests that a failure while storing a completed checkpoint in the completed checkpoint store
will properly fail the originating pending checkpoint and clean upt the completed checkpoint.
Tests that a failure while storing a completed checkpoint in the completed checkpoint store will properly fail the originating pending checkpoint and clean upt the completed checkpoint .
@Test
	public void testFailingCompletedCheckpointStoreAdd() throws Exception {
		JobID jid = new JobID();
		final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID();
		final ExecutionVertex vertex = CheckpointCoordinatorTest.mockExecutionVertex(executionAttemptId);
		final long triggerTimestamp = 1L;
		CheckpointCoordinator coord = new CheckpointCoordinator(
			jid,
			600000,
			600000,
			0,
			Integer.MAX_VALUE,
			ExternalizedCheckpointSettings.none(),
			new ExecutionVertex[]{vertex},
			new ExecutionVertex[]{vertex},
			new ExecutionVertex[]{vertex},
			new StandaloneCheckpointIDCounter(),
			new FailingCompletedCheckpointStore(),
			null,
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		coord.triggerCheckpoint(triggerTimestamp, false);
		assertEquals(1, coord.getNumberOfPendingCheckpoints());
		PendingCheckpoint pendingCheckpoint = coord.getPendingCheckpoints().values().iterator().next();
		assertFalse(pendingCheckpoint.isDiscarded());
		final long checkpointId = coord.getPendingCheckpoints().keySet().iterator().next();
		KeyedStateHandle managedKeyedHandle = mock(KeyedStateHandle.class);
		KeyedStateHandle rawKeyedHandle = mock(KeyedStateHandle.class);
		OperatorStateHandle managedOpHandle = mock(OperatorStateHandle.class);
		OperatorStateHandle rawOpHandle = mock(OperatorStateHandle.class);
		final OperatorSubtaskState operatorSubtaskState = spy(new OperatorSubtaskState(
			managedOpHandle,
			rawOpHandle,
			managedKeyedHandle,
			rawKeyedHandle));
		TaskStateSnapshot subtaskState = spy(new TaskStateSnapshot());
		subtaskState.putSubtaskStateByOperatorID(new OperatorID(), operatorSubtaskState);
		when(subtaskState.getSubtaskStateByOperatorID(OperatorID.fromJobVertexID(vertex.getJobvertexId()))).thenReturn(operatorSubtaskState);
		AcknowledgeCheckpoint acknowledgeMessage = new AcknowledgeCheckpoint(jid, executionAttemptId, checkpointId, new CheckpointMetrics(), subtaskState);
		try {
			coord.receiveAcknowledgeMessage(acknowledgeMessage);
			fail("Expected a checkpoint exception because the completed checkpoint store could not " +
				"store the completed checkpoint.");
		} catch (CheckpointException e) {
		}
		assertTrue(pendingCheckpoint.isDiscarded());
		verify(operatorSubtaskState).discardState();
		verify(operatorSubtaskState.getManagedOperatorState().iterator().next()).discardState();
		verify(operatorSubtaskState.getRawOperatorState().iterator().next()).discardState();
		verify(operatorSubtaskState.getManagedKeyedState().iterator().next()).discardState();
		verify(operatorSubtaskState.getRawKeyedState().iterator().next()).discardState();
	}
@ Test public void testFailingCompletedCheckpointStoreAdd ( ) throws Exception { JobID jid = new JobID ( ) ; final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID ( ) ; final ExecutionVertex vertex = CheckpointCoordinatorTest . mockExecutionVertex ( executionAttemptId ) ; final long triggerTimestamp = 1L ; CheckpointCoordinator coord = new CheckpointCoordinator ( jid , 600000 , 600000 , 0 , Integer . MAX_VALUE , ExternalizedCheckpointSettings . none ( ) , new ExecutionVertex [ ] { vertex } , new ExecutionVertex [ ] { vertex } , new ExecutionVertex [ ] { vertex } , new StandaloneCheckpointIDCounter ( ) , new FailingCompletedCheckpointStore ( ) , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; coord . triggerCheckpoint ( triggerTimestamp , false ) ; assertEquals ( 1 , coord . getNumberOfPendingCheckpoints ( ) ) ; PendingCheckpoint pendingCheckpoint = coord . getPendingCheckpoints ( ) . values ( ) . iterator ( ) . next ( ) ; assertFalse ( pendingCheckpoint . isDiscarded ( ) ) ; final long checkpointId = coord . getPendingCheckpoints ( ) . keySet ( ) . iterator ( ) . next ( ) ; KeyedStateHandle managedKeyedHandle = mock ( KeyedStateHandle . class ) ; KeyedStateHandle rawKeyedHandle = mock ( KeyedStateHandle . class ) ; OperatorStateHandle managedOpHandle = mock ( OperatorStateHandle . class ) ; OperatorStateHandle rawOpHandle = mock ( OperatorStateHandle . class ) ; final OperatorSubtaskState operatorSubtaskState = spy ( new OperatorSubtaskState ( managedOpHandle , rawOpHandle , managedKeyedHandle , rawKeyedHandle ) ) ; TaskStateSnapshot subtaskState = spy ( new TaskStateSnapshot ( ) ) ; subtaskState . putSubtaskStateByOperatorID ( new OperatorID ( ) , operatorSubtaskState ) ; when ( subtaskState . getSubtaskStateByOperatorID ( OperatorID . fromJobVertexID ( vertex . getJobvertexId ( ) ) ) ) . thenReturn ( operatorSubtaskState ) ; AcknowledgeCheckpoint acknowledgeMessage = new AcknowledgeCheckpoint ( jid , executionAttemptId , checkpointId , new CheckpointMetrics ( ) , subtaskState ) ; try { coord . receiveAcknowledgeMessage ( acknowledgeMessage ) ; fail ( "Expected a checkpoint exception because the completed checkpoint store could not " + "store the completed checkpoint." ) ; } catch ( CheckpointException e ) { } assertTrue ( pendingCheckpoint . isDiscarded ( ) ) ; verify ( operatorSubtaskState ) . discardState ( ) ; verify ( operatorSubtaskState . getManagedOperatorState ( ) . iterator ( ) . next ( ) ) . discardState ( ) ; verify ( operatorSubtaskState . getRawOperatorState ( ) . iterator ( ) . next ( ) ) . discardState ( ) ; verify ( operatorSubtaskState . getManagedKeyedState ( ) . iterator ( ) . next ( ) ) . discardState ( ) ; verify ( operatorSubtaskState . getRawKeyedState ( ) . iterator ( ) . next ( ) ) . discardState ( ) ; }
Tests that a failure while storing a completed checkpoint in the completed checkpoint store
 will properly fail the originating pending checkpoint and clean upt the completed checkpoint.
Tests that a failure while storing a completed checkpoint in the completed checkpoint store will properly fail the originating pending checkpoint and clean upt the completed checkpoint .
// set up the coordinator and validate the initial state // ignore because we expected this exception // make sure that the pending checkpoint has been discarded after we could not complete it // make sure that the subtask state has been discarded after we could not complete it.