JobCancellationWithSavepointHandlers { public TriggerHandler getTriggerHandler() { return triggerHandler; } JobCancellationWithSavepointHandlers(
			ExecutionGraphCache currentGraphs,
			Executor executor);  JobCancellationWithSavepointHandlers(
			ExecutionGraphCache currentGraphs,
			Executor executor,
			@Nullable String defaultSavepointDirectory); TriggerHandler getTriggerHandler(); InProgressHandler getInProgressHandler();  }
@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 testSavepointDirectoryConfiguration() 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, "the-default-directory"); JobCancellationWithSavepointHandlers.TriggerHandler handler = handlers.getTriggerHandler(); Map<String, String> params = new HashMap<>(); params.put("jobid", jobId.toString()); JobManagerGateway jobManager = mock(JobManagerGateway.class); when(jobManager.cancelJobWithSavepoint(eq(jobId), anyString(), any(Time.class))).thenReturn(CompletableFuture.completedFuture("foobar")); params.put("targetDirectory", "custom-directory"); handler.handleRequest(params, Collections.<String, String>emptyMap(), jobManager); verify(jobManager).cancelJobWithSavepoint(eq(jobId), eq("custom-directory"), any(Time.class)); params.remove("targetDirectory"); handler.handleRequest(params, Collections.<String, String>emptyMap(), jobManager); verify(jobManager).cancelJobWithSavepoint(eq(jobId), eq("the-default-directory"), any(Time.class)); handlers = new JobCancellationWithSavepointHandlers(holder, executor, null); handler = handlers.getTriggerHandler(); try { handler.handleRequest(params, Collections.<String, String>emptyMap(), jobManager).get(); fail("Did not throw expected test Exception"); } catch (Exception e) { IllegalStateException cause = (IllegalStateException) e.getCause(); assertEquals(true, cause.getMessage().contains(CoreOptions.SAVEPOINT_DIRECTORY.key())); } }