...

The following sequence diagram depicts the flow between the various entities:

Image Added

The flow includes the following steps:

...

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.

...

Proposed Changes

Disabling the Feature

This feature can be disabled on the broker and the client. For the broker, remove client.configs.policy.class.name or set its value to null. When this configuration is missing, the ApiVersions response will not include support for the feature, so the client doesn’t send configuration. For the client, set enable.configs.push to false, in which case the client skips the entire configuration handshake.

Broker Behavior

If a broker is configured with client.configs.policy.class.name, the ApiVersions response advertises support for the PushConfig RPC. Whenever a client sends a PushConfig request, the broker calls the policy with the client configuration for observability.

Configuration Payload Size Enforcement

As described below, client implementations should not attempt to send a payload that is too large in the first place. But as a backup means of preventing the client from sending too much data, the broker checks the new configuration client.configs.max.bytes prior to invoking the policy. If the size of the PushConfig request exceeds client.configs.max.bytes, the broker returns the CONFIG_TOO_LARGE error to the client.

After analyzing the different Java clients’ configuration, a default of 10 KB for client.configs.max.bytes provides more than sufficient capacity:

A default 10 KB limit provides ~5x headroom for the largest expected use case, while preventing abuse from malicious or errant clients.

Excluding Sensitive Configuration

As explained in the concepts section, the ClientConfigPolicy implementation may also provide logic to ensure that clients do not send sensitive configuration. Clients across the Apache Kafka ecosystem do not have a consistent naming convention. As a result, brokers cannot determine sensitivity based on the configuration key name and rely on the incoming ConfigType field. When the ClientConfigPolicy detects sensitive configuration, it includes a description of the violation in the RPC response.

Metrics

This KIP does not introduce any new broker metrics.

Client Behavior

Handshake

A client that supports this configuration interface will identify a node that supports the API using ApiVersions. The client performs a handshake by collecting the values for its configuration issuing a PushConfig RPC to submit the configuration to the broker node. The client sends the RPC after authentication (if any) and before the client starts to use the connection for requests. Similar to the ApiVersions handshake, the PushConfig RPC specifies a fixed timeout of default.timeout.ms. If the RPC exhausts its retries, the client logs the error, but continues execution.

Here’s the sequence:

1. Client connects to broker

2. Send ApiVersions request (internal, automatic)

3. Receive ApiVersions response to determine which features broker supports

4. If enable.configs.push is set and configuration push is supported by the broker

   1. Collect requested client configuration values

   2. Send PushConfig with configuration

   3. Receive PushConfig response

5. User requests can now be sent

The client chooses a randomly selected node for its configuration handshake, in the same way as GetTelemetrySubscriptions. The client configuration is only sent once, not for each broker. The “handshake” is performed on a per-client basis, not for each connection. As connections are closed due to disconnects or aging out, no new configuration handshake is performed. This operation is executed once for each distinct client instance.

The handshake is performed on a best-effort basis. Network or other transient errors that occur when transmitting the configuration data must not prevent the client from functioning.

Blocking Behavior

From the user’s perspective, the client blocks when a client API is invoked until the handshake completes.

For example:

// KafkaProducer constructor returns immediately
KafkaProducer<String, String> producer = new KafkaProducer<>(props);   // ✓ Non-blocking

// First send() blocks until connection is ready (including configuration handshake)
producer.send(record);                                                 // ← Blocks up to max.block.ms waiting for READY state

Excluding Sensitive Configuration

The configuration entries sent in the PushConfig request should exclude any sensitive information. Which configuration keys are considered sensitive is determined by the client library.

Including Configuration Data Types

Depending on the server-side implementation, preserving the data types enables more compact storage, reduces redundant conversion, reduce cognitive overhead in downstream use, and eliminate invalid data (e.g. storing a value of “hi!” in a boolean). It’s also important because the server is not aware of all the different clients and their respective configuration, so providing the name, type, and value, though at times redundant, allows for wider client compatibility.

Metrics

This KIP does not introduce any new client metrics.

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

...

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.