@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));
	}
}