From e0b754dd3938a2d142623001dbb15c92cc2f5cb4 Mon Sep 17 00:00:00 2001
From: Heesung Sohn <103456639+heesung-sn@users.noreply.github.com>
Date: Mon, 30 Sep 2024 18:45:22 -0700
Subject: [PATCH] [improve][broker] check system topic is used before
 configuring system topic (ExtensibleLoadManagerImpl only) (#23381)

---
 .../extensions/ExtensibleLoadManagerImpl.java          | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
index 841f9bfb669d4..d8a279b854576 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -50,6 +50,7 @@
 import org.apache.pulsar.broker.loadbalance.LoadManager;
 import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannel;
 import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateTableViewImpl;
 import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateTableViewSyncer;
 import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLoadData;
 import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
@@ -992,7 +993,7 @@ protected void monitor() {
             if (isChannelOwner) {
                 // System topic config might fail due to the race condition
                 // with topic policy init(Topic policies cache have not init).
-                if (!configuredSystemTopics) {
+                if (isPersistentSystemTopicUsed() && !configuredSystemTopics) {
                     configuredSystemTopics = configureSystemTopics(pulsar, COMPACTION_THRESHOLD);
                 }
                 if (role != Leader) {
@@ -1080,4 +1081,11 @@ boolean running() {
     private boolean disabled() {
         return state.get() == State.DISABLED;
     }
+
+    private boolean isPersistentSystemTopicUsed() {
+        return ServiceUnitStateTableViewImpl.class.getName()
+                .equals(pulsar.getConfiguration().getLoadManagerServiceUnitStateTableViewClassName());
+    }
+
+
 }