[GitHub] [pulsar] poorbarcode commented on a diff in pull request #19692: [fix] [broker] Topic close failure leaves subscription in a permanent fence state

["poorbarcode (via GitHub)" <gi...@apache.org>]

poorbarcode commented on code in PR #19692:
URL: https://github.com/apache/pulsar/pull/19692#discussion_r1122846325


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,54 @@ public CompletableFuture<Void> close() {
      */
     @Override
     public synchronized CompletableFuture<Void> disconnect() {
-        CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+        if (fenceFuture != null && !fenceFuture.isDone()){
+            return fenceFuture;
+        }
+        if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+            return fenceFuture;
+        }
+        fenceFuture = new CompletableFuture<>();
 
         // block any further consumers on this subscription
         IS_FENCED_UPDATER.set(this, TRUE);
 
         (dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
                 .thenCompose(v -> close()).thenRun(() -> {
                     log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
-                    disconnectFuture.complete(null);
+                    fenceFuture.complete(null);
                 }).exceptionally(exception -> {
                     IS_FENCED_UPDATER.set(this, FALSE);
                     if (dispatcher != null) {
                         dispatcher.reset();
                     }
                     log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
                             exception);
-                    disconnectFuture.completeExceptionally(exception);
+                    fenceFuture.completeExceptionally(exception);
                     return null;
                 });
 
-        return disconnectFuture;
+        return fenceFuture;
+    }
+
+    /**
+     * Resume subscription after topic deletion or close failure.
+     */
+    public synchronized CompletableFuture<Void> resumeAfterFence() {
+        CompletableFuture<Void> result = new CompletableFuture<>();
+        fenceFuture.whenComplete((ignore, ignore2) -> {

Review Comment:
   Already fixed



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java:
##########
@@ -897,27 +898,54 @@ public CompletableFuture<Void> close() {
      */
     @Override
     public synchronized CompletableFuture<Void> disconnect() {
-        CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+        if (fenceFuture != null && !fenceFuture.isDone()){
+            return fenceFuture;
+        }
+        if (fenceFuture != null && fenceFuture.isDone() && !fenceFuture.isCompletedExceptionally()){
+            return fenceFuture;
+        }
+        fenceFuture = new CompletableFuture<>();
 
         // block any further consumers on this subscription
         IS_FENCED_UPDATER.set(this, TRUE);
 
         (dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
                 .thenCompose(v -> close()).thenRun(() -> {
                     log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
-                    disconnectFuture.complete(null);
+                    fenceFuture.complete(null);
                 }).exceptionally(exception -> {
                     IS_FENCED_UPDATER.set(this, FALSE);
                     if (dispatcher != null) {
                         dispatcher.reset();
                     }
                     log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
                             exception);
-                    disconnectFuture.completeExceptionally(exception);
+                    fenceFuture.completeExceptionally(exception);
                     return null;
                 });
 
-        return disconnectFuture;
+        return fenceFuture;
+    }
+
+    /**
+     * Resume subscription after topic deletion or close failure.
+     */
+    public synchronized CompletableFuture<Void> resumeAfterFence() {
+        CompletableFuture<Void> result = new CompletableFuture<>();
+        fenceFuture.whenComplete((ignore, ignore2) -> {
+            try {
+                if (IS_FENCED_UPDATER.compareAndSet(this, TRUE, FALSE)) {
+                    if (dispatcher != null) {
+                        dispatcher.reset();
+                    }
+                }
+                result.complete(null);
+            } catch (Exception ex){
+                result.completeExceptionally(ex);
+                log.error("[{}] Resume subscription[{}] failure: {}", topicName, subName, ex.getMessage(), ex);

Review Comment:
   Already fixed



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org