You are viewing an old version of this page. View the current version.

Compare with Current View Page History

Version 1 Next »

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

Kafka's controller currently has no visibility into each broker's static configuration. When a broker registers with the controller via BrokerRegistrationRequest, it reports its supported feature versions, listeners, rack, and log directories, but not its configuration 
values (i.e., the settings defined in server.properties). This creates a fundamental gap that blocks two categories of improvements:

Aligning incrementalAlterConfigs validation (KIP-1256)

The controller and broker currently behave differently when processing Admin.incrementalAlterConfigs requests. One key inconsistency is that brokers immediately reject invalid dynamic configuration values, while the controller silently drops them. To replicate broker-side validation logic, the controller needs each broker's static configuration as context — certain dynamic configuration constraints depend on the static values present on that broker. Without this information, the controller cannot perform faithful broker-equivalent validation.


Enforcing configuration constraints during MetadataVersion upgrades (KIP-1294)

As the cluster's `MetadataVersion` advances, new constraints on configuration values may come into effect. For example, a future MetadataVersion might raise the minimum value of log.segment.bytes from 1 byte to 3 MB. Without knowledge of each broker's static configuration, the controller cannot proactively verify that all brokers satisfy the constraints of the target version before allowing an upgrade
to proceed. The incompatibility can only be discovered after the fact, when a broker fails on restart.

This KIP addresses the root cause shared by both problems: the controller lacks broker static configuration data. We propose that brokers include their non-sensitive static configurations in `BrokerRegistrationRequest`, and that the controller persists this information in RegisterBrokerRecord so it survives controller failovers. This KIP intentionally contains no validation logic — it provides only the infrastructure that KIP-1256 and KIP-1294 build upon.

Public Interfaces

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

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


Proposed Changes

Broker side

sendBrokerRegistration() is updated to collect non-sensitive static configuration entries from `KafkaConfig` and include them in the request when the negotiated protocol version is 6 or above.

Sensitive configurations are excluded. All other broker-scoped static configurations are included, regardless of whether they are also dynamically changeable. This keeps the broker-side logic simple — a straightforward iteration over `ConfigDef` keys with a sensitivity filter.

Controller side

When receiving a BrokerRegistrationRequest v6 and the current MetadataVersion supports static configs, the controller copies the StaticConfigs field into the RegisterBrokerRecord.

The in-memory BrokerRegistration object is extended with a Map<String, String> field to hold the static configs. This field is populated during

ClusterControlManager.replay(RegisterBrokerRecord) and is accessible to higher-level logic (KIP-1256, KIP-1294) without requiring additional requests to the broker.

Compatibility, Deprecation, and Migration Plan

  • What impact (if any) will there be on existing users?
  • If we are changing behavior how will we phase out the older behavior?
  • If we need special migration tools, describe them here.
  • When will we remove the existing behavior?

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.

  • No labels