### Question:
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())); } JobSubmitHandler(
			CompletableFuture<String> localRestAddress,
			GatewayRetriever<? extends DispatcherGateway> leaderRetriever,
			Time timeout,
			Map<String, String> headers);   }### Answer:
@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: mockGatewayRetriever, RpcUtils.INF_TIMEOUT, Collections.emptyMap()); 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()); } }
@Test public void testSuccessfulJobSubmission() 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: mockGatewayRetriever, RpcUtils.INF_TIMEOUT, Collections.emptyMap()); JobGraph job = new JobGraph("testjob"); JobSubmitRequestBody request = new JobSubmitRequestBody(job); handler.handleRequest(new HandlerRequest<>(request, EmptyMessageParameters.getInstance()), mockGateway) .get(); }