This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: UNDER DISCUSS

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Apache Kafka’s configuration validation is currently version-agnostic. Validation logic is defined in ConfigDef and executed during broker startup, which tightly couples the broker’s static codebase with the cluster’s operational configuration state.

This makes it difficult to strengthen configuration constraints over time. For example, if the minimum allowed value for log.segment.bytes is increased (e.g., from 1 byte to 3 MB), brokers with legacy values in server.properties will fail to start after a binary upgrade due to a ConfigException. Because the broker cannot join the cluster, administrators cannot use dynamic APIs such as AlterConfigs to fix the configuration and must instead manually update configuration files across the fleet.

To address the limitations of the current static, version-agnostic validation model, Kafka needs a mechanism to enforce configuration constraints tied to metadata.version. As Kafka evolves, configuration values that were previously valid may become unsafe or semantically incorrect in newer versions. Without version-aware validation, two issues arise:

  1. Unsafe metadata version upgrades: administrators may upgrade metadata.version without realizing that existing configurations violate new constraints.

  2. No Metadata-version-aware enforcement: administrators can still use AlterConfigs or IncrementalAlterConfigs to set values that violate constraints introduced in newer metadata versions.

Additionally, the controller currently has no visibility into broker static configurations (those set in server.properties). Since these configs need to be changed dynamically, they must be validated before an upgrade is allowed — otherwise, a broker whose static config violates a new constraint would fail after the upgrade.

Proposed Changes

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 config entries need to be validated.

  2. Metadata-Version-Aware Validators: Configuration definitions are enhanced to support metadata-version-aware validators. The ConfigDef.define() method accepts a mvValidators parameter — 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. 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.
    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.

Example:

Suppose IBP_4_4_IV1 raises the minimum allowed value for log.segment.bytes from 1 byte to 3 MB. This constraint is registered as an MV-aware validator on the log.segment.bytes config key. Because log.segment.bytes now has an MV validator, it will be included in the static config payload that brokers send to the controller during registration.

  1. Format time:  An administrator prepares a new broker with log.segment.bytes=1024 in server.properties and runs kafka-storage format --release-version 4.4. Before writing any data to disk, the tool validates the static configs against the constraints of IBP_4_4_IV1 and finds that log.segment.bytes=1024 violates the minimum of 3 MB. The command fails with a descriptive error, prompting the administrator to fix server.properties before retrying.
  2. AlterConfigs at runtime: Once the cluster is running at IBP_4_4_IV1, an administrator attempts to set log.segment.bytes=1024 on a topic via AlterConfigs. The controller validates the proposed value against the current MV's constraints and immediately rejects the request with an INVALID_CONFIG error before any change is persisted.
  3. Metadata version upgrade: A cluster is running at IBP_4_3_IV0, where brokers do not yet report static configs to the controller, leaving it with no visibility into each broker's server.properties. Broker node-3 has log.segment.bytes=2048 in its server.properties, but the controller is unaware of this. As the administrator performs a rolling binary upgrade, each restarted broker begins sending StaticConfigs in  BrokerRegistrationRequest version 6; since log.segment.bytes now has an MV validator registered, it is included in the payload and stored by the controller in memory even while the cluster is still at  IBP_4_3_IV0. When the administrator then requests an upgrade to IBP_4_4_IV1 — the version that simultaneously introduces static config reporting and the new log.segment.bytes minimum constraint — the  controller performs a pre-flight audit using the reported static configs and finds that node-3's log.segment.bytes=2048 violates the new minimum of 3 MB. The upgrade is rejected with a descriptive error identifying the broker and the offending value. Since the cluster is still at IBP_4_3_IV0 and fully operational, the administrator can issue an AlterConfigs request to set a cluster-wide dynamic override for log.segment.bytes to a compliant value, which takes precedence over the static value in server.properties without requiring any file editing or broker restart. The administrator then retries the upgrade, the pre-flight audit finds all effective configs compliant, and the upgrade proceeds successfully.

Public Interfaces

ConfigDef

We 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.

ConfigDef
    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
   *                      that applies once that version is reached
   * @param importance    the importance of this config
   * @param documentation the documentation string for the config
   * @return This ConfigDef so you can chain calls
   */
  public ConfigDef define(String name, Type type, Object defaultValue,
                          Validator validator,
                          NavigableMap<Short, Validator> mvValidators,
                          Importance importance, String documentation)

BrokerRegistrationRequest

We propose to bump BrokerRegistrationRequest and BrokerRegistrationResponse to a new version to include a map of static configurations. This allows brokers to report their local server.properties settings to the Controller during the registration phase.

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." }
+    ]}
   ]
 }
BrokerRegistrationResponse.json
// 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." }
  ]
}

BrokerRegistrationRecord

We propose updating RegisterBrokerRecord to Version 5. This version introduces a new tagged field StaticConfigs, which is a collection of BrokerStaticConfig objects. This allows the Controller to persist the broker's reported static settings directly within the metadata log.

BrokerRegerationRecord.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": "static configs from 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." },
+    ]}
   ]
 }


Compatibility, Deprecation, and Migration Plan

  • When admin upgrade or downgrade to specific metadata version, if there is any illegal config, it will return InvalidConfigError.

Test Plan

  • Add  unit test and integration test to cover this new feature.

Rejected Alternatives

  1. Automatically adjust configuration values to meet new constraints: this approach was rejected because silently modifying configurations without administrator awareness is operationally dangerous and violates the principle of explicit configuration management.
  2. Fail immediately with ConfigException on constraint violations: this approach was rejected because it creates operational deadlocks where brokers cannot start and administrators cannot use dynamic APIs to fix the configuration.





  • No labels