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.
Current state: Draft
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).
Client configuration represents the number one pain point hindering organizations’ adoption and/or expansion of Kafka. The large number of available configuration options bewilders teams new to the ecosystem, and even accomplished teams are not immune to the occasional misconfiguration. Additionally, the teams developing and deploying client applications in an organization are often separate and intentionally siloed from the teams that manage the organization's Kafka cluster (the most extreme case being management via a hosted service). This complexity unnecessarily delays projects and contributes to operational overhead, as configuration tuning often requires specialized expertise.
As Kafka’s usage increases within organizations, centralized observability of client configuration becomes increasingly critical. This KIP takes the first step in solving this issue by introducing a standard interface by which clients can push their configuration to the cluster, thus providing the cornerstone for observability and troubleshooting.
Similar to previous initiatives (e.g. KIP-714), this KIP makes the mechanism opt-in on the broker and opt-out on the client. It is strongly advised that this KIP is implemented and made the default in all Kafka client libraries, enabling its functionality without requiring direct interaction from cluster operators. The goal will boost developer productivity, accelerate project deployment time, and reduce operational overhead, thereby lowering the barrier to adoption and expansion for Kafka.
This KIP adds a new RPC to the Kafka protocol that forms a handshake between the client and broker, named PushConfig, by which the client sends the configuration (keys and values) to the broker. Brokers interact with a new ClientConfigPolicy interface to process the above RPCs from the client.
This KIP introduces key concepts used throughout this document and in the protocol and source code.
ClientInstanceId was introduced in KIP-714 and is a UUID version 4-based value that provides a unique client ID. This ID is not a secret or token, but a value with which brokers can correlate different clients. If a client and broker supports the features from both KIP-714 and this KIP, it must use the same ClientInstanceId for both configuration and telemetry. Clients generate a new client instance ID on startup before any network activity. The ClientInstanceId is tied to the client, not its connections—a client uses the same ID for all broker connections. The ID remains valid for the client process lifetime and is stored in memory. A new ClientInstanceId is generated each time the client restarts. Brokers receive the same ClientInstanceId in all ApiVersions requests from a given client but do not coordinate, validate, or track its origin. Brokers implicitly trust the ID; it is not a secret. The value is stored in the RequestContext alongside ClientSoftwareName and ClientSoftwareVersion.
Some configuration is sensitive in that they may leak credentials, PII, or other information that could violate an organization’s data policies. For this reason, clients define a default, fixed set of configuration that will be sent to brokers. The set of configuration is carefully vetted to ensure no sensitive data is leaked. Most of the non-sensitive configuration is numeric (e.g. linger.ms), boolean (e.g. enable.auto.commit), or one of a fixed set of enums (e.g. share.acquire.mode).
In the case that the defined set of non-sensitive configuration is still too sensitive, users can override the default list with the new configuration configs.push.allowed.keys which is a comma separated list of configuration to send instead. configs.push.allowed.keys is itself not sent to the server unless it is explicitly included in the override value.
By default, the Producer sends values for the following configuration keys:
acks
batch.size
buffer.memory
client.id
compression.type
delivery.timeout.ms
enable.idempotence
linger.ms
max.in.flight.requests.per.connection
request.timeout.ms
retries
Most of these values are numeric or one of a fixed set of enums, which greatly limits potential security exposure.
By default, the Consumer sends values for the following configuration keys:
auto.offset.reset
client.id
enable.auto.commit
fetch.min.bytes
fetch.max.wait.ms
group.id
isolation.level
max.poll.interval.ms
max.poll.records
session.timeout.ms
Most of these values are numeric or one of a fixed set of enums, which greatly limits potential security exposure.
By default, the Share Consumer sends values for the following configuration keys:
group.id
share.acknowledgement.mode
share.acquire.mode
share.auto.offset.reset
share.heartbeat.interval.ms
share.isolation.level
share.record.lock.duration.ms
share.session.timeout.ms
All configuration are then subject to another test to ensure they are not considered sensitive. We define configuration as sensitive that meets any of the following criteria:
Security-related
Class or implementation names
Passwords
All configuration keys starting with sasl., security., or ssl.
All configuration keys containing .sasl., .security., or .ssl.
All configuration keys ending in .class or .classes
All configuration of type Password
In the case that a given configuration (either default or from configs.push.allowed.keys) meets the above sensitive criteria, the client logs a warning message and the configuration value for that key is not sent to the broker. In the case that all configuration is deemed sensitive, the client does not send anything configuration-related to the broker (i.e. it doesn’t send a request with an empty set of configuration).
Client libraries that use a different naming convention for its configuration should adjust their exclusion logic appropriately.
This KIP introduces a new broker-side (Java) interface named ClientConfigPolicy. This interface processes (stores, etc.) the client configuration pushed to the broker. Apache Kafka operators can install and configure a ClientConfigPolicy implementation to suit their specific needs using the configuration described later.
The following sequence diagram depicts the flow between the various entities:
The flow includes the following steps:
The client establishes a connection to the broker
The client sends an ApiVersions request
If client.configs.policy.class.name is configured, the broker advertises support for the PushConfig RPC in the ApiVersions response
The client collects the configuration values, constructs a PushConfig request, and sends it to the broker.
The broker validates the PushConfig request and invokes ClientConfigPolicy.process() to handle the configuration.
In this example, the implementation writes the configuration snapshot to external storage for observability.
ClientConfigPolicy.process() completes successfully.
The broker returns a successful PushConfig response. The client completes initialization and is ready for user API calls (e.g., send(), poll()).
ClientPushConfigDataWhen the broker receives a PushConfig request, it creates a ClientConfigData instance that includes the configuration values (ClientConfig) from the client. The broker also generates a UTC timestamp representing when the request was received and includes that in the ClientPushConfigData.
package org.apache.kafka.server.policy.clientconfig;
/**
* Enum representing the type of data of the configuration value. Types CLASS and PASSWORD
* are intentionally omitted per the configuration exclusion rules listed below.
*/
public enum ClientConfigType {
BOOLEAN(0),
STRING(1),
SHORT(2),
INT(3),
LONG(4),
DOUBLE(5),
LIST(6)
} |
package org.apache.kafka.server.policy.clientconfig;
/**
* Record containing an individual Config.
*/
public record ClientConfig(String key, Object value, ClientConfigType type, boolean isDefault) {} |
package org.apache.kafka.server.policy.clientconfig;
/**
* Record containing the PushConfig API data.
*
* <p/>
*
* The client profile configuration values come directly from
* the RPC. The broker will supply its current timestamp for the value of the same name.
*/
public record ClientConfigData(List<ClientConfig> configs, long timestamp) {} |
ClientConfigPolicyThe broker exposes a plugin interface named ClientConfigPolicy that provides the API for processing the configuration sent by the client. The interface is used on the broker to interact with the PushConfig RPC.
package org.apache.kafka.server.policy.clientconfig;
/**
* An interface for intercepting and enforcing client configuration.
*
* <p/>
*
* If <code>client.configs.policy.class.name</code> is defined, Kafka will
* create an instance of the specified class using the default constructor and
* will then pass the broker configs to its <code>configure()</code> method.
* During broker shutdown, the <code>close()</code> method will be invoked
* so that resources can be released (if necessary).
*/
@InterfaceStability.Evolving
public interface ClientConfigPolicy extends Reconfigurable, AutoCloseable {
/**
* Receive the {@link ClientPushConfigData} data for observability.
* <p/>
* <em>Note 1</em>: the implementation of this method must not block.
* <p/>
* <em>Note 2</em>: this method will <em>not</em> be invoked if the {@code Config} array
* of the {@link ClientPushConfigData} was larger than {@code client.configs.max.bytes}.
*/
void process(AuthorizableRequestContext context, ClientConfigData pushConfigData)
throws ClientConfigUnknownProfileException, ClientConfigTooLargeException, ClientConfigPolicyException;
} |
Setting client.configs.policy.class.name to null disables the feature on the broker.
Configuration name | Description | Values |
|---|---|---|
| The client configuration policy class. The class must implement the | Type: Default: |
| Maximum size for the configuration, in bytes | Type: Default: |
Applies to all of KafkaProducer, KafkaConsumer and KafkaAdmin clients, as well as Kafka Streams.
Configuration name | Description | Values |
|---|---|---|
| This configuration controls whether the client performs the configuration handshake during the establishment of a new client connection. | Type: Default: |
| Overrides the default set of configuration keys with the list from this configuration. | Type: Default: |
ApiVersionsRequest{
"apiKey": 18,
"type": "request",
"listeners": ["broker", "controller"],
"name": "ApiVersionsRequest",
// Versions 0 through 2 of ApiVersionsRequest are the same.
//
// Version 3 is the first flexible version and adds ClientSoftwareName and ClientSoftwareVersion.
//
// Version 4 fixes KAFKA-17011, which blocked SupportedFeatures.MinVersion in the response from being 0.
//
// Version 5 adds ClientInstanceId
"validVersions": "0-5",
"flexibleVersions": "3+",
"fields": [
{ "name": "ClientSoftwareName", "type": "string", "versions": "3+",
"ignorable": true, "about": "The name of the client." },
{ "name": "ClientSoftwareVersion", "type": "string", "versions": "3+",
"ignorable": true, "about": "The version of the client." },
{ "name": "ClientInstanceId", "type": "uuid", "versions": "5+",
"ignorable": true, "about": "Unique ID for this client instance."}
]
} |
This KIP adds a ClientInstanceId field to the existing ApiVersions RPC, which already includes ClientSoftwareName and ClientSoftwareVersion.
PushConfigRequest{
"apiKey": NEXT,
"type": "request",
"listeners": ["broker"],
"name": "PushConfigRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Configs", "type": "[]Config", "versions": "0+",
"about": "The client configuration entries.", "fields": [
{ "name": "ConfigKey", "type": "string", "versions": "0+",
"about": "The configuration key."},
{ "name": "ConfigValue", "type": "string", "versions": "0+",
"about": "The configuration value."},
{ "name": "ConfigType", "type": "int8", "versions": "0+",
"about": "ClientConfigType of the ConfigValue field."},
{ "name": "IsDefault", "type": "bool", "versions": "0+",
"about": "Boolean where true means the configuration value wasn't changed by the user."},
]}
]
} |
After receiving ApiVersionsResponse, the client collects the configuration values and sends them in a PushConfigRequest. The client typically sends this request once during bootstrap, before invoking client APIs. Retries use retry.backoff.ms, retry.backoff.max.ms, and default.api.timeout.ms, similar to ApiVersions.
The ConfigType field is an integer that maps to the ClientConfigType enum, defined above.
PushConfigResponse{
"apiKey": NEXT,
"type": "response",
"name": "PushConfigResponse",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{
"name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The 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": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." }
]
} |
PushConfigResponse has a field named ErrorMessage that contains a brief reason for the error.
Clients retry on network and retriable errors. If throttled (ThrottleTimeMs > 0), the client waits before retrying. Fatal errors (e.g., UNSUPPORTED_VERSION, authentication failures) are not retried and should throw runtime exceptions.
Configuration support is performed on a best-effort basis. Failure to send the configuration should not prevent the client from functioning.
The following errors are new for this RPC:
Error Code | Description | Client Action |
|---|---|---|
| Client sent a request in which the | Log the error in |
| The | Log the error in |
Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.
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?
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.