Refactor Pulsar service integration in CustomizedPulsarResourcesExtended and clean up unused variables

Author: coderzc

pulsar-broker/src/main/java/org/apache/pulsar/broker/DefaultPulsarResourcesExtended.java

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import lombok.Getter;
 import org.apache.pulsar.broker.namespace.NamespaceService;
 import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
 import org.apache.pulsar.common.naming.NamespaceName;
@@ -33,6 +34,7 @@
  */
 public class DefaultPulsarResourcesExtended implements PulsarResourcesExtended {
 
+    @Getter
     private PulsarService pulsarService;
 
     @Override

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -901,6 +901,8 @@ public void start() throws PulsarServerException {
                 shouldShutdownConfigurationMetadataStore = false;
             }
             pulsarResources = newPulsarResources();
+            // Initialize PulsarResourcesExtended
+            pulsarResourcesExtended = loadPulsarResourcesExtended();
 
             orderedExecutor = newOrderedExecutor();
 
@@ -921,9 +923,6 @@ public void start() throws PulsarServerException {
             // needs load management service and before start broker service,
             this.startNamespaceService();
 
-            // Initialize PulsarResourcesExtended
-            pulsarResourcesExtended = loadPulsarResourcesExtended();
-
             schemaStorage = createAndStartSchemaStorage();
             schemaRegistryService = SchemaRegistryService.create(
                     schemaStorage, config.getSchemaRegistryCompatibilityCheckers(), this);

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

@@ -59,7 +59,6 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.mutable.MutableBoolean;
 import org.apache.commons.lang3.tuple.Pair;
-import org.apache.pulsar.broker.PulsarResourcesExtended;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
@@ -195,8 +194,6 @@ public class NamespaceService implements AutoCloseable {
     private ConcurrentHashMap<String, CompletableFuture<List<String>>> inProgressQueryUserTopics =
             new ConcurrentHashMap<>();
 
-    private PulsarResourcesExtended pulsarResourcesExtended;
-
     /**
      * Default constructor.
      */
@@ -211,7 +208,6 @@ public NamespaceService(PulsarService pulsar) {
         this.bundleSplitListeners = new CopyOnWriteArrayList<>();
         this.localBrokerDataCache = pulsar.getLocalMetadataStore().getMetadataCache(LocalBrokerData.class);
         this.redirectManager = new RedirectManager(pulsar);
-        this.pulsarResourcesExtended = pulsar.getPulsarResourcesExtended();
 
         this.lookupLatencyHistogram = pulsar.getOpenTelemetry().getMeter()
                 .histogramBuilder(LOOKUP_REQUEST_DURATION_METRIC_NAME)
@@ -1539,7 +1535,7 @@ public CompletableFuture<List<String>> getListOfTopicsByProperties(NamespaceName
         if (MapUtils.isEmpty(properties)) {
             return getListOfTopics(namespaceName, mode);
         } else {
-            return pulsarResourcesExtended.listTopicOfNamespace(namespaceName, mode, properties);
+            return pulsar.getPulsarResourcesExtended().listTopicOfNamespace(namespaceName, mode, properties);
         }
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CustomizedPulsarResourcesExtended.java

@@ -39,7 +39,6 @@
 @Slf4j
 public class CustomizedPulsarResourcesExtended extends DefaultPulsarResourcesExtended {
 
-    private PulsarService pulsar;
     private boolean enabledTopicWatcher;
 
     // Map<Namespace, Map<Topic, Map<PropertyKey, PropertyValue>>>
@@ -89,9 +88,9 @@ public List<String> queryTopicListByProperties(String namespace, Map<String, Str
 
     public void setCustomProperties(TopicName topicName, Map<String, String> properties)
         throws ExecutionException, InterruptedException, TimeoutException {
-        int timeoutSeconds = pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds();
+        int timeoutSeconds = getPulsarService().getConfiguration().getMetadataStoreOperationTimeoutSeconds();
         PartitionedTopicMetadata partitionedTopicMetadata =
-            pulsar.getBrokerService().fetchPartitionedTopicMetadataAsync(topicName)
+            getPulsarService().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName)
                 .get(timeoutSeconds, TimeUnit.SECONDS);
         if (partitionedTopicMetadata.partitions == 0) {
             setNonPartitionedTopicCustomProperties(topicName, properties);
@@ -115,7 +114,7 @@ private void setNonPartitionedTopicCustomProperties(TopicName topicName, Map<Str
 
     @Override
     public void initialize(PulsarService pulsarService) {
-        this.pulsar = pulsarService;
+        super.initialize(pulsarService);
         this.enabledTopicWatcher = pulsarService.getConfiguration().isEnableBrokerTopicListWatcher();
         this.customTopicPropertiesMap = new ConcurrentHashMap<>();
     }