CheckpointConfigHandler extends AbstractExecutionGraphRequestHandler { @Override public CompletableFuture<String> handleRequest(AccessExecutionGraph graph, Map<String, String> params) { return CompletableFuture.supplyAsync( () -> { try { return createCheckpointConfigJson(graph); } catch (IOException e) { throw new CompletionException(new FlinkException("Could not create checkpoint config json.", e)); } }, executor); } CheckpointConfigHandler(ExecutionGraphCache executionGraphHolder, Executor executor); @Override String[] getPaths(); @Override CompletableFuture<String> handleRequest(AccessExecutionGraph graph, Map<String, String> params);  }
@Test public void testEnabledExternalizedCheckpointSettings() throws Exception { GraphAndSettings graphAndSettings = createGraphAndSettings(true, false); AccessExecutionGraph graph = graphAndSettings.graph; ExternalizedCheckpointSettings externalizedSettings = graphAndSettings.externalizedSettings; CheckpointConfigHandler handler = new CheckpointConfigHandler(mock(ExecutionGraphCache.class), Executors.directExecutor()); String json = handler.handleRequest(graph, Collections.<String, String>emptyMap()).get(); ObjectMapper mapper = new ObjectMapper(); JsonNode externalizedNode = mapper.readTree(json).get("externalization"); assertNotNull(externalizedNode); assertEquals(externalizedSettings.externalizeCheckpoints(), externalizedNode.get("enabled").asBoolean()); assertEquals(externalizedSettings.deleteOnCancellation(), externalizedNode.get("delete_on_cancellation").asBoolean()); }