fix: allow automq.zone.cidr.blocks to be set via static server.properties (#3154)

Author: manmohak07

core/src/main/java/kafka/automq/zerozone/DefaultClientRackProvider.java

@@ -18,9 +18,9 @@
  */
 
 package kafka.automq.zerozone;
-
 import kafka.automq.interceptor.ClientIdMetadata;
 import kafka.server.DynamicBrokerConfig;
+import kafka.server.KafkaConfig;
 
 import org.apache.kafka.common.Reconfigurable;
 import org.apache.kafka.common.config.ConfigException;
@@ -39,6 +39,8 @@ public class DefaultClientRackProvider implements ClientRackProvider, Reconfigur
     private static final Logger LOGGER = LoggerFactory.getLogger(DefaultClientRackProvider.class);
     private static final String ZONE_CIDR_BLOCKS_CONFIG_KEY = "automq.zone.cidr.blocks";
     private static final Set<String> RECONFIGURABLE_CONFIGS;
+    
+    private final KafkaConfig kafkaConfig;
     private CIDRMatcher cidrMatcher = new CIDRMatcher("");
 
     static {
@@ -48,6 +50,16 @@ public class DefaultClientRackProvider implements ClientRackProvider, Reconfigur
         RECONFIGURABLE_CONFIGS.forEach(DynamicBrokerConfig.AllDynamicConfigs()::add);
     }
 
+    public DefaultClientRackProvider(KafkaConfig kafkaConfig) {
+        this.kafkaConfig = kafkaConfig;
+        // Read static config from server.properties on initialization
+        final String staticValue = (String) kafkaConfig.originals().get(ZONE_CIDR_BLOCKS_CONFIG_KEY);
+        if (staticValue != null) {
+            this.cidrMatcher = new CIDRMatcher(staticValue);
+            LOGGER.info("Initialized with static zone CIDR blocks: {}", staticValue);
+        }
+    }
+
     @Override
     public String rack(ClientIdMetadata clientId) {
         String rack = clientId.rack();

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -170,7 +170,7 @@ class BrokerServer(
 
   var backPressureManager: BackPressureManager = _
 
-  val clientRackProvider = new DefaultClientRackProvider()
+  val clientRackProvider = new DefaultClientRackProvider(config)
   // init reconfigurable before startup
   config.addReconfigurable(clientRackProvider)