[fix][broker] Fix PulsarService.closeAsync where Condition.signalAll was called without holding a lock (apache#25777)

Co-authored-by: zhaizhibo <zhaizhibo@kuaishou.com>

Author: zhaizhibo

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -754,8 +754,13 @@ public CompletableFuture<Void> closeAsync(boolean waitForWebServiceToStop) {
                 } else {
                     log.warn().exception(t).log("Closed with errors");
                 }
-                state = State.Closed;
-                isClosedCondition.signalAll();
+                mutex.lock();
+                try {
+                    state = State.Closed;
+                    isClosedCondition.signalAll();
+                } finally {
+                    mutex.unlock();
+                }
                 return null;
             });
             return closeFuture;

pulsar-broker/src/test/java/org/apache/pulsar/broker/PulsarServiceCloseTest.java

@@ -18,9 +18,12 @@
  */
 package org.apache.pulsar.broker;
 
+import static org.testng.Assert.fail;
 import static org.testng.AssertJUnit.assertFalse;
 import static org.testng.AssertJUnit.assertTrue;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import lombok.CustomLog;
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
@@ -73,4 +76,29 @@ public void closeInTimeTest() throws Exception {
         }
     }
 
+    @Test(timeOut = 60_000)
+    public void testWaitUntilClosedConcurrentWithCloseAsync() throws Exception {
+        // Start closeAsync() - it initiates close and returns a future
+        CompletableFuture<Void> closeFuture = pulsar.closeAsync();
+
+        // Start waitUntilClosed() in a separate thread BEFORE close completes.
+        // This thread will enter mutex.lock() -> await() and block there,
+        // relying on signalAll() to be woken up when close finishes.
+        CompletableFuture<Void> waitFuture = CompletableFuture.runAsync(() -> {
+            try {
+                pulsar.waitUntilClosed();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException(e);
+            }
+        });
+
+        try {
+            closeFuture.get(30, TimeUnit.SECONDS);
+            waitFuture.get(30, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            fail("Should not throw exception");
+        }
+        log.info("waitUntilClosed() returned successfully while closeAsync() was in progress");
+    }
 }