(pulsar) branch master updated: [fix][broker] Execute the pending callbacks in order before ready for incoming requests (#23266)

[xy...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new ca0fb443ca9 [fix][broker] Execute the pending callbacks in order before ready for incoming requests (#23266)
ca0fb443ca9 is described below

commit ca0fb443ca940c2fa15ecaa9f908d81bed4ebbb1
Author: Yunze Xu <xy...@163.com>
AuthorDate: Sun Sep 8 21:30:42 2024 +0800

    [fix][broker] Execute the pending callbacks in order before ready for incoming requests (#23266)
---
 .../org/apache/pulsar/broker/PulsarService.java    | 25 ++++++++++++++++++++--
 1 file changed, 23 insertions(+), 2 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 0b994c640a9..b2e67bf4883 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -298,6 +298,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
     private final DefaultMonotonicSnapshotClock monotonicSnapshotClock;
     private String brokerId;
     private final CompletableFuture<Void> readyForIncomingRequestsFuture = new CompletableFuture<>();
+    private final List<Runnable> pendingTasksBeforeReadyForIncomingRequests = new ArrayList<>();
 
     public enum State {
         Init, Started, Closing, Closed
@@ -1023,7 +1024,13 @@ public class PulsarService implements AutoCloseable, ShutdownService {
             this.metricsGenerator = new MetricsGenerator(this);
 
             // the broker is ready to accept incoming requests by Pulsar binary protocol and http/https
-            readyForIncomingRequestsFuture.complete(null);
+            final List<Runnable> runnables;
+            synchronized (pendingTasksBeforeReadyForIncomingRequests) {
+                runnables = new ArrayList<>(pendingTasksBeforeReadyForIncomingRequests);
+                pendingTasksBeforeReadyForIncomingRequests.clear();
+                readyForIncomingRequestsFuture.complete(null);
+            }
+            runnables.forEach(Runnable::run);
 
             // Initialize the message protocol handlers.
             // start the protocol handlers only after the broker is ready,
@@ -1082,7 +1089,21 @@ public class PulsarService implements AutoCloseable, ShutdownService {
     }
 
     public void runWhenReadyForIncomingRequests(Runnable runnable) {
-        readyForIncomingRequestsFuture.thenRun(runnable);
+        // Here we don't call the thenRun() methods because CompletableFuture maintains a stack for pending callbacks,
+        // not a queue. Once the future is complete, the pending callbacks will be executed in reverse order of
+        // when they were added.
+        final boolean addedToPendingTasks;
+        synchronized (pendingTasksBeforeReadyForIncomingRequests) {
+            if (readyForIncomingRequestsFuture.isDone()) {
+                addedToPendingTasks = false;
+            } else {
+                pendingTasksBeforeReadyForIncomingRequests.add(runnable);
+                addedToPendingTasks = true;
+            }
+        }
+        if (!addedToPendingTasks) {
+            runnable.run();
+        }
     }
 
     public void waitUntilReadyForIncomingRequests() throws ExecutionException, InterruptedException {