You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@flink.apache.org by zentol <gi...@git.apache.org> on 2018/06/28 09:02:56 UTC

[GitHub] flink pull request #6222: [FLINK-8785][rest] Handle JobSubmissionExceptions

GitHub user zentol opened a pull request:

    https://github.com/apache/flink/pull/6222

    [FLINK-8785][rest] Handle JobSubmissionExceptions

    ## What is the purpose of the change
    
    This PR modifies the `JobSubmitHandler` to handle exceptions contained in the future returned by `DispatcherGateway#submitJob`.
    
    An exception handler was added via `CompletableFuture#exceptionally` to return a proper `ErrorResponseBody` signaling that the job submission has failed.
    
    This PR is pretty much the bare-bones solution; in the JIRA I advocated for including error messages from exceptions since there are various reasons why the submission could fail, but I can't find a satisfying solution.
    
    ## Verifying this change
    
    * see new test in `JobSubmitHandlerTest`
    
    ## Does this pull request potentially affect one of the following parts:
    
      - Dependencies (does it add or upgrade a dependency): (no)
      - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
      - The serializers: (no)
      - The runtime per-record code paths (performance sensitive): (no)
      - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
      - The S3 file system connector: (no)
    
    ## Documentation
    
      - Does this pull request introduce a new feature? (no)
      - If yes, how is the feature documented? (not applicable)


You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/zentol/flink 8785_basic

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/flink/pull/6222.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

    This closes #6222
    
----
commit 32fe49270596cdcf2f91f822c3a6504a14ba40eb
Author: zentol <ch...@...>
Date:   2018-06-28T08:57:01Z

    [FLINK-8785][rest] Handle JobSubmissionExceptions

----


---

[GitHub] flink pull request #6222: [FLINK-8785][rest] Handle JobSubmissionExceptions

Posted by tillrohrmann <gi...@git.apache.org>.

Github user tillrohrmann commented on a diff in the pull request:

    https://github.com/apache/flink/pull/6222#discussion_r199171079
  
    --- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandlerTest.java ---
    @@ -87,4 +89,33 @@ public void testSuccessfulJobSubmission() throws Exception {
     		handler.handleRequest(new HandlerRequest<>(request, EmptyMessageParameters.getInstance()), mockGateway)
     			.get();
     	}
    +
    +	@Test
    +	public void testFailedJobSubmission() throws Exception {
    +		final String errorMessage = "test";
    +		DispatcherGateway mockGateway = mock(DispatcherGateway.class);
    +		when(mockGateway.submitJob(any(JobGraph.class), any(Time.class))).thenReturn(FutureUtils.completedExceptionally(new Exception(errorMessage)));
    +		GatewayRetriever<DispatcherGateway> mockGatewayRetriever = mock(GatewayRetriever.class);
    +
    +		JobSubmitHandler handler = new JobSubmitHandler(
    +			CompletableFuture.completedFuture("http://localhost:1234"),
    +			mockGatewayRetriever,
    +			RpcUtils.INF_TIMEOUT,
    +			Collections.emptyMap());
    +
    +		JobGraph job = new JobGraph("testjob");
    +		JobSubmitRequestBody request = new JobSubmitRequestBody(job);
    +
    +		try {
    +			handler.handleRequest(new HandlerRequest<>(request, EmptyMessageParameters.getInstance()), mockGateway)
    +				.get();
    +		} catch (Exception e) {
    +			Throwable t = ExceptionUtils.stripExecutionException(e);
    +			if (t instanceof RestHandlerException){
    +				Assert.assertTrue(t.getMessage().equals("Job submission failed."));
    +			} else {
    +				throw e;
    +			}
    +		}
    --- End diff --
    
    I think we should make sure that `errorMessage` is part of the `RestHandlerException#message`. Otherwise this information won't be sent to the client in the form of the `ErrorResponseBody`.


---

[GitHub] flink pull request #6222: [FLINK-8785][rest] Handle JobSubmissionExceptions

Posted by tillrohrmann <gi...@git.apache.org>.

Github user tillrohrmann commented on a diff in the pull request:

    https://github.com/apache/flink/pull/6222#discussion_r199170668
  
    --- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandlerTest.java ---
    @@ -87,4 +89,33 @@ public void testSuccessfulJobSubmission() throws Exception {
     		handler.handleRequest(new HandlerRequest<>(request, EmptyMessageParameters.getInstance()), mockGateway)
     			.get();
     	}
    +
    +	@Test
    +	public void testFailedJobSubmission() throws Exception {
    +		final String errorMessage = "test";
    +		DispatcherGateway mockGateway = mock(DispatcherGateway.class);
    +		when(mockGateway.submitJob(any(JobGraph.class), any(Time.class))).thenReturn(FutureUtils.completedExceptionally(new Exception(errorMessage)));
    +		GatewayRetriever<DispatcherGateway> mockGatewayRetriever = mock(GatewayRetriever.class);
    --- End diff --
    
    No need to create a mock. `() -> CompletableFuture.completed(mockGateway)` should be good enough.


---