zentol closed pull request #6731: [FLINK-10312] Propagate exception from server
to client in REST API
URL: https://github.com/apache/flink/pull/6731
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
index 935a07faf89..9eb8cc72c8f 100644
---
a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
+++
b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -376,7 +376,7 @@ public JobSubmissionResult submitJob(JobGraph jobGraph,
ClassLoader classLoader)
(JobSubmitResponseBody jobSubmitResponseBody)
-> new JobSubmissionResult(jobGraph.getJobID()))
.exceptionally(
(Throwable throwable) -> {
- throw new CompletionException(new
JobSubmissionException(jobGraph.getJobID(), "Failed to submit JobGraph.",
throwable));
+ throw new CompletionException(new
JobSubmissionException(jobGraph.getJobID(), "Failed to submit JobGraph.",
ExceptionUtils.stripCompletionException(throwable)));
});
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/FutureUtils.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/FutureUtils.java
index d4a65dec8f6..5a19a3f628d 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/FutureUtils.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/FutureUtils.java
@@ -198,7 +198,10 @@
if (throwable instanceof
CancellationException) {
resultFuture.completeExceptionally(new RetryException("Operation future was
cancelled.", throwable));
} else {
- if (retries > 0 &&
retryPredicate.test(throwable)) {
+ throwable =
ExceptionUtils.stripExecutionException(throwable);
+ if
(!retryPredicate.test(throwable)) {
+
resultFuture.completeExceptionally(throwable);
+ } else if (retries > 0)
{
final
ScheduledFuture<?> scheduledFuture = scheduledExecutor.schedule(
() ->
retryOperationWithDelay(resultFuture, operation, retries - 1, retryDelay,
retryPredicate, scheduledExecutor),
retryDelay.toMilliseconds(),
@@ -207,12 +210,10 @@
resultFuture.whenComplete(
(innerT, innerThrowable) -> scheduledFuture.cancel(false));
} else {
- final String
errorMsg = retries == 0 ?
- "Number
of retries has been exhausted." :
-
"Exception is not retryable.";
-
resultFuture.completeExceptionally(new RetryException(
- "Could
not complete the operation. " + errorMsg,
-
throwable));
+ RetryException
retryException = new RetryException(
+ "Could
not complete the operation. Number of retries has been exhausted.",
+
throwable);
+
resultFuture.completeExceptionally(retryException);
}
}
} else {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java
index e4cec086017..9cfb58e98a8 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.rest.handler;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.rest.AbstractHandler;
import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
@@ -39,6 +40,7 @@
import javax.annotation.Nonnull;
+import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -80,43 +82,26 @@ protected AbstractRestHandler(
}
return response.whenComplete((P resp, Throwable throwable) -> {
- if (throwable != null) {
-
- Throwable error =
ExceptionUtils.stripCompletionException(throwable);
-
- if (error instanceof RestHandlerException) {
- final RestHandlerException rhe =
(RestHandlerException) error;
-
- processRestHandlerException(ctx,
httpRequest, rhe);
- } else {
- log.error("Implementation error:
Unhandled exception.", error);
- HandlerUtils.sendErrorResponse(
- ctx,
- httpRequest,
- new ErrorResponseBody("Internal
server error."),
-
HttpResponseStatus.INTERNAL_SERVER_ERROR,
- responseHeaders);
- }
- } else {
- HandlerUtils.sendResponse(
- ctx,
- httpRequest,
- resp,
- messageHeaders.getResponseStatusCode(),
- responseHeaders);
- }
+ Tuple2<ResponseBody, HttpResponseStatus> r = throwable
!= null ?
+ errorResponse(throwable) : Tuple2.of(resp,
messageHeaders.getResponseStatusCode());
+ HandlerUtils.sendResponse(ctx, httpRequest, r.f0, r.f1,
responseHeaders);
}).thenApply(ignored -> null);
}
- private void processRestHandlerException(ChannelHandlerContext ctx,
HttpRequest httpRequest, RestHandlerException rhe) {
- log.error("Exception occurred in REST handler.", rhe);
-
- HandlerUtils.sendErrorResponse(
- ctx,
- httpRequest,
- new ErrorResponseBody(rhe.getMessage()),
- rhe.getHttpResponseStatus(),
- responseHeaders);
+ private Tuple2<ResponseBody, HttpResponseStatus>
errorResponse(Throwable throwable) {
+ Throwable error =
ExceptionUtils.stripCompletionException(throwable);
+ if (error instanceof RestHandlerException) {
+ final RestHandlerException rhe = (RestHandlerException)
error;
+ log.error("Exception occurred in REST handler.", rhe);
+ return Tuple2.of(new
ErrorResponseBody(rhe.getMessage()), rhe.getHttpResponseStatus());
+ } else {
+ log.error("Implementation error: Unhandled exception.",
error);
+ String stackTrace = String.format("<Exception on server
side:%n%s%nEnd of exception on server side>",
+ ExceptionUtils.stringifyException(throwable));
+ return Tuple2.of(
+ new ErrorResponseBody(Arrays.asList("Internal
server error.", stackTrace)),
+ HttpResponseStatus.INTERNAL_SERVER_ERROR);
+ }
}
/**
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java
index 491ba094f4c..e3101bef5ee 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java
@@ -114,10 +114,7 @@ public JobSubmitHandler(
CompletableFuture<Acknowledge> jobSubmissionFuture =
finalizedJobGraphFuture.thenCompose(jobGraph -> gateway.submitJob(jobGraph,
timeout));
return jobSubmissionFuture.thenCombine(jobGraphFuture,
- (ack, jobGraph) -> new JobSubmitResponseBody("/jobs/" +
jobGraph.getJobID()))
- .exceptionally(exception -> {
- throw new CompletionException(new
RestHandlerException("Job submission failed.",
HttpResponseStatus.INTERNAL_SERVER_ERROR, exception));
- });
+ (ack, jobGraph) -> new JobSubmitResponseBody("/jobs/" +
jobGraph.getJobID()));
}
private CompletableFuture<JobGraph> loadJobGraph(JobSubmitRequestBody
requestBody, Map<String, Path> nameToFile) throws MissingFileException {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FutureUtilsTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FutureUtilsTest.java
index c386952c056..99962a6848e 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FutureUtilsTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FutureUtilsTest.java
@@ -82,7 +82,7 @@ public void testRetrySuccess() throws Exception {
TestingUtils.defaultExecutor());
assertTrue(retryFuture.get());
- assertTrue(retries == atomicInteger.get());
+ assertEquals(retries, atomicInteger.get());
}
/**
@@ -274,7 +274,7 @@ public void testRetryWithDelayAndPredicate() throws
Exception {
throwable instanceof RuntimeException
&& throwable.getMessage().contains(retryableExceptionMessage),
new
ScheduledExecutorServiceAdapter(retryExecutor)).get();
} catch (final ExecutionException e) {
- assertThat(e.getMessage(), containsString("Could not
complete the operation"));
+ assertThat(e.getMessage(), containsString("should
propagate"));
} finally {
retryExecutor.shutdownNow();
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandlerTest.java
index be1cb797748..78ace967bec 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandlerTest.java
@@ -269,11 +269,7 @@ public void testFailedJobSubmission() throws Exception {
.get();
} catch (Exception e) {
Throwable t = ExceptionUtils.stripExecutionException(e);
- if (t instanceof RestHandlerException){
- Assert.assertTrue(t.getMessage().equals("Job
submission failed."));
- } else {
- throw e;
- }
+ Assert.assertEquals(errorMessage, t.getMessage());
}
}
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
