RestClusterClient extends ClusterClient { @Override public CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory) throws Exception { SavepointTriggerHeaders headers = SavepointTriggerHeaders.getInstance(); SavepointMessageParameters params = headers.getUnresolvedMessageParameters(); params.jobID.resolve(jobId); if (savepointDirectory != null) { params.targetDirectory.resolve(Collections.singletonList(savepointDirectory)); } CompletableFuture<SavepointTriggerResponseBody> responseFuture = restClient.sendRequest( restClusterClientConfiguration.getRestServerAddress(), restClusterClientConfiguration.getRestServerPort(), headers, params ); return responseFuture .thenApply(response -> response.location); } RestClusterClient(Configuration config);  RestClusterClient(Configuration config, RestClusterClientConfiguration configuration); @Override void shutdown(); @Override void stop(JobID jobID); @Override void cancel(JobID jobID); @Override String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<Collection<JobStatusMessage>> listJobs(); @Override String getClusterIdentifier(); @Override boolean hasUserJarsInClassPath(List<URL> userJarFiles); @Override void waitForClusterToBeReady(); @Override String getWebInterfaceURL(); @Override GetClusterStatusResponse getClusterStatus(); @Override int getMaxSlots();  }
@Test public void testTriggerSavepoint() throws Exception { Configuration config = new Configuration(); config.setString(JobManagerOptions.ADDRESS, "localhost"); RestServerEndpointConfiguration rsec = RestServerEndpointConfiguration.fromConfiguration(config); String targetSavepointDirectory = "/alternate"; TestSavepointTriggerHandler triggerHandler = new TestSavepointTriggerHandler(targetSavepointDirectory); RestServerEndpoint rse = new RestServerEndpoint(rsec) { @Override protected Collection<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(CompletableFuture<String> restAddressFuture) { Collection<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = new ArrayList<>(); handlers.add(Tuple2.of(triggerHandler.getMessageHeaders(), triggerHandler)); return handlers; } }; RestClusterClient rcc = new RestClusterClient(config); try { rse.start(); JobID id = new JobID(); { CompletableFuture<String> savepointPathFuture = rcc.triggerSavepoint(id, null); String savepointPath = savepointPathFuture.get(); Assert.assertEquals("/universe", savepointPath); } { CompletableFuture<String> savepointPathFuture = rcc.triggerSavepoint(id, targetSavepointDirectory); String savepointPath = savepointPathFuture.get(); Assert.assertEquals(targetSavepointDirectory + "/universe", savepointPath); } } finally { rcc.shutdown(); rse.shutdown(Time.seconds(5)); } }