add static config reporting

Author: TaiJuWu

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

@@ -18,12 +18,13 @@
 // Version 3 adds the PreviousBrokerEpoch for the KIP-966
 // Version 4 fixes KAFKA-17011, which blocked SupportedFeatures.MinVersion in the response from being 0.
 // Version 5 adds the CordonedLogDirs flexible field
+// Version 6 adds StaticConfigs for broker static config reporting.
 {
   "apiKey":62,
   "type": "request",
   "listeners": ["controller"],
   "name": "BrokerRegistrationRequest",
-  "validVersions": "0-5",
+  "validVersions": "0-6",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
@@ -63,6 +64,13 @@
     { "name": "PreviousBrokerEpoch", "type": "int64", "versions": "3+", "default": "-1", "ignorable": true,
       "about": "The epoch before a clean shutdown." },
     { "name": "CordonedLogDirs", "type":  "[]uuid", "versions":  "5+", "taggedVersions": "5+",
-      "tag": "0", "about": "Log directories that are cordoned." }
+      "tag": "0", "about": "Log directories that are cordoned." },
+    { "name": "StaticConfigs", "type": "[]StaticConfig", "versions": "6+",
+      "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+",
+        "about": "The config value. Null if the config is sensitive." }
+    ]}
   ]
-}
+}

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

@@ -18,11 +18,12 @@
 // Version 3 is the same as version 2 (new field in request).
 // Version 4 is the same as version 2 (new field in request).
 // Version 5 is the same as version 2 (new field in request).
+// Version 6 is the same as version 2 (new field in request).
 {
   "apiKey": 62,
   "type": "response",
   "name": "BrokerRegistrationResponse",
-  "validVersions": "0-5",
+  "validVersions": "0-6",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",

metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.metadata.RegisterBrokerRecord;
 import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerEndpoint;
 import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerFeature;
+import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerStaticConfig;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.image.writer.ImageWriterOptions;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
@@ -55,6 +56,7 @@ public static class Builder {
         private boolean isMigratingZkBroker;
         private List<Uuid> directories;
         private List<Uuid> cordonedDirectories;
+        private Map<String, String> staticConfigs;
 
         public Builder() {
             this.id = 0;
@@ -68,6 +70,7 @@ public Builder() {
             this.isMigratingZkBroker = false;
             this.directories = List.of();
             this.cordonedDirectories = List.of();
+            this.staticConfigs = Map.of();
         }
 
         public Builder setId(int id) {
@@ -135,6 +138,11 @@ public Builder setCordonedDirectories(List<Uuid> cordonedDirectories) {
             return this;
         }
 
+        public Builder setStaticConfigs(Map<String, String> staticConfigs) {
+            this.staticConfigs = staticConfigs;
+            return this;
+        }
+
         public BrokerRegistration build() {
             return new BrokerRegistration(
                 id,
@@ -147,7 +155,8 @@ public BrokerRegistration build() {
                 inControlledShutdown,
                 isMigratingZkBroker,
                 directories,
-                cordonedDirectories);
+                cordonedDirectories,
+                staticConfigs);
         }
     }
 
@@ -162,6 +171,7 @@ public BrokerRegistration build() {
     private final boolean isMigratingZkBroker;
     private final List<Uuid> directories;
     private final List<Uuid> cordonedDirectories;
+    private final Map<String, String> staticConfigs;
 
     private BrokerRegistration(
         int id,
@@ -174,7 +184,8 @@ private BrokerRegistration(
         boolean inControlledShutdown,
         boolean isMigratingZkBroker,
         List<Uuid> directories,
-        List<Uuid> cordonedDirectories
+        List<Uuid> cordonedDirectories,
+        Map<String, String> staticConfigs
     ) {
         this.id = id;
         this.epoch = epoch;
@@ -197,6 +208,7 @@ private BrokerRegistration(
         directories.sort(Uuid::compareTo);
         this.directories = Collections.unmodifiableList(directories);
         this.cordonedDirectories = Collections.unmodifiableList(cordonedDirectories);
+        this.staticConfigs = Collections.unmodifiableMap(staticConfigs);
     }
 
     public static BrokerRegistration fromRecord(RegisterBrokerRecord record) {
@@ -212,6 +224,10 @@ public static BrokerRegistration fromRecord(RegisterBrokerRecord record) {
             supportedFeatures.put(feature.name(), VersionRange.of(
                 feature.minSupportedVersion(), feature.maxSupportedVersion()));
         }
+        Map<String, String> staticConfigs = new HashMap<>();
+        for (BrokerStaticConfig sc : record.staticConfigs()) {
+            staticConfigs.put(sc.name(), sc.value());
+        }
         return new BrokerRegistration(record.brokerId(),
             record.brokerEpoch(),
             record.incarnationId(),
@@ -222,7 +238,8 @@ public static BrokerRegistration fromRecord(RegisterBrokerRecord record) {
             record.inControlledShutdown(),
             record.isMigratingZkBroker(),
             record.logDirs(),
-            record.cordonedLogDirs());
+            record.cordonedLogDirs(),
+            staticConfigs);
     }
 
     public int id() {
@@ -277,6 +294,10 @@ public List<Uuid> cordonedDirectories() {
         return cordonedDirectories;
     }
 
+    public Map<String, String> staticConfigs() {
+        return staticConfigs;
+    }
+
     public boolean hasOnlineDir(Uuid dir) {
         return DirectoryId.isOnline(dir, directories);
     }
@@ -353,14 +374,22 @@ public ApiMessageAndVersion toRecord(ImageWriterOptions options) {
                 setMaxSupportedVersion(entry.getValue().max()));
         }
 
+        if (options.metadataVersion().isStaticConfigReportingSupported()) {
+            for (Entry<String, String> entry : staticConfigs.entrySet()) {
+                registrationRecord.staticConfigs().add(new BrokerStaticConfig().
+                        setName(entry.getKey()).
+                        setValue(entry.getValue()));
+            }
+        }
+
         return new ApiMessageAndVersion(registrationRecord,
             options.metadataVersion().registerBrokerRecordVersion());
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(id, epoch, incarnationId, listeners, supportedFeatures,
-            rack, fenced, inControlledShutdown, isMigratingZkBroker, directories, cordonedDirectories);
+                rack, fenced, inControlledShutdown, isMigratingZkBroker, directories, cordonedDirectories, staticConfigs);
     }
 
     @Override
@@ -376,7 +405,8 @@ public boolean equals(Object o) {
             other.inControlledShutdown == inControlledShutdown &&
             other.isMigratingZkBroker == isMigratingZkBroker &&
             other.directories.equals(directories) &&
-            other.cordonedDirectories.equals(cordonedDirectories);
+            other.cordonedDirectories.equals(cordonedDirectories) &&
+            other.staticConfigs.equals(staticConfigs);
     }
 
     @Override
@@ -399,6 +429,7 @@ public String toString() {
                 ", isMigratingZkBroker=" + isMigratingZkBroker +
                 ", directories=" + directories +
                 ", cordonedDirectories=" + cordonedDirectories +
+                ", staticConfigs=" + staticConfigs +
                 ")";
     }
 
@@ -430,7 +461,8 @@ public BrokerRegistration cloneWith(
             newInControlledShutdownChange,
             isMigratingZkBroker,
             newDirectories,
-            newCordonedDirectories
+            newCordonedDirectories,
+            staticConfigs
         );
     }
-}
+}

metadata/src/main/resources/common/metadata/RegisterBrokerRecord.json

@@ -17,11 +17,12 @@
 // Version 2 adds IsMigratingZkBroker
 // Version 3 adds LogDirs
 // Version 4 adds CordonedLogDirs
+// Version 5 adds StaticConfigs for broker static config reporting.
 {
   "apiKey": 0,
   "type": "metadata",
   "name": "RegisterBrokerRecord",
-  "validVersions": "0-4",
+  "validVersions": "0-5",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
@@ -61,6 +62,14 @@
     { "name": "LogDirs", "type":  "[]uuid", "versions":  "3+", "taggedVersions": "3+", "tag": 0,
       "about": "Log directories configured in this broker which are available." },
     { "name": "CordonedLogDirs", "type":  "[]uuid", "versions":  "4+", "taggedVersions": "4+", "tag": "1",
-      "about": "Log directories that are cordoned." }
+      "about": "Log directories that are cordoned." },
+    { "name": "StaticConfigs", "type": "[]BrokerStaticConfig", "versions": "5+",
+      "taggedVersions": "5+", "tag": 2,
+      "about": "Non-default static configs reported by the broker.", "fields": [
+      { "name": "Name", "type": "string", "versions": "5+",
+        "about": "The config name." },
+      { "name": "Value", "type": "string", "versions": "5+", "nullableVersions": "5+",
+        "about": "The config value. Null if the config is sensitive." }
+    ]}
   ]
-}
+}

server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java

@@ -130,7 +130,9 @@ public enum MetadataVersion {
     // they have set the configuration unstable.feature.versions.enable=true.
     // Please move this comment when updating the LATEST_PRODUCTION constant.
     //
-    IBP_4_4_IV0(31, "4.4", "IV0", false);
+    IBP_4_4_IV0(31, "4.4", "IV0", false),
+
+    IBP_4_4_IV1(32, "4.4", "IV1", true);
 
 
     // NOTES when adding a new version:
@@ -219,8 +221,14 @@ public boolean isCordonedLogDirsSupported() {
         return this.isAtLeast(MetadataVersion.IBP_4_3_IV0);
     }
 
+    public boolean isStaticConfigReportingSupported() {
+        return this.isAtLeast(MetadataVersion.IBP_4_4_IV1);
+    }
+
     public short registerBrokerRecordVersion() {
-        if (isCordonedLogDirsSupported()) {
+        if (isStaticConfigReportingSupported()) {
+            return (short) 5;
+        } else if (isCordonedLogDirsSupported()) {
             // new cordonedLogDirs field
             return (short) 4;
         } else if (isDirectoryAssignmentSupported()) {

server/src/main/java/org/apache/kafka/server/BrokerLifecycleManager.java

@@ -517,6 +517,12 @@ private void sendBrokerRegistration() {
         );
         List<Uuid> sortedLogDirs = new ArrayList<>(logDirs);
         sortedLogDirs.sort(Uuid::compareTo);
+        List<BrokerRegistrationRequestData.StaticConfig> staticConfigs = new ArrayList<>();
+        for (Map.Entry<String, ?> entry : config.values().entrySet()) {
+            staticConfigs.add(new BrokerRegistrationRequestData.StaticConfig()
+                .setName(entry.getKey())
+                .setValue(String.valueOf(entry.getValue())));
+        }
         BrokerRegistrationRequestData data = new BrokerRegistrationRequestData()
             .setBrokerId(nodeId)
             .setIsMigratingZkBroker(false)
@@ -527,7 +533,8 @@ private void sendBrokerRegistration() {
             .setRack(rack.orElse(null))
             .setPreviousBrokerEpoch(previousBrokerEpoch.orElse(-1L))
             .setLogDirs(sortedLogDirs)
-            .setCordonedLogDirs(cordonedLogDirs.entrySet().stream().filter(Map.Entry::getValue).map(Map.Entry::getKey).toList());
+            .setCordonedLogDirs(cordonedLogDirs.entrySet().stream().filter(Map.Entry::getValue).map(Map.Entry::getKey).toList())
+            .setStaticConfigs(staticConfigs);
         if (logger.isDebugEnabled()) {
             logger.debug("Sending broker registration {}", data);
         }