[fix][broker] Return getOwnerAsync without waiting on source broker u…

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

@@ -521,7 +521,16 @@ public CompletableFuture<Optional<String>> getOwnerAsync(String serviceUnit) {
                 return getActiveOwnerAsync(serviceUnit, state, Optional.of(data.sourceBroker()));
             }
             case Assigning, Releasing -> {
-                return getActiveOwnerAsync(serviceUnit, state, Optional.empty());
+                if (isTargetBroker(data.dstBroker())) {
+                    return getActiveOwnerAsync(serviceUnit, state, Optional.of(data.dstBroker()));
+                }
+                // If this broker is not the dst broker, return the dst broker as the owner(or empty).
+                // Clients need to connect(redirect) to the dst broker anyway
+                // and wait for the dst broker to receive `Owned`.
+                // This is also required to return getOwnerAsync on the src broker immediately during unloading.
+                // Otherwise, topic creation(getOwnerAsync) could block unloading bundles,
+                // if the topic creation(getOwnerAsync) happens during unloading on the src broker.
+                return CompletableFuture.completedFuture(Optional.ofNullable(data.dstBroker()));
             }
             case Init, Free -> {
                 return CompletableFuture.completedFuture(Optional.empty());

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java

@@ -167,8 +167,6 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
     private LookupService lookupService;
 
     private static void initConfig(ServiceConfiguration conf){
-        conf.setLoadBalancerInFlightServiceUnitStateWaitingTimeInMillis(5 * 1000);
-        conf.setLoadBalancerServiceUnitStateMonitorIntervalInSeconds(1);
         conf.setForceDeleteNamespaceAllowed(true);
         conf.setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED);
         conf.setAllowAutoTopicCreation(true);

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java

@@ -487,19 +487,17 @@ public void transferTestWhenDestBrokerFails()
         var owner1 = channel1.getOwnerAsync(bundle);
         var owner2 = channel2.getOwnerAsync(bundle);
 
-        assertFalse(owner1.isDone());
+        assertTrue(owner1.isDone());
+        assertEquals(brokerId2, owner1.get().get());
         assertFalse(owner2.isDone());
 
-        assertEquals(1, getOwnerRequests1.size());
+        assertEquals(0, getOwnerRequests1.size());
         assertEquals(1, getOwnerRequests2.size());
 
         // In 10 secs, the getOwnerAsync requests(lookup requests) should time out.
-        Awaitility.await().atMost(10, TimeUnit.SECONDS)
-                .untilAsserted(() -> assertTrue(owner1.isCompletedExceptionally()));
         Awaitility.await().atMost(10, TimeUnit.SECONDS)
                 .untilAsserted(() -> assertTrue(owner2.isCompletedExceptionally()));
 
-        assertEquals(0, getOwnerRequests1.size());
         assertEquals(0, getOwnerRequests2.size());
 
         // recovered, check the monitor update state : Assigned -> Owned
@@ -1136,12 +1134,10 @@ public void assignTestWhenDestBrokerProducerFails()
         var owner1 = channel1.getOwnerAsync(bundle);
         var owner2 = channel2.getOwnerAsync(bundle);
 
-        assertFalse(owner1.isDone());
+        assertTrue(owner1.isDone());
         assertFalse(owner2.isDone());
 
         // In 10 secs, the getOwnerAsync requests(lookup requests) should time out.
-        Awaitility.await().atMost(10, TimeUnit.SECONDS)
-                .untilAsserted(() -> assertTrue(owner1.isCompletedExceptionally()));
         Awaitility.await().atMost(10, TimeUnit.SECONDS)
                 .untilAsserted(() -> assertTrue(owner2.isCompletedExceptionally()));
 
@@ -1320,6 +1316,68 @@ public void testIsOwner() throws IllegalAccessException {
         assertFalse(channel1.isOwner(bundle));
     }
 
+    @Test(priority = 15)
+    public void testGetOwnerAsync() throws Exception {
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Owned, brokerId1, 1));
+        var owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(brokerId1, owner.get().get());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Owned, brokerId2, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(brokerId2, owner.get().get());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Assigning, brokerId1, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(!owner.isDone());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Assigning, brokerId2, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(brokerId2, owner.get().get());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Releasing, brokerId1, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(!owner.isDone());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Releasing, brokerId2, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(brokerId2, owner.get().get());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Releasing, null, brokerId1, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(Optional.empty(), owner.get());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Splitting, null, brokerId1, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(brokerId1, owner.get().get());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Splitting, null, brokerId2, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(brokerId2, owner.get().get());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Free, null, brokerId1, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(Optional.empty(), owner.get());
+
+        overrideTableView(channel1, bundle, new ServiceUnitStateData(Deleted, null, brokerId1, 1));
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertTrue(owner.isCompletedExceptionally());
+
+        overrideTableView(channel1, bundle, null);
+        owner = channel1.getOwnerAsync(bundle);
+        assertTrue(owner.isDone());
+        assertEquals(Optional.empty(), owner.get());
+    }
+
     @Test(priority = 16)
     public void splitAndRetryFailureTest() throws Exception {
         channel1.publishAssignEventAsync(bundle3, brokerId1);

tests/integration/src/test/java/org/apache/pulsar/tests/integration/loadbalance/ExtensibleLoadManagerTest.java

@@ -322,7 +322,7 @@ public void testAntiaffinityPolicy() throws PulsarAdminException {
         assertEquals(result.size(), NUM_BROKERS);
     }
 
-    @Test(timeOut = 240 * 1000)
+    @Test(timeOut = 300 * 1000)
     public void testIsolationPolicy() throws Exception {
         final String namespaceIsolationPolicyName = "my-isolation-policy";
         final String isolationEnabledNameSpace = DEFAULT_TENANT + "/my-isolation-policy" + nsSuffix;
@@ -371,7 +371,7 @@ public void testIsolationPolicy() throws Exception {
             }
         }
 
-        Awaitility.await().atMost(30, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(
+        Awaitility.await().atMost(60, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(
                 () -> {
                     List<String> activeBrokers = admin.brokers().getActiveBrokers();
                     assertEquals(activeBrokers.size(), 2);
@@ -390,7 +390,7 @@ public void testIsolationPolicy() throws Exception {
             }
         }
 
-        Awaitility.await().atMost(30, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(
+        Awaitility.await().atMost(60, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(
             () -> {
                 List<String> activeBrokers = admin.brokers().getActiveBrokers();
                 assertEquals(activeBrokers.size(), 1);