4/17 part 2

heesung-sn · Apr 18, 2024 · 888d764 · 888d764
1 parent b1db27e
commit 888d764
Show file tree

Hide file tree

Showing 3 changed files with 39 additions and 18 deletions.
diff --git a/.../src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/.../src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -34,6 +34,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import javax.annotation.Nonnull;
+import org.apache.commons.lang3.concurrent.ConcurrentInitializer;
+import org.apache.commons.lang3.concurrent.LazyInitializer;
 import org.apache.commons.lang3.tuple.MutablePair;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.PulsarService;
@@ -70,7 +72,19 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
     private final PulsarService pulsarService;
     private final HashSet localCluster;
     private final String clusterName;
-    private volatile NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
+
+    private final ConcurrentInitializer<NamespaceEventsSystemTopicFactory>
+            namespaceEventsSystemTopicFactoryLazyInitializer = new LazyInitializer<>() {
+        @Override
+        protected NamespaceEventsSystemTopicFactory initialize() {
+            try {
+                return new NamespaceEventsSystemTopicFactory(pulsarService.getClient());
+            } catch (PulsarServerException e) {
+                log.error("Create namespace event system topic factory error.", e);
+                throw new RuntimeException(e);
+            }
+        }
+    };
 
     @VisibleForTesting
     final Map<TopicName, TopicPolicies> policiesCache = new ConcurrentHashMap<>();
@@ -102,7 +116,7 @@ public SystemTopicBasedTopicPoliciesService(PulsarService pulsarService) {
                     });
                 })
                 .buildAsync((namespaceName, executor) -> {
-                    SystemTopicClient<PulsarEvent> systemTopicClient = namespaceEventsSystemTopicFactory
+                    SystemTopicClient<PulsarEvent> systemTopicClient = getNamespaceEventsSystemTopicFactory()
                             .createTopicPoliciesSystemTopicClient(namespaceName);
                     return systemTopicClient.newWriterAsync();
                 });
@@ -301,7 +315,7 @@ public CompletableFuture<TopicPolicies> getTopicPoliciesBypassCacheAsync(TopicNa
             result.complete(null);
             return result;
         }
-        SystemTopicClient<PulsarEvent> systemTopicClient = namespaceEventsSystemTopicFactory
+        SystemTopicClient<PulsarEvent> systemTopicClient = getNamespaceEventsSystemTopicFactory()
                 .createTopicPoliciesSystemTopicClient(topicName.getNamespaceObject());
         systemTopicClient.newReaderAsync().thenAccept(r ->
                 fetchTopicPoliciesAsyncAndCloseReader(r, topicName, null, result));
@@ -373,7 +387,7 @@ protected CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> createSystemT
         } catch (PulsarServerException ex) {
             return FutureUtil.failedFuture(ex);
         }
-        final SystemTopicClient<PulsarEvent> systemTopicClient = namespaceEventsSystemTopicFactory
+        final SystemTopicClient<PulsarEvent> systemTopicClient = getNamespaceEventsSystemTopicFactory()
                 .createTopicPoliciesSystemTopicClient(namespace);
         return systemTopicClient.newReaderAsync();
     }
@@ -561,7 +575,7 @@ private void refreshTopicPoliciesCache(Message<PulsarEvent> msg) {
                         log.error("Failed to create system topic factory");
                         break;
                     }
-                    SystemTopicClient<PulsarEvent> systemTopicClient = namespaceEventsSystemTopicFactory
+                    SystemTopicClient<PulsarEvent> systemTopicClient = getNamespaceEventsSystemTopicFactory()
                             .createTopicPoliciesSystemTopicClient(topicName.getNamespaceObject());
                     systemTopicClient.newWriterAsync().thenAccept(writer
                             -> writer.deleteAsync(getEventKey(topicName),
@@ -595,18 +609,18 @@ private boolean hasReplicateTo(Message<?> message) {
     }
 
     private void createSystemTopicFactoryIfNeeded() throws PulsarServerException {
-        if (namespaceEventsSystemTopicFactory == null) {
-            synchronized (this) {
-                if (namespaceEventsSystemTopicFactory == null) {
-                    try {
-                        namespaceEventsSystemTopicFactory =
-                                new NamespaceEventsSystemTopicFactory(pulsarService.getClient());
-                    } catch (PulsarServerException e) {
-                        log.error("Create namespace event system topic factory error.", e);
-                        throw e;
-                    }
-                }
-            }
+        try {
+            getNamespaceEventsSystemTopicFactory();
+        } catch (Exception e) {
+            throw new PulsarServerException(e);
+        }
+    }
+
+    private NamespaceEventsSystemTopicFactory getNamespaceEventsSystemTopicFactory() {
+        try {
+            return namespaceEventsSystemTopicFactoryLazyInitializer.get();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
         }
     }
 

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -1621,7 +1621,12 @@ public void testDoNotReplicateSystemTopic() throws Exception {
 
     private void initTransaction(int coordinatorSize, PulsarAdmin admin, String ServiceUrl,
                                  PulsarService pulsarService) throws Exception {
-        admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString(), coordinatorSize);
+
+        if (!admin.namespaces().getNamespaces(NamespaceName.SYSTEM_NAMESPACE.getTenant())
+                .contains(NamespaceName.SYSTEM_NAMESPACE.toString())) {
+            admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString(), coordinatorSize);
+        }
+
         pulsarService.getPulsarResources()
                 .getNamespaceResources()
                 .getPartitionedTopicResources()

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TenantTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TenantTest.java
@@ -20,6 +20,7 @@
 
 import com.google.common.collect.Sets;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
@@ -45,6 +46,7 @@ protected void cleanup() throws Exception {
 
     @Test
     public void testMaxTenant() throws Exception {
+        conf.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
         conf.setMaxTenants(2);
         super.internalSetup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());