Status

Current state: Under Discussion

Discussion thread: here

Vote thread: here

JIRA: KAFKA-19112

Motivation

In Kafka, some configurations allow users to specify one or more values as a comma-separated string. However, these configurations currently do not validate whether the input is empty or contains duplicate values, and the documentation does not clearly state whether empty values or null values are allowed.

This ambiguity can easily mislead users into thinking that providing an empty list or null is acceptable. Although duplicate values do not affect system behavior, they introduce semantic redundancy. To maintain clarity and avoid potential misconfigurations, empty lists should be disallowed for those configurations that cause the system to malfunction and duplicate values should be ignored unless they have well defined semantics.

In systems design, a fail-fast system is one that immediately reports at its interface any condition that is likely to indicate a failure. Fail-fast systems are usually designed to stop normal operation rather than attempt to continue a possibly flawed process. 

To improve clarity and prevent configuration errors, we propose rejecting empty or null entries and ignoring duplicate values for the affected configurations. This approach aligns with the fail-fast principle: invalid inputs should be caught as early as possible ideally during configuration parsing rather than during runtime or server startup.

The following configurations will be updated to enforce this validation:

• group.coordinator.rebalance.protocols  
• process.roles  
• etc...

Going forward, passing an empty list in the list to any of these configs will result in a ConfigException.

To support this change, we should also update the Kafka documentation to clearly state that these configurations do not allow empty values. Additionally, the system should throw a ConfigException during configuration parsing to immediately alert users to any invalid input.

We should also align the behavior of LIST-type and STRING-type configurations.

• Currently, there are three representations for an empty list default value: null, an empty string (""), and an empty list ([]). We should standardize all LIST-type configuration default values to use List.of() 
• Additionally, some configurations are currently defined as String but use comma-separated values to represent lists. These should be updated to use the proper List-type instead of relying on string parsing.

For the cleanup.policy  empty value, we defined a new meaning: it indicates infinite retention, which is equivalent to setting retention.ms=-1 and retention.bytes=-1.

• If cleanup.policy is empty and remote.storage.enable is set to true, local log segments will be cleaned based on the values of log.local.retention.bytes and log.local.retention.ms.
• If cleanup.policy is empty and remote.storage.enable is set to false, local log segments will not be deleted automatically. However, records can still be removed explicitly through deleteRecords API calls, which will advance the log start offset and delete the corresponding log segments.

Public Interfaces

• org.apache.kafka.common.config.ConfigDef$ValidList 
• org.apache.kafka.common.config.TopicConfig 
• org.apache.kafka.common.config.SaslConfigs
• org.apache.kafka.server.config.KRaftConfigs 
• org.apache.kafka.coordinator.group.GroupCoordinatorConfig 
• org.apache.kafka.tools.BrokerApiVersionsCommand$AdminClient 
• org.apache.kafka.connect.mirror.MirrorSourceTaskConfig 
• org.apache.kafka.connect.mirror.MirrorCheckpointTaskConfig 
• org.apache.kafka.connect.mirror.MirrorCheckpointConfig 
• org.apache.kafka.connect.runtime.rest.RestServerConfig 

• org.apache.kafka.connect.mirror.MirrorClientConfig 

• org.apache.kafka.server.config.ServerConfigs 
• org.apache.kafka.network.SocketServerConfigs 
• org.apache.kafka.server.config.ServerLogConfigs 
• org.apache.kafka.server.metrics.ClientMetricsConfigs 
• org.apache.kafka.server.metrics.MetricConfigs 
• org.apache.kafka.connect.transforms.ReplaceField 
• org.apache.kafka.connect.transforms.MaxkField 
• org.apache.kafka.connect.transforms.DropHeaders 
• org.apache.kafka.connect.transforms.ValueToKey 
• org.apache.kafka.connect.runtime.SinkConnectorConfig 
• org.apache.kafka.connect.runtime.WorkerConfig 
• org.apache.kafka.connect.mirror.MirrorMakerConfig 
• org.apache.kafka.connect.mirror.MirrorConnectorConfig 
• org.apache.kafka.clients.consumer.ConsumerConfig 
• org.apache.kafka.clients.producer.ProducerConfig 
• org.apache.kafka.connect.mirror.MirrorSourceConfig 
• org.apache.kafka.connect.mirror.DefaultConfigPropertyFilter 
• org.apache.kafka.connect.mirror.DefaultTopicFilter 
• org.apache.kafka.common.config.internals.BrokerSecurityConfigs 

Proposed Changes

We will make the following changes to ValidList :

• Introduce in(String, boolean) public method: Create validators that can optionally reject empty lists while still enforcing valid string constraints.
• Introduce anyNonDuplicateValues()  public method: Creates validators that explicitly reject value(null, empty list) by use setting
• Update the ensureValid() method to throw a ConfigException  if the user provides duplicate values in the list, and also add logic to check null value

public static class ValidList implements Validator {

    final ValidString validString;
    final boolean isEmptyAllowed;
    final boolean isNullAllowed;

    private ValidList(List<String> validStrings, boolean isEmptyAllowed, boolean isNullAllowed) {
        this.validString = new ValidString(validStrings);
        this.isEmptyAllowed = isEmptyAllowed;
        this.isNullAllowed = isNullAllowed;
    }

    public static ValidList anyNonDuplicateValues(boolean isEmptyAllowed, boolean isNullAllowed) {
        return new ValidList(List.of(), isEmptyAllowed, isNullAllowed);
    }

    public static ValidList in(String... validStrings) {
        return new ValidList(List.of(validStrings), true, false);
    }

    public static ValidList in(boolean isEmptyAllowed, String... validStrings) {
        if (validStrings.length == 0) {
            throw new IllegalArgumentException("Valid strings list cannot be empty for inNonEmpty validator");
        }
        return new ValidList(List.of(validStrings), isEmptyAllowed, false);
    }            

    @Override
    public void ensureValid(final String name, final Object value) {
        if (value == null) {
            if (isNullAllowed)
                return;
            else
                throw new ConfigException("Configuration '" + name + "' values must not be null.");
        }

        @SuppressWarnings("unchecked")
        List<Object> values = (List<Object>) value;
        if (!isEmptyAllowed && values.isEmpty()) {
            String validString = this.validString.validStrings.isEmpty() ? "any non-empty value" : this.validString.toString();
            throw new ConfigException("Configuration '" + name + "' must not be empty. Valid values include: " + validString);
        }

        if (values.size() > 1 && Set.copyOf(values).size() != values.size()) {
            throw new ConfigException("Configuration '" + name + "' values must not be duplicated.");
        }

        validateIndividualValues(name, values);
    }

    private void validateIndividualValues(String name, List<Object> values) {
        boolean hasValidStrings = !validString.validStrings.isEmpty();

        for (Object value : values) {
            if (value instanceof String) {
                String string = (String) value;
                if (string.isEmpty()) {
                    throw new ConfigException("Configuration '" + name + "' values must not be empty.");
                }
                if (hasValidStrings) {
                    validString.ensureValid(name, value);
                }
            }
        }
    }

    public String toString() {
        return !validString.validStrings.isEmpty() ? validString.toString() : "";
    }
}

The following configurations will be modified, If the value is not modified, a dash will be used instead.

Note:

1. If cleanup.policy is empty and remote.storage.enable is set to true, the local log segments will be cleaned based on the values of log.local.retention.bytes and log.local.retention.ms. 

2. For early.start.listeners, this configuration is optional (users are not required to set it). This means the field can be omitted or set to an empty list. However, duplicate values will still be rejected regardless of these settings.
3. For log.dirs, this configuration is also optional (users are not required to set it). However when provided, it must be a non-empty list with no duplicate values. A null value is acceptable, but an empty list will be rejected. 
4. For log.dir, this configuration is also optional (users are not required to set it). However when provided, it must be a non-empty list with no duplicate values. A null value is acceptable, but an empty list will be rejected
5. For plugin.path, this configuration is also optional (users are not required to set it). However when provided, it must be a non-empty list with no duplicate values. A null value is acceptable, but an empty list will be rejected. 
6. Currently, the default value of bootstrap.servers is "localhost:9092". However, this is somewhat odd, as the default value should ideally be useful, and "localhost:9092" is not useful in most scenarios. 
7. Currently, process.roles allows an empty value during the storage format validation stage, but it will fail during server startup. Since the Kafka node relies on this configuration to determine its role, it is a required setting and should not be left empty.
8. Currently, if controller.listener.names is set to an empty list or null, it fails for different reasons depending on the node role. A broker node will return the error: "controller.listener.names must contain at least one value when running KRaft with just the broker role", while a controller node will encounter validation errors when attempting to match the listeners configuration
9. Currently, if listeners is set to an empty list, which means no network interfaces and protocols that a Kafka node listens on.
10. Currently, if advertised.listeners is set to an empty string, an exception is thrown only when the node is running as a broker. In other scenarios, no exception is raised. However, this configuration still results in the node not advertising any listener addresses, making it unreachable to clients.
11. Currently, if log.dirs is set to an empty list, and log.dir is empty, Kafka will fail during the storage format phase with an error stating that "At least one log directory must be defined via log.dirs or log.dir".
12. Currently, if log.dir is set to an empty list, and log.dirs is empty, Kafka will fail during the storage format phase with an error stating that "At least one log directory must be defined via log.dirs or log.dir".
13. Currently, if group.coordinator.rebalance.protocols is set to an empty list, the server cannot determine which rebalance protocol to use for the consumer group.
14. Currently, if bootstrap.servers is set to an empty list, It will fail to initialize KafkaAdminClient when lookup Kafka cluster id.
15. Currently, if plugin.path is set to an empty list, no user-specified plugin locations are processed, but plugins from the parent classloader are still loaded, which may go against the user's explicit intention, as they might expect that no plugins would be loaded at all. In contrast, if plugin.path is not set (null), it's reasonable to load plugins from the parent classloader, as the user is likely relying on the default behavior
16. Currently, if bootstrap.servers is set to an empty list, It will fail to initialize ForwardingAdmin.

17. Currently, if task.assigned.partitions is set to an empty list, which means no MirrorSourceTask can be created.

18. Currently, if task.assigned.groups is set to an empty list, which means no Checkpoint can be created.
19. Currently, if bootstrap.servers is set to an empty list, it will fail to construct a consumer because there are no resolvable bootstrap URLs.
20. Currently, if bootstrap.servers is set to an empty list, it will fail to construct a producer because there are no resolvable bootstrap URLs.
21. Currently, if bootstrap.servers is set to an empty list, It will fail to initialize KafkaAdminClient and consumer.

Compatibility, Deprecation, and Migration Plan

We add the three main changes in this KIP

1. Disallowing null values for most LIST-type configurations makes sense, since users cannot explicitly set a configuration to null in a properties file. Therefore, only configurations with a default value of null should be allowed to accept null.

2. Ignoring duplicate values is reasonable, as less Kafka configuration currently requires repeating the same value, and allowing duplicates can confuse users. Therefore, most LIST-type configurations no longer accept duplicates unless explicitly supported. For backward compatibility, unsupported duplicate entries will be ignored and a warning will be logged.

3. Disallowing empty list, even though many configurations currently accept them. In practice, setting an empty list for several of these configurations can lead to server startup failures or unexpected behavior. Therefore, enforcing non-empty lists helps prevent misconfiguration and improves system robustness.

These changes may introduce some backward incompatibility, but this trade-off is justified by the significant improvements in safety, consistency, and overall user experience. 

Additionally, we introduce two minor adjustments:

1. Reclassify some STRING-type configurations as LIST-type, particularly those using comma-separated values to represent multiple entries. This change reflects the actual semantics used in Kafka.
2. Update the default values for some configurations to better align with other configs.

These changes will not introduce any compatibility issues.

The migration plan.

• Introduce new methods in(), anyNonDuplicateValues() in ValidList 
• Update the usage of all configurations to follow the validation rules defined above.

Test Plan

• Existing test should be all passing
• Add new test for ValidList#anyNonDuplicateValues , ValidList#in 

Rejected Alternatives

Using new cleanup.policy type 'none'

. This approach was rejected for the following reasons:

• Infinite retention is already achievable by setting log.retention.ms = -1 and log.retention.bytes = -1.
• Introducing a new type could introduce unnecessary complexity and diverge from the existing retention semantics.
• Allowing an empty list for cleanup.policy (which implies no compaction or deletion) is more consistent with the general configuration model for list-based configs.