[FLINK-25893][runtime] Do not report error if ResourceManagerServiceI…

…mpl#deregisterApplication is called when there's no leading RM.
BeSky · Feb 21, 2022 · 9f1d43b · 9f1d43b
1 parent d0c8547
commit 9f1d43b
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 3 deletions.
diff --git a/...me/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImpl.java b/...me/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImpl.java
@@ -155,9 +155,8 @@ public CompletableFuture<Void> deregisterApplication(
     }
 
     private static CompletableFuture<Void> deregisterWithoutLeaderRm() {
-        return FutureUtils.completedExceptionally(
-                new FlinkException(
-                        "Cannot deregister application. Resource manager service is not available."));
+        LOG.warn("Cannot deregister application. Resource manager service is not available.");
+        return FutureUtils.completedVoidFuture();
     }
 
     @Override

diff --git a/...rc/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImplTest.java b/...rc/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImplTest.java
@@ -482,6 +482,16 @@ public void deregisterApplication_leaderRmNotStarted() throws Exception {
         deregisterApplicationFuture.get(TIMEOUT.getSize(), TIMEOUT.getUnit());
     }
 
+    @Test
+    public void deregisterApplication_noLeaderRm() throws Exception {
+        createAndStartResourceManager();
+        final CompletableFuture<Void> deregisterApplicationFuture =
+                resourceManagerService.deregisterApplication(ApplicationStatus.CANCELED, null);
+
+        // should not report error
+        deregisterApplicationFuture.get(TIMEOUT.getSize(), TIMEOUT.getUnit());
+    }
+
     private static void blockOnFuture(CompletableFuture<?> future) {
         try {
             future.get();