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 response when a request fails.
	 */
Tests response when a request fails.
Tests response when a request fails .
@Test
	public void testFailedCancellation() throws Exception {
		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,
				1L,
				1L,
				1,
				ExternalizedCheckpointSettings.none(),
				true));
		JobCancellationWithSavepointHandlers handlers = new JobCancellationWithSavepointHandlers(holder, executor);
		JobCancellationWithSavepointHandlers.TriggerHandler trigger = handlers.getTriggerHandler();
		JobCancellationWithSavepointHandlers.InProgressHandler progress = handlers.getInProgressHandler();
		Map<String, String> params = new HashMap<>();
		params.put("jobid", jobId.toString());
		params.put("targetDirectory", "custom-directory");
		JobManagerGateway jobManager = mock(JobManagerGateway.class);
		CompletableFuture<String> unsuccessfulCancelWithSavepoint = FutureUtils.completedExceptionally(new Exception("Test Exception"));
		when(jobManager.cancelJobWithSavepoint(eq(jobId), eq("custom-directory"), any(Time.class))).thenReturn(unsuccessfulCancelWithSavepoint);
		trigger.handleRequest(params, Collections.<String, String>emptyMap(), jobManager);
		verify(jobManager).cancelJobWithSavepoint(eq(jobId), eq("custom-directory"), any(Time.class));
		params.put("requestId", "1");
		FullHttpResponse response = progress.handleRequest(params, Collections.<String, String>emptyMap(), jobManager).get();
		assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR, response.getStatus());
		assertEquals("application/json; charset=UTF-8", response.headers().get(HttpHeaders.Names.CONTENT_TYPE));
		assertEquals(Integer.toString(response.content().readableBytes()), response.headers().get(HttpHeaders.Names.CONTENT_LENGTH));
		String json = response.content().toString(Charset.forName("UTF-8"));
		JsonNode root = new ObjectMapper().readTree(json);
		assertEquals("failed", root.get("status").asText());
		assertEquals("1", root.get("request-id").asText());
		assertEquals("Test Exception", root.get("cause").asText());
	}
@ Test public void testFailedCancellation ( ) throws Exception { 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 , 1L , 1L , 1 , ExternalizedCheckpointSettings . none ( ) , true ) ) ; JobCancellationWithSavepointHandlers handlers = new JobCancellationWithSavepointHandlers ( holder , executor ) ; JobCancellationWithSavepointHandlers . TriggerHandler trigger = handlers . getTriggerHandler ( ) ; JobCancellationWithSavepointHandlers . InProgressHandler progress = handlers . getInProgressHandler ( ) ; Map < String , String > params = new HashMap < > ( ) ; params . put ( "jobid" , jobId . toString ( ) ) ; params . put ( "targetDirectory" , "custom-directory" ) ; JobManagerGateway jobManager = mock ( JobManagerGateway . class ) ; CompletableFuture < String > unsuccessfulCancelWithSavepoint = FutureUtils . completedExceptionally ( new Exception ( "Test Exception" ) ) ; when ( jobManager . cancelJobWithSavepoint ( eq ( jobId ) , eq ( "custom-directory" ) , any ( Time . class ) ) ) . thenReturn ( unsuccessfulCancelWithSavepoint ) ; trigger . handleRequest ( params , Collections . < String , String > emptyMap ( ) , jobManager ) ; verify ( jobManager ) . cancelJobWithSavepoint ( eq ( jobId ) , eq ( "custom-directory" ) , any ( Time . class ) ) ; params . put ( "requestId" , "1" ) ; FullHttpResponse response = progress . handleRequest ( params , Collections . < String , String > emptyMap ( ) , jobManager ) . get ( ) ; assertEquals ( HttpResponseStatus . INTERNAL_SERVER_ERROR , response . getStatus ( ) ) ; assertEquals ( "application/json; charset=UTF-8" , response . headers ( ) . get ( HttpHeaders . Names . CONTENT_TYPE ) ) ; assertEquals ( Integer . toString ( response . content ( ) . readableBytes ( ) ) , response . headers ( ) . get ( HttpHeaders . Names . CONTENT_LENGTH ) ) ; String json = response . content ( ) . toString ( Charset . forName ( "UTF-8" ) ) ; JsonNode root = new ObjectMapper ( ) . readTree ( json ) ; assertEquals ( "failed" , root . get ( "status" ) . asText ( ) ) ; assertEquals ( "1" , root . get ( "request-id" ) . asText ( ) ) ; assertEquals ( "Test Exception" , root . get ( "cause" ) . asText ( ) ) ; }
Tests response when a request fails.
Tests response when a request fails .
// Successful // Trigger // Query progress