remove redundant staticConfig field

Author: TaiJuWu

clients/src/main/resources/common/message/BrokerRegistrationRequest.json

@@ -66,7 +66,7 @@
     { "name": "CordonedLogDirs", "type":  "[]uuid", "versions":  "5+", "taggedVersions": "5+",
       "tag": "0", "about": "Log directories that are cordoned." },
     { "name": "StaticConfigs", "type": "[]StaticConfig", "versions": "6+",
-      "about": "Non-default static configs from the broker's server.properties.", "fields": [
+      "about": "static configs from the broker's server.properties and default value.", "fields": [
       { "name": "Name", "type": "string", "versions": "6+",
         "about": "The config name." },
       { "name": "Value", "type": "string", "versions": "6+", "nullableVersions": "6+",

metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java

@@ -271,11 +271,6 @@ boolean check() {
      */
     private final TimelineHashMap<Uuid, Integer> directoryToBroker;
 
-    /**
-     * Maps broker IDs to their non-default static configs reported during registration.
-     */
-    private final Map<ConfigResource, Map<String, String>> brokerStaticConfigs;
-
     /**
      * Manages the kafka.controller:type=KafkaController,name=TimeSinceLastHeartbeatReceivedMs,broker=<brokerId> metrics.
      */
@@ -305,7 +300,6 @@ private ClusterControlManager(
         this.featureControl = featureControl;
         this.controllerRegistrations = new TimelineHashMap<>(snapshotRegistry, 0);
         this.directoryToBroker = new TimelineHashMap<>(snapshotRegistry, 0);
-        this.brokerStaticConfigs = new HashMap<>();
         this.brokerShutdownHandler = brokerShutdownHandler;
         this.metrics = metrics;
     }
@@ -348,7 +342,14 @@ Map<Integer, BrokerRegistration> brokerRegistrations() {
     }
 
     public Map<ConfigResource, Map<String, String>> brokerStaticConfigs() {
-        return Collections.unmodifiableMap(brokerStaticConfigs);
+        Map<ConfigResource, Map<String, String>> result = new HashMap<>();
+        for (Map.Entry<Integer, BrokerRegistration> entry : brokerRegistrations.entrySet()) {
+            Map<String, String> statics = entry.getValue().staticConfigs();
+            if (!statics.isEmpty()) {
+                result.put(new ConfigResource(ConfigResource.Type.BROKER, String.valueOf(entry.getKey())), statics);
+            }
+        }
+        return Collections.unmodifiableMap(result);
     }
 
     Map<Integer, ControllerRegistration> controllerRegistrations() {
@@ -448,16 +449,6 @@ public ControllerResult<BrokerRegistrationReply> registerBroker(
             record.setCordonedLogDirs(request.cordonedLogDirs());
         }
 
-        // Store broker static configs in-memory (always available for pre-upgrade validation)
-        // Skip sensitive configs — their values are null
-        Map<String, String> statics = new HashMap<>();
-        for (BrokerRegistrationRequestData.StaticConfig sc : request.staticConfigs()) {
-            if (!sc.isSensitive()) {
-                statics.put(sc.name(), sc.value());
-            }
-        }
-        brokerStaticConfigs.put(new ConfigResource(ConfigResource.Type.BROKER, String.valueOf(brokerId)), Map.copyOf(statics));
-
         // Write static configs to the record when MV supports it
         if (featureControl.metadataVersionOrThrow().isStaticConfigReportingSupported()) {
             for (BrokerRegistrationRequestData.StaticConfig sc : request.staticConfigs()) {
@@ -612,6 +603,13 @@ public void replay(RegisterBrokerRecord record, long offset) {
             features.put(feature.name(), VersionRange.of(
                 feature.minSupportedVersion(), feature.maxSupportedVersion()));
         }
+        // Parse static configs from record, skipping sensitive ones (their values are null)
+        Map<String, String> staticConfigs = new HashMap<>();
+        for (RegisterBrokerRecord.BrokerStaticConfig sc : record.staticConfigs()) {
+            if (!sc.isSensitive()) {
+                staticConfigs.put(sc.name(), sc.value());
+            }
+        }
         // Update broker registrations.
         BrokerRegistration prevRegistration = brokerRegistrations.put(brokerId,
             new BrokerRegistration.Builder().
@@ -626,16 +624,8 @@ public void replay(RegisterBrokerRecord record, long offset) {
                 setIsMigratingZkBroker(record.isMigratingZkBroker()).
                 setDirectories(record.logDirs()).
                 setCordonedDirectories(record.cordonedLogDirs()).
+                setStaticConfigs(staticConfigs).
                     build());
-        // Restore static configs from metadata log (for controller failover)
-        // Skip sensitive configs — their values are null
-        Map<String, String> replayedStatics = new HashMap<>();
-        for (RegisterBrokerRecord.BrokerStaticConfig sc : record.staticConfigs()) {
-            if (!sc.isSensitive()) {
-                replayedStatics.put(sc.name(), sc.value());
-            }
-        }
-        brokerStaticConfigs.put(new ConfigResource(ConfigResource.Type.BROKER, String.valueOf(record.brokerId())), Map.copyOf(replayedStatics));
 
         updateDirectories(brokerId, prevRegistration == null ? null : prevRegistration.directories(), record.logDirs());
         if (heartbeatManager != null) {