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: [One of "Under Discussion", "Accepted", "Rejected"]

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

Today, Apache Kafka lacks a mechanism to enforce configuration constraints that are tied to specific metadata versions. This creates two problems:

Unsafe metadata version upgrades: Administrators may not be aware that their configuration changes were ignored, leading to unexpected cluster behavior
No runtime enforcement of version-specific constraints: Even after upgrading, administrators can use AlterConfigs or IncrementalAlterConfigs to set config values that violate constraints introduced by the current metadata version. There is no guardrail beyond the static ConfigDef validators, which are version-agnostic.

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:

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.
Metadata-Version-Aware Validators::Configuration definitions can now specify validators that apply at specific metadata version feature levels. The ConfigDef.define() method accepts a mvValidators parameter — a map from feature level (short) to ConfigDef.Validator instances.
1. When a broker register to cluster, if it has illegal config, the controller should reject its join.
2. When alterConfig and incrementalAlterConfig, metadata-version validator should be called to check new config is satisfy to metadata version requirements.
3. When admin.upgrade and downgrade, all configurations (static config and dynamic config) should be validated it is satisfy to metadata version requirements.

Public Interfaces

ConfigDef

We add new field mvValidator to make configuration can be validated according to metadata version.

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

BrokerRegistrationRequest

Bump BrokerRegistrationRequest and BrokerRegistrationResponse to add static configuration fields.

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

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

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 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." },
+      { "name": "IsSensitive", "type": "bool", "versions": "5+", "default": "false",
+        "about": "True if this configuration is sensitive." }
+    ]}
   ]
 }

Compatibility, Deprecation, and Migration Plan

Add unit test and integration test to cover new feature.
When admin upgrade or downgrade to specific metadata version, if there is any illegal config, it will return InvalidConfigError

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.