fix import into AdminApiTest and introduce disable-namespaceBundle unit test

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/BrokerService.java

@@ -439,7 +439,7 @@ public PulsarClient getReplicationClient(String cluster) {
      * @return CompletableFuture<Topic>
      * @throws RuntimeException
      */
-    private CompletableFuture<Topic> createPersistentTopic(final String topic) throws RuntimeException {
+    protected CompletableFuture<Topic> createPersistentTopic(final String topic) throws RuntimeException {
         checkTopicNsOwnership(topic);
 
         final CompletableFuture<Topic> topicFuture = new CompletableFuture<>();

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/AdminApiTest.java

@@ -57,6 +57,7 @@
 import com.yahoo.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import com.yahoo.pulsar.broker.namespace.NamespaceEphemeralData;
 import com.yahoo.pulsar.broker.namespace.NamespaceService;
+import com.yahoo.pulsar.broker.service.BrokerService;
 import com.yahoo.pulsar.client.admin.PulsarAdmin;
 import com.yahoo.pulsar.client.admin.PulsarAdminException;
 import com.yahoo.pulsar.client.admin.PulsarAdminException.ConflictException;
@@ -1697,34 +1698,4 @@ public void failed(Throwable e) {
         assertEquals(uriStats.get().subscriptions.size(), 1);
     }
 
-    /**
-     * Verifies that deleteNamespace cleans up policies(global,local), bundle cache and bundle ownership
-     * 
-     * @throws Exception
-     */
-    @Test
-    public void testDeleteNamespace() throws Exception {
-
-        final String namespace = "prop-xyz/use/deleteNs";
-        admin.namespaces().createNamespace(namespace, 100);
-        assertEquals(admin.namespaces().getPolicies(namespace).bundles.numBundles, 100);
-
-        // (1) Force topic creation and namespace being loaded
-        final String topicName = "persistent://" + namespace + "/my-topic";
-        DestinationName destination = DestinationName.get(topicName);
-
-        Producer producer = pulsarClient.createProducer(topicName);
-        producer.close();
-        NamespaceBundle bundle1 = pulsar.getNamespaceService().getBundle(destination);
-        // (2) Delete topic
-        admin.persistentTopics().delete(topicName);
-        // (3) Delete ns
-        admin.namespaces().deleteNamespace(namespace);
-        // (4) check bundle
-        NamespaceBundle bundle2 = pulsar.getNamespaceService().getBundle(destination);
-        assertNotEquals(bundle1.getBundleRange(), bundle2.getBundleRange());
-        // returns full bundle if policies not present 
-        assertEquals("0x00000000_0xffffffff", bundle2.getBundleRange());
-
-    }
 }

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/admin/NamespacesTest.java

@@ -24,6 +24,7 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
@@ -63,6 +64,7 @@
 import com.yahoo.pulsar.broker.web.PulsarWebResource;
 import com.yahoo.pulsar.broker.web.RestException;
 import com.yahoo.pulsar.client.admin.PulsarAdminException;
+import com.yahoo.pulsar.client.api.Producer;
 import com.yahoo.pulsar.common.naming.DestinationName;
 import com.yahoo.pulsar.common.naming.NamespaceBundle;
 import com.yahoo.pulsar.common.naming.NamespaceBundles;
@@ -1063,6 +1065,37 @@ public void testIsLeader() throws Exception {
         assertTrue(namespaces.isLeaderBroker());
     }
 
+    /**
+     * Verifies that deleteNamespace cleans up policies(global,local), bundle cache and bundle ownership
+     * 
+     * @throws Exception
+     */
+    @Test
+    public void testDeleteNamespace() throws Exception {
+
+        final String namespace = this.testProperty + "/use/deleteNs";
+        admin.namespaces().createNamespace(namespace, 100);
+        assertEquals(admin.namespaces().getPolicies(namespace).bundles.numBundles, 100);
+
+        // (1) Force topic creation and namespace being loaded
+        final String topicName = "persistent://" + namespace + "/my-topic";
+        DestinationName destination = DestinationName.get(topicName);
+
+        Producer producer = pulsarClient.createProducer(topicName);
+        producer.close();
+        NamespaceBundle bundle1 = pulsar.getNamespaceService().getBundle(destination);
+        // (2) Delete topic
+        admin.persistentTopics().delete(topicName);
+        // (3) Delete ns
+        admin.namespaces().deleteNamespace(namespace);
+        // (4) check bundle
+        NamespaceBundle bundle2 = pulsar.getNamespaceService().getBundle(destination);
+        assertNotEquals(bundle1.getBundleRange(), bundle2.getBundleRange());
+        // returns full bundle if policies not present 
+        assertEquals("0x00000000_0xffffffff", bundle2.getBundleRange());
+
+    }
+
     private void mockWebUrl(URL localWebServiceUrl, NamespaceName namespace) throws Exception {
         doReturn(localWebServiceUrl).when(nsSvc).getWebServiceUrl(Mockito.argThat(new Matcher<NamespaceBundle>() {
             @Override

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/BrokerServiceTest.java

@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -728,4 +729,33 @@ public void testLookupThrottlingForClientByClient() throws Exception {
             // ok as throttling set to 0
         }
     }
+
+    @Test
+    public void testTopicLoadingOnDisableNamespaceBundle() throws Exception {
+        final String namespace = "prop/use/disableBundle";
+        admin.namespaces().createNamespace(namespace);
+
+        // own namespace bundle
+        final String topicName = "persistent://" + namespace + "/my-topic";
+        DestinationName destination = DestinationName.get(topicName);
+        Producer producer = pulsarClient.createProducer(topicName);
+        producer.close();
+
+        // disable namespace-bundle
+        NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(destination);
+        pulsar.getNamespaceService().getOwnershipCache().updateBundleState(bundle, false);
+
+        // try to create topic which should fail as bundle is disable
+        CompletableFuture<Topic> futureResult = pulsar.getBrokerService().createPersistentTopic(topicName);
+
+        try {
+            futureResult.get();
+            fail("Topic creation should fail due to disable bundle");
+        } catch (Exception e) {
+            if (!(e.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException)) {
+                fail("Topic creation should fail with ServiceUnitNotReadyException");
+            }
+
+        }
+    }
 }