(pulsar) branch master updated: [fix][broker] Flaky-test: ExtensibleLoadManagerImplTest.testDisableBroker (#24770)

[xyz]

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 e44e084faf6 [fix][broker] Flaky-test: 
ExtensibleLoadManagerImplTest.testDisableBroker (#24770)
e44e084faf6 is described below

commit e44e084faf66738a22d61892a86d9fc9943bc484
Author: sinan liu <liusinan1...@gmail.com>
AuthorDate: Sun Sep 28 23:12:08 2025 +0800

    [fix][broker] Flaky-test: ExtensibleLoadManagerImplTest.testDisableBroker 
(#24770)
---
 .../channel/ServiceUnitStateChannelImpl.java       | 27 +++++++++++++++++++++-
 1 file changed, 26 insertions(+), 1 deletion(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index ecd98f188b4..50e325cecfa 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -460,7 +460,32 @@ public class ServiceUnitStateChannelImpl implements 
ServiceUnitStateChannel {
             String serviceUnit,
             ServiceUnitState state,
             Optional<String> owner) {
-
+        // When the channel is disabled/closed, do not perform liveness 
verification, return according to the status:
+        if (channelState == Disabled || channelState == Closed) {
+            switch (state) {
+                // Owned/Splitting: Directly return owner (for isOwner 
judgment as true)
+                case Owned:
+                case Splitting:
+                    return CompletableFuture.completedFuture(owner);
+                case Assigning:
+                case Releasing:
+                    if (owner.isPresent()) {
+                        if (isTargetBroker(owner.get())) {
+                            // This machine is the target taker,
+                            // return an unfinished future with "waiting for 
ownership"
+                            return 
dedupeGetOwnerRequest(serviceUnit).thenApply(Optional::ofNullable);
+                        } else {
+                            // The target is another broker, return directly 
so that the upper layer can redirect
+                            return CompletableFuture.completedFuture(owner);
+                        }
+                    } else {
+                        return 
CompletableFuture.completedFuture(Optional.empty());
+                    }
+                // Other status: return empty
+                default:
+                    return CompletableFuture.completedFuture(Optional.empty());
+            }
+        }
         // If this broker's registry does not exist(possibly suffering from 
connecting to the metadata store),
         // we return the owner without its activeness check.
         // This broker tries to serve lookups on a best efforts basis when 
metadata store connection is unstable.