DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
- 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.
- When alterConfig and incrementalAlterConfig, metadata-version validator should be called to check new config is satisfy to metadata version requirements.
- When admin.upgrade and downgrade, all configurations (static config and dynamic config) should be validated it is satisfy to metadata version requirements.
Public Interfaces
Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.
A public interface is any change to the following:
Binary log format
The network protocol and api behavior
Any class in the public packages under clientsConfiguration, especially client configuration
org/apache/kafka/common/serialization
org/apache/kafka/common
org/apache/kafka/common/errors
org/apache/kafka/clients/producer
org/apache/kafka/clients/consumer (eventually, once stable)
Monitoring
Command line tools and arguments
- Anything else that will likely break existing users in some way when they upgrade
ConfigDef
We add new field mvValidator to make configuration can be validated according to metadata version.
| Code Block | ||||
|---|---|---|---|---|
| ||||
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.
| Code Block | ||||
|---|---|---|---|---|
| ||||
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 | ||||||
|---|---|---|---|---|---|---|
| ||||||
// 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." } ] } |
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?Add unit test and integration test to cover new feature.
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?
...