flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
/**
	 * Tests that late acknowledge checkpoint messages are properly cleaned up. Furthermore it tests
	 * that unknown checkpoint messages for the same job a are cleaned up as well. In contrast
	 * checkpointing messages from other jobs should not be touched. A late acknowledge
	 * message is an acknowledge message which arrives after the checkpoint has been declined.
	 *
	 * @throws Exception
	 */
Tests that late acknowledge checkpoint messages are properly cleaned up. Furthermore it tests
that unknown checkpoint messages for the same job a are cleaned up as well. In contrast
checkpointing messages from other jobs should not be touched. A late acknowledge
message is an acknowledge message which arrives after the checkpoint has been declined.

@throws Exception
Tests that late acknowledge checkpoint messages are properly cleaned up . Furthermore it tests that unknown checkpoint messages for the same job a are cleaned up as well . In contrast checkpointing messages from other jobs should not be touched . A late acknowledge message is an acknowledge message which arrives after the checkpoint has been declined . @throws Exception
@Test
	public void testStateCleanupForLateOrUnknownMessages() throws Exception {
		final JobID jobId = new JobID();
		final ExecutionAttemptID triggerAttemptId = new ExecutionAttemptID();
		final ExecutionVertex triggerVertex = mockExecutionVertex(triggerAttemptId);
		final ExecutionAttemptID ackAttemptId1 = new ExecutionAttemptID();
		final ExecutionVertex ackVertex1 = mockExecutionVertex(ackAttemptId1);
		final ExecutionAttemptID ackAttemptId2 = new ExecutionAttemptID();
		final ExecutionVertex ackVertex2 = mockExecutionVertex(ackAttemptId2);
		final long timestamp = 1L;
		CheckpointCoordinator coord = new CheckpointCoordinator(
			jobId,
			20000L,
			20000L,
			0L,
			1,
			ExternalizedCheckpointSettings.none(),
			new ExecutionVertex[] { triggerVertex },
			new ExecutionVertex[] {triggerVertex, ackVertex1, ackVertex2},
			new ExecutionVertex[0],
			new StandaloneCheckpointIDCounter(),
			new StandaloneCompletedCheckpointStore(1),
			null,
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		assertTrue(coord.triggerCheckpoint(timestamp, false));
		assertEquals(1, coord.getNumberOfPendingCheckpoints());
		PendingCheckpoint pendingCheckpoint = coord.getPendingCheckpoints().values().iterator().next();
		long checkpointId = pendingCheckpoint.getCheckpointId();
		OperatorID opIDtrigger = OperatorID.fromJobVertexID(triggerVertex.getJobvertexId());
		TaskStateSnapshot taskOperatorSubtaskStatesTrigger = spy(new TaskStateSnapshot());
		OperatorSubtaskState subtaskStateTrigger = mock(OperatorSubtaskState.class);
		taskOperatorSubtaskStatesTrigger.putSubtaskStateByOperatorID(opIDtrigger, subtaskStateTrigger);
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, triggerAttemptId, checkpointId, new CheckpointMetrics(), taskOperatorSubtaskStatesTrigger));
		verify(subtaskStateTrigger, never()).discardState();
		TaskStateSnapshot unknownSubtaskState = mock(TaskStateSnapshot.class);
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, new ExecutionAttemptID(), checkpointId, new CheckpointMetrics(), unknownSubtaskState));
		verify(unknownSubtaskState, times(1)).discardState();
		TaskStateSnapshot differentJobSubtaskState = mock(TaskStateSnapshot.class);
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(new JobID(), new ExecutionAttemptID(), checkpointId, new CheckpointMetrics(), differentJobSubtaskState));
		verify(differentJobSubtaskState, never()).discardState();
		TaskStateSnapshot triggerSubtaskState = mock(TaskStateSnapshot.class);
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, triggerAttemptId, checkpointId, new CheckpointMetrics(), triggerSubtaskState));
		verify(triggerSubtaskState, never()).discardState();
		reset(subtaskStateTrigger);
		coord.receiveDeclineMessage(new DeclineCheckpoint(jobId, ackAttemptId1, checkpointId));
		assertTrue(pendingCheckpoint.isDiscarded());
		verify(subtaskStateTrigger, times(1)).discardState();
		TaskStateSnapshot ackSubtaskState = mock(TaskStateSnapshot.class);
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, ackAttemptId2, checkpointId, new CheckpointMetrics(), ackSubtaskState));
		verify(ackSubtaskState, times(1)).discardState();
		reset(differentJobSubtaskState);
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(new JobID(), new ExecutionAttemptID(), checkpointId, new CheckpointMetrics(), differentJobSubtaskState));
		verify(differentJobSubtaskState, never()).discardState();
		TaskStateSnapshot unknownSubtaskState2 = mock(TaskStateSnapshot.class);
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, new ExecutionAttemptID(), checkpointId, new CheckpointMetrics(), unknownSubtaskState2));
		verify(unknownSubtaskState2, times(1)).discardState();
	}
@ Test public void testStateCleanupForLateOrUnknownMessages ( ) throws Exception { final JobID jobId = new JobID ( ) ; final ExecutionAttemptID triggerAttemptId = new ExecutionAttemptID ( ) ; final ExecutionVertex triggerVertex = mockExecutionVertex ( triggerAttemptId ) ; final ExecutionAttemptID ackAttemptId1 = new ExecutionAttemptID ( ) ; final ExecutionVertex ackVertex1 = mockExecutionVertex ( ackAttemptId1 ) ; final ExecutionAttemptID ackAttemptId2 = new ExecutionAttemptID ( ) ; final ExecutionVertex ackVertex2 = mockExecutionVertex ( ackAttemptId2 ) ; final long timestamp = 1L ; CheckpointCoordinator coord = new CheckpointCoordinator ( jobId , 20000L , 20000L , 0L , 1 , ExternalizedCheckpointSettings . none ( ) , new ExecutionVertex [ ] { triggerVertex } , new ExecutionVertex [ ] { triggerVertex , ackVertex1 , ackVertex2 } , new ExecutionVertex [ 0 ] , new StandaloneCheckpointIDCounter ( ) , new StandaloneCompletedCheckpointStore ( 1 ) , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; assertTrue ( coord . triggerCheckpoint ( timestamp , false ) ) ; assertEquals ( 1 , coord . getNumberOfPendingCheckpoints ( ) ) ; PendingCheckpoint pendingCheckpoint = coord . getPendingCheckpoints ( ) . values ( ) . iterator ( ) . next ( ) ; long checkpointId = pendingCheckpoint . getCheckpointId ( ) ; OperatorID opIDtrigger = OperatorID . fromJobVertexID ( triggerVertex . getJobvertexId ( ) ) ; TaskStateSnapshot taskOperatorSubtaskStatesTrigger = spy ( new TaskStateSnapshot ( ) ) ; OperatorSubtaskState subtaskStateTrigger = mock ( OperatorSubtaskState . class ) ; taskOperatorSubtaskStatesTrigger . putSubtaskStateByOperatorID ( opIDtrigger , subtaskStateTrigger ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , triggerAttemptId , checkpointId , new CheckpointMetrics ( ) , taskOperatorSubtaskStatesTrigger ) ) ; verify ( subtaskStateTrigger , never ( ) ) . discardState ( ) ; TaskStateSnapshot unknownSubtaskState = mock ( TaskStateSnapshot . class ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , new ExecutionAttemptID ( ) , checkpointId , new CheckpointMetrics ( ) , unknownSubtaskState ) ) ; verify ( unknownSubtaskState , times ( 1 ) ) . discardState ( ) ; TaskStateSnapshot differentJobSubtaskState = mock ( TaskStateSnapshot . class ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( new JobID ( ) , new ExecutionAttemptID ( ) , checkpointId , new CheckpointMetrics ( ) , differentJobSubtaskState ) ) ; verify ( differentJobSubtaskState , never ( ) ) . discardState ( ) ; TaskStateSnapshot triggerSubtaskState = mock ( TaskStateSnapshot . class ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , triggerAttemptId , checkpointId , new CheckpointMetrics ( ) , triggerSubtaskState ) ) ; verify ( triggerSubtaskState , never ( ) ) . discardState ( ) ; reset ( subtaskStateTrigger ) ; coord . receiveDeclineMessage ( new DeclineCheckpoint ( jobId , ackAttemptId1 , checkpointId ) ) ; assertTrue ( pendingCheckpoint . isDiscarded ( ) ) ; verify ( subtaskStateTrigger , times ( 1 ) ) . discardState ( ) ; TaskStateSnapshot ackSubtaskState = mock ( TaskStateSnapshot . class ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , ackAttemptId2 , checkpointId , new CheckpointMetrics ( ) , ackSubtaskState ) ) ; verify ( ackSubtaskState , times ( 1 ) ) . discardState ( ) ; reset ( differentJobSubtaskState ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( new JobID ( ) , new ExecutionAttemptID ( ) , checkpointId , new CheckpointMetrics ( ) , differentJobSubtaskState ) ) ; verify ( differentJobSubtaskState , never ( ) ) . discardState ( ) ; TaskStateSnapshot unknownSubtaskState2 = mock ( TaskStateSnapshot . class ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , new ExecutionAttemptID ( ) , checkpointId , new CheckpointMetrics ( ) , unknownSubtaskState2 ) ) ; verify ( unknownSubtaskState2 , times ( 1 ) ) . discardState ( ) ; }
Tests that late acknowledge checkpoint messages are properly cleaned up.
Tests that late acknowledge checkpoint messages are properly cleaned up .
// acknowledge the first trigger vertex // verify that the subtask state has not been discarded // receive an acknowledge message for an unknown vertex // we should discard acknowledge messages from an unknown vertex belonging to our job // receive an acknowledge message from an unknown job // we should not interfere with different jobs // duplicate acknowledge message for the trigger vertex // duplicate acknowledge messages for a known vertex should not trigger discarding the state // let the checkpoint fail at the first ack vertex // check that we've cleaned up the already acknowledged state // late acknowledge message from the second ack vertex // check that we also cleaned up this state // receive an acknowledge message from an unknown job // we should not interfere with different jobs // receive an acknowledge message for an unknown vertex // we should discard acknowledge messages from an unknown vertex belonging to our job