@Test
public void testSerializationFailureHandling() throws Exception {

	DispatcherGateway mockGateway = mock(DispatcherGateway.class);
	when(mockGateway.submitJob(any(JobGraph.class), any(Time.class))).thenReturn(CompletableFuture.completedFuture(Acknowledge.get()));
	GatewayRetriever<DispatcherGateway> mockGatewayRetriever = mock(GatewayRetriever.class);
	JobSubmitHandler handler = new JobSubmitHandler(
		CompletableFuture.completedFuture("http://localhost:1234"),
		mockGatewayRetriever,
		RpcUtils.INF_TIMEOUT);
	JobSubmitRequestBody request = new JobSubmitRequestBody(new byte[0]);
	try {
		handler.handleRequest(new HandlerRequest<>(request, EmptyMessageParameters.getInstance()), mockGateway);
		Assert.fail();
	} catch (RestHandlerException rhe) {
		Assert.assertEquals(HttpResponseStatus.BAD_REQUEST, rhe.getHttpResponseStatus());
	}
}