flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobCancellationWithSavepointHandlersTest.java
/**
	 * Tests that the cancellation ask timeout respects the checkpoint timeout.
	 * Otherwise, AskTimeoutExceptions are bound to happen for large state.
	 */
Tests that the cancellation ask timeout respects the checkpoint timeout.
Otherwise, AskTimeoutExceptions are bound to happen for large state.
Tests that the cancellation ask timeout respects the checkpoint timeout . Otherwise AskTimeoutExceptions are bound to happen for large state .
@Test
	public void testAskTimeoutEqualsCheckpointTimeout() throws Exception {
		long timeout = 128288238L;
		JobID jobId = new JobID();
		ExecutionGraphCache holder = mock(ExecutionGraphCache.class);
		ExecutionGraph graph = mock(ExecutionGraph.class);
		when(holder.getExecutionGraph(eq(jobId), any(JobManagerGateway.class))).thenReturn(CompletableFuture.completedFuture(graph));
		when(graph.getCheckpointCoordinatorConfiguration()).thenReturn(
			new CheckpointCoordinatorConfiguration(
				1L,
				timeout,
				1L,
				1,
				ExternalizedCheckpointSettings.none(),
				true));
		JobCancellationWithSavepointHandlers handlers = new JobCancellationWithSavepointHandlers(holder, executor);
		JobCancellationWithSavepointHandlers.TriggerHandler handler = handlers.getTriggerHandler();
		Map<String, String> params = new HashMap<>();
		params.put("jobid", jobId.toString());
		params.put("targetDirectory", "placeholder");
		JobManagerGateway jobManager = mock(JobManagerGateway.class);
		when(jobManager.cancelJobWithSavepoint(eq(jobId), anyString(), any(Time.class))).thenReturn(CompletableFuture.completedFuture("foobar"));
		handler.handleRequest(params, Collections.emptyMap(), jobManager);
		verify(jobManager).cancelJobWithSavepoint(eq(jobId), anyString(), any(Time.class));
	}
@ Test public void testAskTimeoutEqualsCheckpointTimeout ( ) throws Exception { long timeout = 128288238L ; JobID jobId = new JobID ( ) ; ExecutionGraphCache holder = mock ( ExecutionGraphCache . class ) ; ExecutionGraph graph = mock ( ExecutionGraph . class ) ; when ( holder . getExecutionGraph ( eq ( jobId ) , any ( JobManagerGateway . class ) ) ) . thenReturn ( CompletableFuture . completedFuture ( graph ) ) ; when ( graph . getCheckpointCoordinatorConfiguration ( ) ) . thenReturn ( new CheckpointCoordinatorConfiguration ( 1L , timeout , 1L , 1 , ExternalizedCheckpointSettings . none ( ) , true ) ) ; JobCancellationWithSavepointHandlers handlers = new JobCancellationWithSavepointHandlers ( holder , executor ) ; JobCancellationWithSavepointHandlers . TriggerHandler handler = handlers . getTriggerHandler ( ) ; Map < String , String > params = new HashMap < > ( ) ; params . put ( "jobid" , jobId . toString ( ) ) ; params . put ( "targetDirectory" , "placeholder" ) ; JobManagerGateway jobManager = mock ( JobManagerGateway . class ) ; when ( jobManager . cancelJobWithSavepoint ( eq ( jobId ) , anyString ( ) , any ( Time . class ) ) ) . thenReturn ( CompletableFuture . completedFuture ( "foobar" ) ) ; handler . handleRequest ( params , Collections . emptyMap ( ) , jobManager ) ; verify ( jobManager ) . cancelJobWithSavepoint ( eq ( jobId ) , anyString ( ) , any ( Time . class ) ) ; }
Tests that the cancellation ask timeout respects the checkpoint timeout.
Tests that the cancellation ask timeout respects the checkpoint timeout .