[fix][broker] Flaky-test: ExtensibleLoadManagerImplTest.testDisableBroker

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

@@ -460,7 +460,32 @@ private CompletableFuture<Optional<String>> getActiveOwnerAsync(
             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.