class JobSubmitHandler extends AbstractRestHandler<DispatcherGateway, JobSubmitRequestBody, JobSubmitResponseBody, EmptyMessageParameters> {

	@Override
	protected CompletableFuture<JobSubmitResponseBody> handleRequest(@Nonnull HandlerRequest<JobSubmitRequestBody, EmptyMessageParameters> request, @Nonnull DispatcherGateway gateway) throws RestHandlerException {
		JobGraph jobGraph;
		try {
			ObjectInputStream objectIn = new ObjectInputStream(new ByteArrayInputStream(request.getRequestBody().serializedJobGraph));
			jobGraph = (JobGraph) objectIn.readObject();
		} catch (Exception e) {
			throw new RestHandlerException(
				"Failed to deserialize JobGraph.",
				HttpResponseStatus.BAD_REQUEST,
				e);
		}

		return gateway.submitJob(jobGraph, timeout)
			.thenApply(ack -> new JobSubmitResponseBody("/jobs/" + jobGraph.getJobID()));
	}

	public  JobSubmitHandler(
			CompletableFuture<String> localRestAddress,
			GatewayRetriever<? extends DispatcherGateway> leaderRetriever,
			Time timeout,
			Map<String, String> headers);

}