...

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: DraftUnder discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

...

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 configsconfig.push.allowed.keys which is a comma separated list of configuration to send instead. configsconfig.push.allowed.keys is itself not sent to the server unless it is explicitly included in the override value.

...

• 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

• request.timeout.ms

• session.timeout.ms

Most of these values are numeric or one of a fixed set of enums, which greatly limits potential security exposure.

Default Configuration for Apache Kafka Java

...

Admin Client

By default, the Share Consumer Admin client sends values for the following configuration keys:

• groupclient.id

• sharerequest.acknowledgementtimeout.modems

• retries

Except for client.id these values are numeric, which greatly limits potential security exposure.

Default Configuration for Apache Kafka Java Share Consumer

By default, the Share Consumer sends values for the following configuration keys:

• client.id
• fetch.max.wait.ms
• fetch.min.bytes
• group.id
• max.poll.interval.ms
• max.poll.records
• share.acknowledgement.mode
• share.acquire.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

• (in the Java reference implementation, where ConfigDef.Type equals CLASS)

• Passwords (in the Java reference implementation, where ConfigDef.Type equals PASSWORD)

• Custom configuration (e.g. used by application, SerDes, etc.) since Kafka cannot determine if these could contain sensitive information

• All configuration keys starting with sasl., All configuration keys starting with sasl., security., or ssl.

• All configuration keys containing .sasl., .security., or .ssl.

• All configuration keys ending in .class or .classesAll configuration of type Password

In the case that a given configuration (either default or from configsconfig.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).

...

1. The client establishes a connection to the broker

2. The client sends an ApiVersions request

3. If client.configsconfig.policy.class.name is configured, the broker advertises support for the PushConfig RPC in the ApiVersions response

4. The client collects the configuration values, constructs a PushConfig request, and sends it to the broker.

5. The broker validates the PushConfig request and invokes ClientConfigPolicy.process() to handle the configuration.

   1. In this example, the implementation writes the configuration snapshot to external storage for observability.

6. ClientConfigPolicy.process() completes successfully.

7. The broker returns a successful PushConfig response. The client completes initialization and is ready for user API calls (e.g., send(), poll()).

...

package org.apache.kafka.server.policy.clientconfig;
/**
 * EnumRecord representingcontaining thean typeindividual of data of the configuration value. Types CLASS and PASSWORDConfig.
 */
public record ClientConfig(String key, Object value, ConfigDef.Type type, boolean isDefault) {}

package org.apache.kafka.server.policy.clientconfig;
/**
 * areRecord intentionally omittedcontaining per the configurationPushConfig exclusion rules listed belowAPI data.
 */
public enum ClientConfigType {* <p/>
 *
  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* The client configuration values come directly from the RPC. The broker
 * will supply its current timestamp for the value of the same name.
 */
public record ClientConfigClientConfigData(StringList<ClientConfig> keyconfigs, Object value, ClientConfigType type, boolean isDefaultlong timestamp) {}

ClientConfigPolicy

The 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;

/**
 * RecordAn containinginterface thefor PushConfigintercepting API dataconfiguration sent by the client.
 *
 * <p/>
 *
 * The client profile configuration values come directly fromIf <code>client.config.policy.class.name</code> is defined, Kafka will
 * thecreate RPC.an Theinstance brokerof willthe supplyspecified its current timestamp forclass using the valuedefault of the same name.constructor and
 */
public recordwill ClientConfigData(List<ClientConfig> configs, long timestamp) {}

ClientConfigPolicy

The 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 {
  
  /**
   * Receivethen 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} data for observabilitywas larger than {@code client.config.max.bytes}.
   * <p/>
  void * <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;process(AuthorizableRequestContext context, ClientConfigData pushConfigData);
}

ClientConfigTooLargeException

The broker checks the size of the configuration data sent via the PushConfig RPC. If it is larger than 

package org.apache.kafka.common.errors;

/**
 * This exception indicates that the size of the client configuration data exceeded the
 * broker's client.config.max.bytes configuration.
 */
public class ClientConfigTooLargeException extends ApiException {

    public ClientConfigTooLargeException(String message) {
        super(message);
    }
}

Configuration

Broker

Setting client.configsconfig.policy.class.name to null disables the feature on the broker.

Client

Applies to all of KafkaProducer, KafkaConsumer and KafkaAdmin clients, as well as Kafka Streams.

...

This KIP adds a ClientInstanceId field to the existing ApiVersions RPC, which already includes ClientSoftwareName and ClientSoftwareVersion.

PushConfigRequest

Errors

The following error is new for this RPC:

PushConfigRequest

{
  "apiKey": NEXT,
  

{
  "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": "ClientConfigTypeConfigDef.Type 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 abovethe ConfigDef.Type enum.

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

...

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

...

CONFIG_TOO_LARGE

...

Client sent a request in which the PushConfig request was too large (see client.configs.max.bytes)

Proposed Changes

Disabling the Feature

This feature can be disabled on the broker and the client. For the broker, remove client.config.

Log the error in ErrorMessage then continue

...

INVALID_CONFIG

...

The ClientConfigPolicy implementation rejected as invalid the data sent by the client in the Configs field. For example, this could occur if the client sends the wrong type for a configuration, an un-parseable value, etc. The broker sets the error code to INVALID_CONFIG, and the ErrorMessage will contain details for the failed entry.

...

Log the error in ErrorMessage then continue

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.configsconfig.push to false, in which case the client skips the entire configuration handshake.

...

If a broker is configured with client.configsconfig.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.

...

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.configsconfig.max.bytes prior to invoking the policy. If the size of the PushConfig request exceeds client.configsconfig.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.configsconfig.max.bytes provides more than sufficient capacity:

...

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

...

field. When the ClientConfigPolicy detects sensitive configuration, it includes a description of the violation in the RPC response.

Metrics

The following new broker metrics are added:

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.

...

1. Client connects to broker

2. Send ApiVersions request (internal, automatic)

3. Receive ApiVersions response to determine which features broker supports

4. If enable.configsconfig.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

...

• 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.configsconfig.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

...

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

2. Test producer, consumer, admin, and Kafka Streams client with enable.configsconfig.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.configsconfig.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

...

1. Test client retries PushConfig on UNKNOWN_CONFIG_PROFILEretriable errors

2. Test client does not retry on CONFIG_TOO_LARGE or INVALID_CONFIG

3. Test exponential backoff is applied correctly

...

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

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

...