Fix: Exception being ignored in PulsarAdmin (apache#7510)

Co-authored-by: Jerry Peng <[email protected]>
sarvex · Jul 11, 2020 · 77ea9b8 · 77ea9b8
1 parent 5fd96e2
commit 77ea9b8
Show file tree

Hide file tree

Showing 3 changed files with 59 additions and 18 deletions.
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
@@ -357,6 +357,10 @@ public CompletableFuture<Void> createFunctionAsync(FunctionConfig functionConfig
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
 
         } catch (Exception e) {
@@ -472,6 +476,10 @@ public CompletableFuture<Void> updateFunctionAsync(
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
         } catch (Exception e) {
             future.completeExceptionally(getApiException(e));
@@ -755,6 +763,10 @@ public CompletableFuture<Void> uploadFunctionAsync(String sourceFile, String pat
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
         } catch (Exception e) {
             future.completeExceptionally(getApiException(e));
@@ -844,24 +856,29 @@ public void onThrowable(Throwable t) {
                             }
                         }).toCompletableFuture();
 
-            statusFuture.thenAccept(status -> {
-                try {
-                    os.close();
-                } catch (Exception e) {
-                    future.completeExceptionally(getApiException(e));
-                    return;
-                }
-
-                if (status.getStatusCode() < 200 || status.getStatusCode() >= 300) {
-                    future.completeExceptionally(
-                            getApiException(Response
-                                    .status(status.getStatusCode())
-                                    .entity(status.getStatusText())
-                                    .build()));
-                } else {
-                    future.complete(null);
-                }
-            });
+            statusFuture
+                    .thenAccept(status -> {
+                        try {
+                            os.close();
+                        } catch (Exception e) {
+                            future.completeExceptionally(getApiException(e));
+                            return;
+                        }
+
+                        if (status.getStatusCode() < 200 || status.getStatusCode() >= 300) {
+                            future.completeExceptionally(
+                                    getApiException(Response
+                                            .status(status.getStatusCode())
+                                            .entity(status.getStatusText())
+                                            .build()));
+                        } else {
+                            future.complete(null);
+                        }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
+                    });
         } catch (Exception e) {
             future.completeExceptionally(getApiException(e));
         }
@@ -978,6 +995,10 @@ public CompletableFuture<Void> putFunctionStateAsync(
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
 
         } catch (Exception e) {
@@ -1025,6 +1046,10 @@ public CompletableFuture<Void> updateOnWorkerLeaderAsync(String tenant, String n
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
 
         } catch (Exception e) {

diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java
@@ -259,6 +259,10 @@ public CompletableFuture<Void> createSinkAsync(SinkConfig sinkConfig, String fil
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
         } catch (Exception e) {
             future.completeExceptionally(getApiException(e));
@@ -358,6 +362,10 @@ public CompletableFuture<Void> updateSinkAsync(
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
         } catch (Exception e) {
             future.completeExceptionally(getApiException(e));

diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java
@@ -260,6 +260,10 @@ public CompletableFuture<Void> createSourceAsync(SourceConfig sourceConfig, Stri
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
         } catch (Exception e) {
             future.completeExceptionally(getApiException(e));
@@ -360,6 +364,10 @@ public CompletableFuture<Void> updateSourceAsync(
                         } else {
                             future.complete(null);
                         }
+                    })
+                    .exceptionally(throwable -> {
+                        future.completeExceptionally(getApiException(throwable));
+                        return null;
                     });
         } catch (Exception e) {
             future.completeExceptionally(getApiException(e));