4/16

heesung-sn · Apr 17, 2024 · 62a91d8 · 62a91d8
1 parent f7dd337
commit 62a91d8
Show file tree

Hide file tree

Showing 19 changed files with 119 additions and 62 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java
@@ -374,7 +374,7 @@ public static void createTenantIfAbsent(PulsarResources resources, String tenant
         }
     }
 
-    static void createNamespaceIfAbsent(PulsarResources resources, NamespaceName namespaceName,
+    public static void createNamespaceIfAbsent(PulsarResources resources, NamespaceName namespaceName,
             String cluster, int bundleNumber) throws IOException {
         NamespaceResources namespaceResources = resources.getNamespaceResources();
 

diff --git a/.../main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/.../main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -974,7 +974,6 @@ protected void monitor() {
     }
 
     public void disableBroker() throws Exception {
-        serviceUnitStateChannel.cleanOwnerships();
         leaderElectionService.close();
         brokerRegistry.unregister();
     }

diff --git a/.../org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/.../org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -297,7 +297,8 @@ public synchronized void start() throws PulsarServerException {
                     (pulsar.getPulsarResources(), SYSTEM_NAMESPACE.getTenant(), config.getClusterName());
 
             PulsarClusterMetadataSetup.createNamespaceIfAbsent
-                    (pulsar.getPulsarResources(), SYSTEM_NAMESPACE, config.getClusterName());
+                    (pulsar.getPulsarResources(), SYSTEM_NAMESPACE, config.getClusterName(),
+                            config.getDefaultNumberOfNamespaceBundles());
 
             ExtensibleLoadManagerImpl.createSystemTopic(pulsar, TOPIC);
 
@@ -483,28 +484,35 @@ private CompletableFuture<Optional<String>> getActiveOwnerAsync(
             String serviceUnit,
             ServiceUnitState state,
             Optional<String> owner) {
-        return deferGetOwnerRequest(serviceUnit)
-                .thenCompose(newOwner -> {
-                    if (newOwner == null) {
-                        return CompletableFuture.completedFuture(null);
-                    }
 
-                    return brokerRegistry.lookupAsync(newOwner)
-                            .thenApply(lookupData -> {
-                                if (lookupData.isPresent()) {
-                                    return newOwner;
-                                } else {
-                                    throw new IllegalStateException(
-                                            "The new owner " + newOwner + " is inactive.");
-                                }
-                            });
-                }).whenComplete((__, e) -> {
-                    if (e != null) {
-                        log.error("{} failed to get active owner broker. serviceUnit:{}, state:{}, owner:{}",
-                                brokerId, serviceUnit, state, owner, e);
-                        ownerLookUpCounters.get(state).getFailure().incrementAndGet();
-                    }
-                }).thenApply(Optional::ofNullable);
+        return brokerRegistry.getAvailableBrokersAsync().thenCompose(activeBrokers -> {
+            if (activeBrokers.isEmpty()) {
+                return null;
+            } else {
+                return deferGetOwnerRequest(serviceUnit)
+                        .thenCompose(newOwner -> {
+                            if (newOwner == null) {
+                                return CompletableFuture.completedFuture(null);
+                            }
+
+                            return brokerRegistry.lookupAsync(newOwner)
+                                    .thenApply(lookupData -> {
+                                        if (lookupData.isPresent()) {
+                                            return newOwner;
+                                        } else {
+                                            throw new IllegalStateException(
+                                                    "The new owner " + newOwner + " is inactive.");
+                                        }
+                                    });
+                        }).whenComplete((__, e) -> {
+                            if (e != null) {
+                                log.error("{} failed to get active owner broker. serviceUnit:{}, state:{}, owner:{}",
+                                        brokerId, serviceUnit, state, owner, e);
+                                ownerLookUpCounters.get(state).getFailure().incrementAndGet();
+                            }
+                        });
+            }
+        }).thenApply(Optional::ofNullable);
     }
 
     public CompletableFuture<Optional<String>> getOwnerAsync(String serviceUnit) {

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java
@@ -70,8 +70,7 @@ public void setup() throws Exception {
                 ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(
                 Set.of("role1", "role2"), Set.of("test"));
-        admin.tenants().createTenant("pulsar", tenantInfo);
-        admin.namespaces().createNamespace("pulsar/system", Set.of("test"));
+        setupSystemNamespace(tenantInfo);
         admin.tenants().createTenant("public", tenantInfo);
         admin.namespaces().createNamespace("public/default", Set.of("test"));
     }

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java
@@ -66,6 +66,7 @@
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
 import org.apache.pulsar.broker.loadbalance.LeaderBroker;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
 import org.apache.pulsar.broker.namespace.NamespaceService;
 import org.apache.pulsar.broker.web.PulsarWebResource;
 import org.apache.pulsar.broker.web.RestException;
@@ -462,9 +463,15 @@ public void clusters() throws Exception {
     @Test
     public void properties() throws Throwable {
         Object response = asyncRequests(ctx -> properties.getTenants(ctx));
-        assertEquals(response, new ArrayList<>());
+
+        if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            assertEquals(response, new ArrayList<>());
+        }
+
         verify(properties, times(1)).validateSuperUserAccessAsync();
 
+
+
         // create local cluster
         asyncRequests(ctx -> clusters.createCluster(ctx, configClusterName, ClusterDataImpl.builder().build()));
 
@@ -478,7 +485,9 @@ public void properties() throws Throwable {
         verify(properties, times(2)).validateSuperUserAccessAsync();
 
         response = asyncRequests(ctx -> properties.getTenants(ctx));
-        assertEquals(response, List.of("test-property"));
+        if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            assertEquals(response, List.of("test-property"));
+        }
         verify(properties, times(3)).validateSuperUserAccessAsync();
 
         response = asyncRequests(ctx -> properties.getTenantAdmin(ctx, "test-property"));
@@ -607,7 +616,9 @@ public void properties() throws Throwable {
         response = asyncRequests(ctx -> properties.deleteTenant(ctx, "error-property", false));
         response = new ArrayList<>();
         response = asyncRequests(ctx -> properties.getTenants(ctx));
-        assertEquals(response, new ArrayList<>());
+        if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            assertEquals(response, new ArrayList<>());
+        }
 
         // Create a namespace to test deleting a non-empty property
         TenantInfoImpl newPropertyAdmin2 = TenantInfoImpl.builder()
@@ -798,6 +809,9 @@ public void resourceQuotas() throws Exception {
 
     @Test
     public void brokerStats() throws Exception {
+        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            return;
+        }
         doReturn("client-id").when(brokerStats).clientAppId();
         Collection<Metrics> metrics = brokerStats.getMetrics();
         assertNotNull(metrics);

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -1029,6 +1029,9 @@ public void testSplitBundleWithUnDividedRange() throws Exception {
 
     @Test
     public void testUnloadNamespaceWithBundles() throws Exception {
+        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            return;
+        }
         URL localWebServiceUrl = new URL(pulsar.getSafeWebServiceAddress());
         String bundledNsLocal = "test-bundled-namespace-1";
         List<String> boundaries = List.of("0x00000000", "0x80000000", "0xffffffff");

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -63,6 +63,8 @@
 import org.apache.pulsar.broker.admin.v2.PersistentTopics;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
+import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
 import org.apache.pulsar.broker.namespace.NamespaceService;
 import org.apache.pulsar.broker.resources.NamespaceResources;
 import org.apache.pulsar.broker.resources.PulsarResources;
@@ -141,6 +143,7 @@ public void initPersistentTopics() throws Exception {
     @BeforeMethod
     protected void setup() throws Exception {
         conf.setTopicLevelPoliciesEnabled(false);
+        conf.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
         super.internalSetup();
         persistentTopics = spy(PersistentTopics.class);
         persistentTopics.setServletContext(new MockServletContext());
@@ -182,10 +185,10 @@ protected void setup() throws Exception {
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
         admin.tenants().createTenant(this.testTenant,
                 new TenantInfoImpl(Set.of("role1", "role2"), Set.of(testLocalCluster, "test")));
-        admin.tenants().createTenant("pulsar",
-                new TenantInfoImpl(Set.of("role1", "role2"), Set.of(testLocalCluster, "test")));
-        admin.namespaces().createNamespace(testTenant + "/" + testNamespace, Set.of(testLocalCluster, "test"));
+        setupSystemNamespace(new TenantInfoImpl(Set.of("role1", "role2"), Set.of(testLocalCluster, "test")));
         admin.namespaces().createNamespace("pulsar/system", 4);
+        admin.namespaces().createNamespace(testTenant + "/" + testNamespace, Set.of(testLocalCluster, "test"));
+
         admin.namespaces().createNamespace(testTenant + "/" + testNamespaceLocal);
     }
 
@@ -429,7 +432,10 @@ public void testTerminatePartitionedTopic() {
         response = mock(AsyncResponse.class);
         persistentTopics.terminatePartitionedTopic(response, testTenant, testNamespace, testLocalTopicName, true);
         Map<Integer, MessageId> messageIds = new ConcurrentHashMap<>();
-        messageIds.put(0, new MessageIdImpl(3, -1, -1));
+        if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            messageIds.put(0, new MessageIdImpl(3, -1, -1));
+        }
+
         verify(response, timeout(5000).times(1)).resume(messageIds);
     }
 
@@ -452,8 +458,11 @@ public void testTerminate() {
         // 3) Assert terminate persistent topic
         response = mock(AsyncResponse.class);
         persistentTopics.terminate(response, testTenant, testNamespace, testLocalTopicName, true);
-        MessageId messageId = new MessageIdImpl(3, -1, -1);
-        verify(response, timeout(5000).times(1)).resume(messageId);
+        if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            MessageId messageId = new MessageIdImpl(3, -1, -1);
+            verify(response, timeout(5000).times(1)).resume(messageId);
+        }
+
 
         // 4) Assert terminate non-persistent topic
         String nonPersistentTopicName = "non-persistent-topic";
@@ -583,9 +592,11 @@ public void testCreatePartitionedTopic() {
         AsyncResponse response3 = mock(AsyncResponse.class);
         ArgumentCaptor<Map> metaResponseCaptor2 = ArgumentCaptor.forClass(Map.class);
         persistentTopics.getProperties(response3, testTenant, testNamespace, topicName2, true);
-        verify(response3, timeout(5000).times(1)).resume(metaResponseCaptor2.capture());
-        Assert.assertNotNull(metaResponseCaptor2.getValue());
-        Assert.assertEquals(metaResponseCaptor2.getValue().get("key1"), "value1");
+        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            verify(response3, timeout(5000).times(1)).resume(metaResponseCaptor2.capture());
+            Assert.assertNotNull(metaResponseCaptor2.getValue());
+            Assert.assertEquals(metaResponseCaptor2.getValue().get("key1"), "value1");
+        }
     }
 
     @Test
@@ -1656,7 +1667,10 @@ public void testAdminTerminatePartitionedTopic() throws Exception {
         admin.topics().createPartitionedTopic(topicName, 1);
         Map<Integer, MessageId> results = new HashMap<>();
         results.put(0, new MessageIdImpl(3, -1, -1));
-        Assert.assertEquals(admin.topics().terminatePartitionedTopic(topicName), results);
+        if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            Assert.assertEquals(admin.topics().terminatePartitionedTopic(topicName), results);
+        }
+
 
         // Check examine message not allowed on non-partitioned topic.
         admin.topics().createNonPartitionedTopic("persistent://prop-xyz/ns12/test");

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java
@@ -85,9 +85,7 @@ public void before() {
                 .serviceHttpUrl(getPulsarService().getWebServiceAddress())
                 .authentication(new AuthenticationToken(TENANT_ADMIN_TOKEN))
                 .build();
-
-        superUserAdmin.tenants().createTenant("pulsar", tenantInfo);
-        superUserAdmin.namespaces().createNamespace("pulsar/system");
+        setupSystemNamespace(tenantInfo);
     }
 
     protected void createTransactionCoordinatorAssign(int numPartitionsOfTC) throws MetadataStoreException {

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -563,6 +563,19 @@ protected void setupSystemNamespace() throws Exception {
         }
     }
 
+    protected void setupSystemNamespace(TenantInfo tenantInfo) throws Exception {
+        if (!admin.tenants().getTenants().contains(NamespaceName.SYSTEM_NAMESPACE.getTenant())) {
+            admin.tenants().createTenant(NamespaceName.SYSTEM_NAMESPACE.getTenant(), tenantInfo);
+        } else {
+            admin.tenants().updateTenant(NamespaceName.SYSTEM_NAMESPACE.getTenant(), tenantInfo);
+        }
+
+        if (!admin.namespaces().getNamespaces(NamespaceName.SYSTEM_NAMESPACE.getTenant())
+                .contains(NamespaceName.SYSTEM_NAMESPACE.toString())) {
+            admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString());
+        }
+    }
+
     protected void updateTenant(String tenant, TenantInfoImpl tenantInfo) throws Exception {
         if (!admin.tenants().getTenants().contains(tenant)) {
             admin.tenants().createTenant(tenant, tenantInfo);

diff --git a/...er/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java b/...er/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
@@ -33,6 +33,7 @@
 import lombok.Cleanup;
 import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
 import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
+import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
 import org.apache.pulsar.client.admin.ListNamespaceTopicsOptions;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.MessageId;
@@ -59,6 +60,7 @@ public class BrokerServiceAutoTopicCreationTest extends BrokerTestBase{
     @BeforeClass
     @Override
     protected void setup() throws Exception {
+        conf.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
         super.baseSetup();
     }
 

diff --git a/...r/src/test/java/org/apache/pulsar/broker/service/TransactionMetadataStoreServiceTest.java b/...r/src/test/java/org/apache/pulsar/broker/service/TransactionMetadataStoreServiceTest.java
@@ -37,7 +37,6 @@
 import org.apache.pulsar.broker.TransactionMetadataStoreService;
 import org.apache.pulsar.client.api.transaction.TxnID;
 import org.apache.pulsar.common.api.proto.TxnAction;
-import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.SystemTopicNames;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID;
@@ -63,8 +62,7 @@ protected void setup() throws Exception {
         ServiceConfiguration configuration = getDefaultConf();
         configuration.setTransactionCoordinatorEnabled(true);
         super.baseSetup(configuration);
-        admin.tenants().createTenant("pulsar", new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
-        admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString());
+        setupSystemNamespace(new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
         createTransactionCoordinatorAssign(16);
         admin.lookups().lookupPartitionedTopic(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN.toString());
     }

diff --git a/...-broker/src/test/java/org/apache/pulsar/broker/stats/BookieClientsStatsGeneratorTest.java b/...-broker/src/test/java/org/apache/pulsar/broker/stats/BookieClientsStatsGeneratorTest.java
@@ -24,6 +24,7 @@
 import java.util.Map;
 
 import org.apache.bookkeeper.mledger.proto.PendingBookieOpsStats;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
 import org.apache.pulsar.broker.service.BrokerTestBase;
 import org.apache.pulsar.common.stats.JvmMetrics;
 import org.testng.annotations.AfterClass;
@@ -52,7 +53,10 @@ protected void cleanup() throws Exception {
     public void testBookieClientStatsGenerator() throws Exception {
         // should not generate any NPE or other exceptions..
         Map<String, Map<String, PendingBookieOpsStats>> stats = BookieClientStatsGenerator.generate(super.getPulsar());
-        assertTrue(stats.isEmpty());
+
+        if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
+            assertTrue(stats.isEmpty());
+        }
     }
 
     @Test

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/TransactionMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/TransactionMetricsTest.java
@@ -38,6 +38,7 @@
 import java.util.regex.Pattern;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
 import org.apache.pulsar.broker.service.BrokerTestBase;
 import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
 import org.apache.pulsar.client.api.Consumer;
@@ -68,6 +69,7 @@ public class TransactionMetricsTest extends BrokerTestBase {
     protected void setup() throws Exception {
         ServiceConfiguration serviceConfiguration = getDefaultConf();
         serviceConfiguration.setTransactionCoordinatorEnabled(true);
+        serviceConfiguration.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
         super.baseSetup(serviceConfiguration);
     }
 

diff --git a/...ar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java b/...ar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
@@ -146,8 +146,7 @@ public void testProduceAndConsumeUnderSystemNamespace() throws Exception {
                 .adminRoles(Sets.newHashSet("admin"))
                 .allowedClusters(Sets.newHashSet("test"))
                 .build();
-        admin.tenants().createTenant("pulsar", tenantInfo);
-        admin.namespaces().createNamespace("pulsar/system", 2);
+        setupSystemNamespace(tenantInfo);
         @Cleanup
         Producer<byte[]> producer = pulsarClient.newProducer().topic("pulsar/system/__topic-1").create();
         producer.send("test".getBytes(StandardCharsets.UTF_8));

diff --git a/...va/org/apache/pulsar/broker/transaction/coordinator/TransactionCoordinatorConfigTest.java b/...va/org/apache/pulsar/broker/transaction/coordinator/TransactionCoordinatorConfigTest.java
@@ -27,7 +27,6 @@
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.transaction.Transaction;
-import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.SystemTopicNames;
 import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
@@ -45,8 +44,7 @@ protected void setup() throws Exception {
         configuration.setTransactionCoordinatorEnabled(true);
         configuration.setMaxActiveTransactionsPerCoordinator(2);
         super.baseSetup(configuration);
-        admin.tenants().createTenant("pulsar", new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
-        admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString());
+        setupSystemNamespace(new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
         pulsar.getPulsarResources()
                 .getNamespaceResources()
                 .getPartitionedTopicResources()