...

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

...

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 RPC from the client.

Concepts

This KIP section introduces key concepts used throughout this document and in the protocol and source codeKIP.

Client Instance ID

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.

...

Compatibility, Deprecation, and Migration Plan

Impact on Existing User

Broker

• No policy configured: Feature is effectively disabled

  • No new RPCs are advertised in ApiVersions

  • If client sends requests, broker throws an error

  • No performance or behavioral impact

• Policy configured: Feature is enabled

  • New RPCs are advertised in ApiVersions

  • Broker receives and processes config pushes from supporting clients

  • Older clients (no support) are unaffected

  • Minimal resource overhead

• Upgrade path: Rolling upgrade safe

  • Old brokers: don't advertise config push APIs, clients skip handshake

  • New brokers: advertise APIs, clients perform handshake if enabled

  • No cross-version issues

Client

• Older clients (no support): No impact

  • Don't check for config push APIs

  • Behavior identical to pre-KIP

• Newer clients with feature disabled: No impact

  • enable.configs.push=false

  • Handshake skipped

  • Behaviorally identical to older clients

• Newer clients with feature enabled (default): Minor impact

  • Additional RTT during connection setup (PushConfig)

  • Estimated 10-50ms added latency to first user request (depends on network RTT)

  • One-time cost per client instance lifetime

Migration

This is a purely additive feature and requires no migration.

• Before KIP: No client configuration visibility

• After KIP: Incremental visibility as clients upgrade

• Breaking changes: n/a

• Deprecation: n/a

Test Plan

Integration Tests

Happy Path

1. Test complete handshake with success

Client Startup

1. Test producer, consumer, admin, and Kafka Streams client with enable.configs.push set to true performs handshake

2. Test producer, consumer, admin, and Kafka Streams client with enable.configs.push set to false skip handshake entirely

3. Test Kafka Streams does not perform separate handshake for embedded producer/consumer/admin clients

Broker Configuration

1. Test broker with client.configs.policy.class.name set advertises APIs in ApiVersions

2. Test broker without policy (null) does not advertise config push APIs

3. Test broker with policy invokes process() on successful PushConfig

Mixed Broker Versions (Rolling Upgrade)

1. Test old brokers (no config push support) don't advertise APIs

2. Test new brokers advertise APIs

3. Test clients detect support via ApiVersions and only handshake with new brokers

4. Test clients work correctly when connecting to mix of old and new brokers

Retries

1. Test client retries PushConfig on UNKNOWN_CONFIG_PROFILE

2. Test client does not retry on CONFIG_TOO_LARGE or INVALID_CONFIG

3. Test exponential backoff is applied correctly

Timeout Handling

1. Test handshake respects default.api.timeout.ms

2. Test client continues if handshake times out (best-effort feature)

3. Test timeout does not block subsequent operations

Throttling

1. Test client waits for ThrottleTimeMs before retrying if throttled

System Tests

Multiple Client Types

1. Test Java KafkaProducer, KafkaConsumer, AdminClient, and KafkaStreams application and verify each client type sends appropriate configs based on the type of client

Large Payloads

1. Test config payload near client.configs.max.bytes limit

2. Test config payload exceeding client.configs.max.bytes returns CONFIG_TOO_LARGE

Policy

1. Test custom ClientConfigPolicy rejects configs via InvalidConfigException

2. Test client receives INVALID_CONFIG error with an appropriate message

Rejected Alternatives

Exclusion of Default Values

The number of configuration entries is getting larger with each release, the vast majority of which use default settings. The question arises: how should we handle configuration entries that use a default value? Here are some options for the client:

1. Send all configuration entries to the cluster, including those with default values.

2. Omit any configuration entries that use their respective default values.

3. Send all configuration entries, but omit the value value and demarcate those that use default values.

It’s redundant to send configuration with known default values. Preparing and sending the name, type, and value for scores of configuration could add up to a couple of KB in network transit. Additionally, that then means that the server side node that receives the handshake request then has to handle requests that consist mostly of entries with default values. It’s easy to argue that the defaults are superfluous and not include them.

Whether or not to include configuration entries with default values somewhat depends on what the ClientConfigPolicy implementation plans to do with those entries. Also, keep in mind that the server receiving the handshakes may service many different clients from different languages and versions. Even though the client knows when the configuration entry’s value is the default value, the broker handling the handshake may have no idea that, for example, topic.compression.level=low from the 2.5.2 version of the Visual Basic Kafka client is the default value for that client. Requiring each implementor of ClientConfigPolicy to maintain a listing of all the default values across all available Kafka clients and their respective versions seems like overkill.

The stance of this KIP is that there is no need to exclude default configuration, based on the following:

1. The configuration handshake only occurs once during the lifetime of a client.

2. The configuration handshake request size is negligible compared with the amount of network traffic over the course of the lifetime of a client.

3. The server node handling the incoming configuration handshake can drop any entries that come in with a default value.

Requiring that broker know a priori the default values (to fill in the missing information) is a maintenance problem.

Including Configuration Storage Detail

Storage and retention of the configuration data is outside the scope of this KIP. The ClientConfigPolicy implementation is responsible for managing the storage, if any, of the configuration payload once the broker invokes it.