Versions Compared

Old Version 11

changes.mady.by.user TaiJuWu

Saved on

compared with

New Version 12

changes.mady.by.user TaiJuWu

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

This KIP introduces three interconnected features:

  1. Broker static reporting: Brokers report their static (non-sensitive) configurations to the controller during registration. This enables the controller to perform pre-flight validation of static configs during metadata version upgrades. On the broker side, BrokerLifecycleManager is responsible for collecting and sending static configs. During registration, it iterates  over all non-internal config entries.

  2. Metadata-Version-Aware Validators:

    :Configuration

     Configuration definitions

    enhance

    are enhanced to support metadata-version-aware validators. The ConfigDef.define() method accepts a mvValidators parameter —

    a navigableMap from feature level (short) to ConfigDef.Validator instances.

    Uses NavigableMap.floorEntry() to find the validator registered at the highest feature level ≤ the given featureLevel. This allows  constraints to be introduced incrementally across metadata versions.

    1. During AlterConfigs and IncrementalAlterConfigs operations, the
    2. During AlterConfigs and IncrementalAlterConfigs operations, the Metadata-Version-Aware validators will be invoked to ensure that any proposed changes are compliant with the cluster's current metadata version. Any configuration that fails this runtime validation will be rejected with an INVALID_CONFIG error.
    3. When an administrator performs a metadata.version upgrade or downgrade, the Controller will execute a comprehensive audit of all existing configurations—both static (via BrokerRegistrationRecord) and dynamic (via ConfigRecord). The transition will only proceed if all current settings satisfy the constraints of the target metadata version, ensuring a safe and atomic version migration.

Public Interfaces

ConfigDef

    1. When an administrator runs kafka-storage format --release-version <version>, the tool now validates the broker's static configs from  server.properties against the MV constraints of the specified release version. If any config violates the constraints, the format command fails with a descriptive error before writing any data to disk.

Public Interfaces

ConfigDef

We are introducing a new mvValidators field to the ConfigDef classWe are introducing a new mvValidators field to the ConfigDef class, enabling configurations to be validated against specific metadata.version thresholds. This allows the system to enforce version-specific constraints dynamically as the cluster evolves.

Code Block
languagejava
titleConfigDef
java
titleConfigDef
    public static class ConfigKey {
        public final String name;
        public final Type type;
        public final String documentation;
        public final Object defaultValue;
        public final Validator validator;
        public final Importance importance;
        public final String group;
        public final int orderInGroup;
        public final Width width;
        public final String displayName;
        public final List<String> dependents;
        public final Recommender recommender;
        public final boolean internalConfig;
        public final String alternativeString;
        public final NavigableMap<Short, Validator> mvValidators; // new field
...
}

  /**
   * Define a new configuration with both a base validator and MV-specific validators.
   * @param name          the name of the config parameter
   * @param type          the type of the config
   * @param defaultValue  the default value to use if this config isn't present
   * @param validator     the base validator to use in checking the correctness of the config
   * @param mvValidators  a sorted map from metadata version feature level to the validator
   *            public static class ConfigKey {
        public final String name;
        public final Type type;
   that applies once that version publicis finalreached
 String documentation;
 * @param importance    the publicimportance finalof Objectthis defaultValue;config
   * @param documentation the documentation publicstring finalfor Validatorthe validator;config
   * @return This ConfigDef so publicyou finalcan Importancechain importance;calls
    */
    public finalConfigDef define(String group;
name, Type type,      public final int orderInGroup;
Object defaultValue,
           public final Width width;
        public final String displayName;
 Validator validator,
      public final List<String> dependents;
        public final Recommender recommender;
      NavigableMap<Short, Validator> publicmvValidators,
 final boolean internalConfig;
        public final String alternativeString;
        public final NavigableMap<Short, Validator> mvValidators; // new field
...
}Importance importance, String documentation)

BrokerRegistrationRequest

...

Code Block
languagediff
linenumberstrue
diff --git a/clients/src/main/resources/common/message/BrokerRegistrationRequest.json b/clients/src/main/resources/common/message/BrokerRegistrationRequest.json
index 53e37f21d5..cdd53c4466 100644
--- a/clients/src/main/resources/common/message/BrokerRegistrationRequest.json
+++ b/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,15 @@
     { "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." }
+    ]}
   ]
 }
Code Block
languagediff
titleBrokerRegistrationResponse.json
linenumberstrue
// Version 1 adds Zk broker epoch to the request if the broker is migrating from Zk mode to KRaft mode.
// Version 2 adds the PreviousBrokerEpoch to the request for the KIP-966
// 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-6",
-  "validVersions": "0-5",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "Duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "BrokerEpoch", "type": "int64", "versions": "0+", "default": "-1",
      "about": "The broker's assigned epoch, or -1 if none was assigned." }
  ]
}

...

Code Block
languagediff
titleBrokerRegerationRecord.json
diff --git a/metadata/src/main/resources/common/metadata/RegisterBrokerRecord.json b/metadata/src/main/resources/common/metadata/RegisterBrokerRecord.json
index b7db680cbd..5570cbd96f 100644
--- a/metadata/src/main/resources/common/metadata/RegisterBrokerRecord.json
+++ b/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,16 @@
     { "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 byfrom the broker's server.properties and default value.", "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." },
+    ]}
   ]
 }


Compatibility, Deprecation, and Migration Plan

...