Status

Current state: "Under Discussion"

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

JIRA: KAFKA-6240

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Kafka brokers are configured using properties (typically provided in a file named server.properties) that cannot be updated without restarting the broker. Broker configuration options can be described using the new AdminClient, but the configuration cannot currently be altered.

There are several use cases where dynamic updates of broker configs will help to avoid time-consuming restart of brokers. For example:

1. Update short-lived SSL keystores on the broker
2. Performance tuning based on metrics (e.g. increase network/IO threads)
3. Add, remove or reconfigure metrics reporters
4. Update configuration of all topics consistently across the whole cluster (e.g. unclean.leader.election.enable)
5. Update log cleaner configuration for tuning
6. Update listener/security configuration

Goals:

1. Use existing dynamic configuration mechanism to make commonly updated broker configs dynamic so that they can be updated without restarting the broker.
2. Support dynamic default configs that are picked up by all brokers in the cluster, to enable consistent configuration across all brokers (e.g. unclean.leader.election.enable)
3. Support individual broker config overrides for all dynamic configs to enable testing on one broker before the changes are picked up by all brokers.
4. Enable broker configs to be described and altered using AdminClient.
5. Continue to support simple property file based configuration so that brokers can be easily started up without any additional configuration steps.

In the first phase of the implementation, commonly updated broker properties will be made dynamic based on known use cases. The goal will be to make more configs dynamic later if required.

Public Interfaces

Dynamic configuration

Dynamic configs for the entity type brokers will be added for a limited set of configuration options by extending the existing dynamic replication quota config for brokers. The corresponding static config in server.properties will be continue to be supported to enable quick start of Kafka brokers with a properties file. Where applicable, default cluster-wide configs can be configured to allow consistent values to be used across all brokers (stored in ZK at /config/brokers/<default>). For per-broker configs, broker-specific config overrides can be configured (stored in ZK at /config/brokers/id).

The precedence used for configs will be:

1. Dynamic broker-specific config from /config/brokers/id
2. Dynamic cluster-wide default config from /config/brokers/<default>
3. Static config from server.properties

Individual broker config overrides will be provided for all the dynamic configs to enable testing. AdminClient will return all configured configs and defaults in the order of precedence in DescribeConfigsResponse if synonyms are requested. For configs that we want to keep consistent across all brokers, kafka-configs.sh and AdminClient will allow updates of cluster-wide defaults, but it will still be possible to configure per-broker overrides for testing. Brokers will handle all configs consistently with the precedence specified above.

Dynamic broker configs will be stored in ZooKeeper in JSON format along with the existing replication quota config for entity type brokers. With the current dynamic configuration implementation, brokers watch configuration update node in ZooKeeper and invoke config handlers when update notifications are received. The config handler for brokers will be updated to handle KafkaConfig changes as well.

The existing ACL for AlterConfigs operation on Cluster resource will be applied for broker config updates. As with topic configuration updates, alter.config.policy.class.name can be configured to specify a custom  AlterConfigsPolicy.

New Broker Configuration Option

sasl.jaas.config will be supported for brokers to enable dynamic SASL configuration of brokers. The property will use the same format as clients and may specify one or more login context entries (one for each supported SASL mechanism). The property may be preceded by listener name if multiple listeners are configured to use SASL.

Format: One or more login context entries using the same format JAAS configuration:

<LoginModuleClass> <ControlFlag> *(<OptionName>=<OptionValue>);

ControlFlag = required|requisite|sufficient|optional

Example:

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required user_alice=”alice-secret”; \
org.apache.kafka.common.security.scram.ScramLoginModule required;

Securing passwords in ZooKeeper

A new broker configuration will be added to specify a secret key that is used to encrypt passwords stored in ZooKeeper. The SecretKeyFactory and Cipher algorithms as well as the iteration count used will be also be made configurable. The actual algorithms, salt and iteration count will be stored along with the encrypted password to ensure that the password can be decrypted even if the encryption parameters are altered (though these configs are not being made dynamic at the moment).

• Name: config.secret  Type: Password
• Name: config.secret.keyfactory.algorithm  Type: String Default: PBKDF2WithHmacSHA512 if available, otherwise PBKDF2WithHmacSHA1 (e.g. Java7)
• Name: config.secret.cipher.algorithm  Type: String  Default: AES/CBC/PKCS5Padding 
  
• Name: config.secret.iterations  Type: Integer Default: 2048

The secret will not be dynamically configurable and hence will never be stored in ZooKeeper. All the dynamic password configs are per-broker configs and hence there is no requirement to maintain the same secret across all brokers.

Broker configuration in ZooKeeper will be protected using ACLs and will no longer be world-readable by default. It is expected that secure deployments of Kafka will also use network segmentation to limit ZooKeeper access.

Protocol Changes

A new option include_synonyms will be added to DescribeConfigsRequest to return all the configs which may be used as the value of the specified config if the config was removed. For example, flush.ms config for a topic will return the broker config log.flush.interval.ms as a synonym if include_synonyms=true. DescribeConfigsResponse will be return all synonyms in the order of precedence. This is particularly useful to obtain the default value that the config will revert to if a config override is removed. As dynamic broker configs are being added at per-broker and cluster-default levels, include_synonyms can be useful to list all the configured values and the precedence used to obtain the currently configured value.

DescribeConfigsRequest version will be bumped up to 1.

DescribeConfigs Request (Version: 1) => [resource [config_name]] include_synonyms
  resource => resource_type resource_name
    resource_type => INT8
    resource_name => STRING
  config_name => STRING
  include_synonyms => BOOLEAN

DescribeConfigs Response (Version: 1) => throttle_time_ms [entities]
  throttle_time_ms => INT32
  entities => error_code error_message resource [configs]
    error_code => INT16
    error_message => STRING
    resource => resource_type resource_name
      resource_type => INT8
      resource_name => STRING
    configs => [config_entry [synonym]]    <= Added [synonym]
    config_entry =>
      config_name => STRING
      config_value => NULLABLE_STRING
      read_only => BOOLEAN
      config_source => INT8                <= Replaced boolean is_default with more generic config_source (see below for values)
      is_sensitive => BOOLEAN
    synonym =>                             <= NEW
      config_name => STRING
      config_value => NULLABLE_STRING
      config_source => INT8                <= may be one of (TOPIC|DYNAMIC_BROKER|DYNAMIC_DEFAULT_BROKER|STATIC_BROKER|DEFAULT)

Public Interface Changes

A new interface Reconfigurable will be added to notify reconfigurable objects of configuration changes. For example, metrics reporters that support reconfiguration can implement the interface Reconfigurable to enable reconfiguration without broker restart. The interface will also be implemented by all internal classes which support reconfiguration (e.g. ChannelBuilder)

package org.apache.kafka.common;

import java.util.Map;
import java.util.Set;

/**
 * Interface for reconfigurable classes that support dynamic configuration.
 */
public interface Reconfigurable extends Configurable {

    /**
     * Returns the names of configs that may be reconfigured.
     */
    Set<String> reconfigurableConfigs();

    /**
     * Validates the provided configuration. The provided map contains
     * all configs including any reconfigurable configs that may be different
     * from the initial configuration.
     */
    boolean validateReconfiguration(Map<String, ?> configs);

   /**
     * Reconfigures this instance with the given key-value pairs. The provided
     * map contains all configs including any reconfigurable configs that
     * may have changed since the object was initially configured using
     * {@link Configurable#configure(Map)}.
     */
    void reconfigure(Map<String, ?> configs);

}

The classes DescribeConfigsOptions and DescribeConfigsResult used by AdminClient will be updated to include config synonyms in the result.

/**
 * Return true if synonym configs should be returned in the response.
 */
public boolean includeSynonyms() {
    return includeSynonyms;
}

/**
 * Set to true if synonym configs should be returned in the response.
 */
public DescribeConfigsOptions includeSynonyms(boolean includeSynonyms) {
    this.includeSynonyms = includeSynonyms;
    return this;
}

/**
 * Returns all config values that may be used as the value of this config along with their source,
 * in the order of precedence. The list starts with the value returned in this ConfigEntry.
 * The list is empty if synonyms were not requested using {@link DescribeConfigsOptions#includeSynonyms(boolean)}. 
 */
public List<ConfigSynonym> synonyms() {
    return  synonyms;
}

public static class ConfigSynonym {

    /**
     * Returns the name of this configuration.
     */
    public String name() {
        return name;
    }

    /**
     * Returns the value of this configuration, which may be null if the configuration is sensitive.
     */
    public String value() {
        return value;
    }

    /**
     * Returns the source of this configuration.
     */
    public ConfigSource source() {
        return source;
    }
}

public enum ConfigSource {
    TOPIC_CONFIG,                   // dynamic topic config that is configured for a specific topic
    DYNAMIC_BROKER_CONFIG,          // dynamic broker config that is configured for a specific broker
    DYNAMIC_DEFAULT_BROKER_CONFIG,  // dynamic broker config that is configured as default for all brokers in the cluster
    STATIC_BROKER_CONFIG,           // static broker config provided as broker properties at start up (e.g. server.properties file)
    DEFAULT_CONFIG                  // built-in default configuration for configs that have a default value
}

Tools

kafka-configs.sh will be updated to configure defaults and overrides for dynamic configuration options for the entity type brokers.

Example:

Add/update broker-specific config for broker with broker id 0:

bin/kafka-configs  --zookeeper localhost:2181 --alter --add-config 'ssl.keystore.location=keystore1.jks,ssl.keystore.password=password1' --entity-name 0 --entity-type brokers

Add/update cluster-wide default config:

bin/kafka-configs  --zookeeper localhost:2181 --alter --add-config 'unclean.leader.election.enable=false' --entity-default --entity-type brokers

Delete broker-specific config (the actual config will revert to its default value, see precedence of configs):

bin/kafka-configs.sh --zookeeper localhost --alter --delete-config unclean.leader.election.enable --entity-type brokers --entity-name 0

Proposed Changes

SSL keystore

Use case: Support short-lived SSL certificates for increased security

Config scope: Broker (/config/brokers/id)

Config options:

• ssl.keystore.type
• ssl.keystore.location
• ssl.keystore.password
• ssl.key.password

Dynamic update changes:

• Keystore will be updated by reconfiguring the channel builder to create a new SslFactory. Existing connections will not be affected, new connections will use the new keystore.

Threads

Usecase:: Increase or decrease threads based on traffic pattern. This is useful for performance tuning based on metrics at runtime.

Config scope: Default for whole cluster (/configs/brokers/<default>)

Config options:

• num.network.threads
• num.io.threads
• num.replica.fetchers
• num.recovery.threads.per.data.dir
• background.threads

Dynamic update changes:

• Thread pools that don’t require thread affinity will be updated to use resizable thread pools:
  • num.io.threads
  • num.recovery.threads.per.data.dir
  • background.threads
• num.network.threads: To increase threads, existing connections will continue to be processed in their processor threads and new connections will be processed on new threads. Allocation of processors for new connections will take connection count into account (instead of the current round-robin allocation) to ensure that connections are balanced across processors. When thread count is decreased, threads with the smallest number of active connections will be terminated after closing any connections being processed on those threads. New connections will be processed on remaining threads.
• num.replica.fetchers: Affinity of partitions to threads will be preserved for ordering.

Metrics reporters and their custom configs

Use case: Add new metrics reporter or reconfigure existing metrics reporter

Config scope: Default for whole cluster (/configs/brokers/<default>)

Config options:

• metric.reporters

Dynamic update changes:

• Addition and removal of metrics reporters will be supported. Since metrics reporters may use custom configs, dynamic configuration options for brokers may specify any custom configuration. AdminClient and kafka-configs.sh will be updated to accept custom configs.
  
• Metrics reporters that support reconfiguration may implement the interface Reconfigurable. If any of the configs specified in Reconfigurable#reconfigurableConfigs is updated dynamically, the reporter instance is reconfigured using Reconfigurable#reconfigure(Map).

Log Cleaner configs

Use case: Add more log cleaner threads or increase buffer size for tuning

Config scope: Default for whole cluster (/configs/brokers/<default>)

Config options:

• log.cleaner.threads
• log.cleaner.io.max.bytes.per.second
• log.cleaner.dedupe.buffer.size
• log.cleaner.io.buffer.size
• log.cleaner.io.buffer.load.factor
• log.cleaner.backoff.ms

Dynamic update changes:

• Config changes will be applied to existing log cleaner threads from the next iteration. Threads may be added or removed. If any of the old log cleaner threads had died due to errors, new threads will be created to replace those too so that the number of threads after an update reflects the configured log.cleaner.threads.

Default Topic configs

Use case: Even though topic configs can be set per-topic dynamically, it is convenient to configure properties like unclean.leader.election.enable consistently across the whole cluster.

Config scope: Default for whole cluster (/configs/brokers/<default>)

Config options (and the equivalent topic configs):

• log.segment.bytes (segment.bytes)
• log.roll.ms, log.roll.hours (segment.ms)
• log.roll.jitter.ms, log.roll.jitter.hours (segment.jitter.ms)
• log.index.size.max.bytes (segment.index.bytes)
• log.flush.interval.messages (flush.messages)
• log.flush.interval.ms (flush.ms)
• log.retention.bytes (retention.bytes)
• log.retention.ms, log.retention.minutes, log.retention.hours (retention.ms)
• log.index.interval.bytes (index.interval.bytes)
• log.cleaner.delete.retention.ms (delete.retention.ms)
• log.cleaner.min.compaction.lag.ms (min.compaction.lag.ms)
• log.cleaner.min.cleanable.ratio (min.cleanable.dirty.ratio)
• log.cleanup.policy (cleanup.policy)
• log.segment.delete.delay.ms (file.delete.delay.ms)
• unclean.leader.election.enable (unclean.leader.election.enable)
• min.insync.replicas (min.insync.replicas)
• max.message.bytes (max.message.bytes)
• compression.type (compression.type)
• log.preallocate (preallocate)
• log.message.timestamp.type (message.timestamp.type)
• log.message.timestamp.difference.max.ms (message.timestamp.difference.max.ms)

Dynamic update changes:

• These properties will be supported as dynamic defaults that are used by all brokers since we don’t want to have different default values on different brokers. Configuration precedence for these properties will be:
• Dynamic topic level (/configs/topics/topicName)
• Dynamic broker level (/configs/brokers/id)
• Dynamic default for all brokers in cluster (/configs/brokers/<default>)
• Static server.properties

Listeners and Security configs

Use cases:

• Add a new listener, e.g. to add a new security protocol or make inconsistent changes to existing security protocol (e.g change CA for SSL)
• Configure SSL/SASL configs for new security protocol
• Remove an old listener after new listener has been added to all brokers (e.g. after changing CA for SSL)
• Update advertised listeners, e.g. to set IP address after broker has been started since in some environments broker address is known only after broker has started up

Config options:

Listener Configs

• listeners
• advertised.listeners
• listener.security.protocol.map
• inter.broker.listener.name
• security.inter.broker.protocol
• sasl.mechanism.inter.broker.protocol

Common security config

• principal.builder.class

SSL Configs

• ssl.protocol
• ssl.provider
• ssl.cipher.suites
• ssl.enabled.protocols
• ssl.truststore.type
• ssl.truststore.location
• ssl.truststore.password
• ssl.keymanager.algorithm
• ssl.trustmanager.algorithm
• ssl.endpoint.identification.algorithm
• ssl.secure.random.implementation
• ssl.client.auth

SASL Configs

• sasl.jaas.config
• sasl.kerberos.service.name
• sasl.kerberos.kinit.cmd
• sasl.kerberos.ticket.renew.window.factor
• sasl.kerberos.ticket.renew.jitter
• sasl.kerberos.min.time.before.relogin
• sasl.enabled.mechanisms
• sasl.kerberos.principal.to.local.rules

 Dynamic update changes:

• SSL configs will be updated by reconfiguring ChannelBuilder and creating a new SslFactory. If SSL is used for inter-broker communication, inconsistent changes (e.g changing CA) should be made by adding a new listener with the new properties. This is true for SASL as well.
• SASL configuration updates will be supported using the dynamic JAAS configuration option sasl.jaas.config
• Updates to advertised.listeners will re-register the new listener in ZK. This update will be not allowed for the listener used in inter-broker communication. In addition to this, AdminClient will not allow updates to the listener that was used to make the alter request.
• When changes are made to listeners, additional logic will be required in the controller to broadcast the updated metadata to all brokers.

 Limitations:

• Configuration updates will not be allowed for the listener used in inter-broker communication. To update inter-broker configs without restarting the broker, a new listener can be created with new configs and the inter-broker listener can be changed to the new listener (similar to the currently documented security upgrade process).

Future Work

Message Format and Inter Broker Protocol

Use case: Avoid additional rolling restarts during upgrade

Config scope: Default for whole cluster (/configs/brokers/<default>)

Config options:

• log.message.format.version
• inter.broker.protocol.version

Dynamic update changes:

• AdminClient needs to verify that all brokers in the cluster support the new version before making an update. To enable this, the version of the broker will be added to the JSON registered by each broker during startup at /brokers/ids/id. To avoid all brokers reading version information from ZooKeeper, the controller should propagate this information in UpdateMetadataRequest. This requires protocol change. Note also that this approach only allows verification of the version of brokers that are alive at the time AlterConfigs is executed.

Compatibility, Deprecation, and Migration Plan

• Broker configuration using a properties object or the properties file server.properties will continue to be supported. So users who do not configure dynamic broker options will not be impacted.
• When downgrading from a new broker to an older version that does not support dynamic configs, any configs that were dynamically configured needs to be added to server.properties. The current configs which include dynamic configs can be obtained using kafka-configs.sh.
• If a config is renamed, broker will be updated to automatically register a dynamic config with the new name that has the value of the old config, if the old config was dynamically configured. So no action is required during upgrade. It will be up to the user to delete the old config when no downgrades are expected.

Rejected Alternatives

Deprecate static config in server.properties when a config is made dynamic

It is useful to support static configs in server.properties to enable quick start of Kafka without setting up new config options in ZooKeeper. To retain the simple user experience, we will continue to enable starting up Kafka with a properties file.