[FLINK-???][runtime] Does proper error handling

XComp · Mar 25, 2024 · fbfa5b8 · fbfa5b8
1 parent 2db44f0
commit fbfa5b8
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 8 deletions.
diff --git a/...me/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java b/...me/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java
@@ -436,8 +436,17 @@ private CompletableFuture<Void> stopJobMasterServiceProcess() {
 
         return jobMasterServiceProcess
                 .closeAsync()
-                .exceptionally(
-                        t -> handleAsyncOperationError(t, "Could not suspend the JobMaster."));
+                .handle(
+                        (ignored, t) -> {
+                            final JobNotFinishedException error =
+                                    new JobNotFinishedException(getJobID());
+                            if (t != null) {
+                                error.addSuppressed(ExceptionUtils.stripCompletionException(t));
+                            }
+
+                            return handleAsyncOperationError(
+                                    error, "Could not stop the JobMaster.");
+                        });
     }
 
     @Override

diff --git a/...rc/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java b/...rc/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java
@@ -214,7 +214,8 @@ void testExceptionallyCompletedResultFutureFromJobMasterServiceProcessIsForwarde
 
         assertThatFuture(jobManagerRunner.getResultFuture())
                 .eventuallyFailsWith(ExecutionException.class)
-                .withCause(cause);
+                .havingRootCause()
+                .isEqualTo(cause);
     }
 
     @Test
@@ -608,6 +609,9 @@ void testJobStatusCancellingIsClearedOnLeadershipLoss() throws Exception {
     @Test
     void testJobMasterServiceProcessClosingExceptionIsForwardedToResultFuture() throws Exception {
         final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+        final FlinkException testException = new FlinkException("Test exception");
+        terminationFuture.completeExceptionally(testException);
+
         final JobMasterServiceLeadershipRunner jobManagerRunner =
                 newJobMasterServiceLeadershipRunnerBuilder()
                         .withSingleJobMasterServiceProcess(
@@ -621,12 +625,13 @@ void testJobMasterServiceProcessClosingExceptionIsForwardedToResultFuture() thro
         leaderElection.isLeader(UUID.randomUUID());
         leaderElection.notLeader();
 
-        final FlinkException testException = new FlinkException("Test exception");
-        terminationFuture.completeExceptionally(testException);
-
         assertThatFuture(jobManagerRunner.getResultFuture())
                 .eventuallyFailsWith(ExecutionException.class)
-                .satisfies(cause -> assertThat(cause).hasRootCause(testException));
+                .havingRootCause()
+                .isInstanceOf(JobNotFinishedException.class)
+                .satisfies(
+                        rootCause ->
+                                assertThat(rootCause.getSuppressed()).containsOnly(testException));
     }
 
     @Test
@@ -649,7 +654,9 @@ void testJobMasterServiceProcessCreationFailureIsForwardedToResultFuture() throw
 
         assertThatFuture(jobManagerRunner.getResultFuture())
                 .eventuallyFailsWith(ExecutionException.class)
-                .satisfies(cause -> assertThat(cause).hasRootCause(testException));
+                .havingRootCause()
+                .isEqualTo(testException)
+                .withNoCause();
     }
 
     @Test