flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointTest.java
/**
	 * Tests that abort discards state.
	 */
Tests that abort discards state.
Tests that abort discards state .
@Test
	@SuppressWarnings("unchecked")
	public void testAbortDiscardsState() throws Exception {
		CheckpointProperties props = new CheckpointProperties(false, true, false, false, false, false, false, false);
		QueueExecutor executor = new QueueExecutor();
		OperatorState state = mock(OperatorState.class);
		doNothing().when(state).registerSharedStates(any(SharedStateRegistry.class));
		String targetDir = tmpFolder.newFolder().getAbsolutePath();
		PendingCheckpoint pending = createPendingCheckpoint(props, targetDir, executor);
		setTaskState(pending, state);
		pending.abortDeclined();
		executor.runQueuedCommands();
		verify(state, times(1)).discardState();
		Mockito.reset(state);
		pending = createPendingCheckpoint(props, targetDir, executor);
		setTaskState(pending, state);
		pending.abortError(new Exception("Expected Test Exception"));
		executor.runQueuedCommands();
		verify(state, times(1)).discardState();
		Mockito.reset(state);
		pending = createPendingCheckpoint(props, targetDir, executor);
		setTaskState(pending, state);
		pending.abortExpired();
		executor.runQueuedCommands();
		verify(state, times(1)).discardState();
		Mockito.reset(state);
		pending = createPendingCheckpoint(props, targetDir, executor);
		setTaskState(pending, state);
		pending.abortSubsumed();
		executor.runQueuedCommands();
		verify(state, times(1)).discardState();
	}
@ Test @ SuppressWarnings ( "unchecked" ) public void testAbortDiscardsState ( ) throws Exception { CheckpointProperties props = new CheckpointProperties ( false , true , false , false , false , false , false , false ) ; QueueExecutor executor = new QueueExecutor ( ) ; OperatorState state = mock ( OperatorState . class ) ; doNothing ( ) . when ( state ) . registerSharedStates ( any ( SharedStateRegistry . class ) ) ; String targetDir = tmpFolder . newFolder ( ) . getAbsolutePath ( ) ; PendingCheckpoint pending = createPendingCheckpoint ( props , targetDir , executor ) ; setTaskState ( pending , state ) ; pending . abortDeclined ( ) ; executor . runQueuedCommands ( ) ; verify ( state , times ( 1 ) ) . discardState ( ) ; Mockito . reset ( state ) ; pending = createPendingCheckpoint ( props , targetDir , executor ) ; setTaskState ( pending , state ) ; pending . abortError ( new Exception ( "Expected Test Exception" ) ) ; executor . runQueuedCommands ( ) ; verify ( state , times ( 1 ) ) . discardState ( ) ; Mockito . reset ( state ) ; pending = createPendingCheckpoint ( props , targetDir , executor ) ; setTaskState ( pending , state ) ; pending . abortExpired ( ) ; executor . runQueuedCommands ( ) ; verify ( state , times ( 1 ) ) . discardState ( ) ; Mockito . reset ( state ) ; pending = createPendingCheckpoint ( props , targetDir , executor ) ; setTaskState ( pending , state ) ; pending . abortSubsumed ( ) ; executor . runQueuedCommands ( ) ; verify ( state , times ( 1 ) ) . discardState ( ) ; }
Tests that abort discards state.
Tests that abort discards state .
// Abort declined // execute asynchronous discard operation // Abort error // execute asynchronous discard operation // Abort expired // execute asynchronous discard operation // Abort subsumed // execute asynchronous discard operation