...

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

• groupclient.id
• sharefetch.max.acknowledgementwait.modemsshare
• fetch.acquiremin.modebytes

• share.auto.offset.reset

• group.id
• max.pollshare.heartbeat.interval.msshare
• max.isolationpoll.levelrecords
• share.record.lock.duration.msacknowledgement.mode
• share.sessionacquire.timeout.msmode

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

...

package org.apache.kafka.server.policy.clientconfig;
/**
 * EnumRecord containing representingan theindividual 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)
}

Config.
 */
public record ClientConfig(String key, Object value, ConfigDef.Type type, boolean isDefault) {}

package org.apache.kafka.server.policy.clientconfig;
/**
 * Record containing the PushConfig API data.
 *
 * <p/>
 *
 * The client configuration values come directly from the RPC. The broker
 * will supply its current timestamp for the value of the same namepackage org.apache.kafka.server.policy.clientconfig;
/**
 * Record containing an individual Config.
 */
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;

/**
 * An Recordinterface for containingintercepting theand PushConfigenforcing APIclient dataconfiguration.
 *
 * <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) {}

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.

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);
}

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.configs.max.bytes configuration.
 */
public class ClientConfigTooLargeException extends ApiException {

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

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

Configuration

Broker

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

...

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

...

CONFIG_TOO_LARGE

...

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

runtime exceptions.

Configuration support is performed on a best-effort basis. Failure to send the configuration should not prevent the client from functioning.

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

...

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

...