...

Current state: Under Discussion

Discussion thread: WIP https://lists.apache.org/thread/vlqvcyl5j95t7w9rk3nnz09d0050yzh9

JIRA:

Jira

server	ASF JIRA
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-10043

...

Command line arguments
Properties passed as key=value pairs via command line (Command line properties)
Configuration files

However, there is no standardized or consistent approach for determining standard, consistent way to determine which source takes precedence when multiple sources are provided. This inconsistency can confuse users and make the system more error-prone.

...

Furthermore, the validation of required arguments, such as --bootstrap-server and --bootstrap-controller, should consider multiple sources. For most existing tool scripts, these arguments configurations are only validated when provided via the validated only by checking if they are provided through command line arguments. However, they could also be specified through properties passed as key=value pairs via the command line properties or configuration files.

Public Interfaces

...

Command line arguments
Properties passed as key=value pairs via command line (Command line properties)
Configuration files
Default values set by tool scripts and default values of command line arguments
Default values set by Kafka components

2) A new option is added for all affected tool scripts not following the proposed precedence:

In addition, a warning is issued when the same configuration appears in multiple sources (command line, command line properties, or configuration file), which is helpful for troubleshooting if users did not intend to specify the same configuration multiple times.

Note: property or map entries may have empty string or null values, as some configurations allow them, so the value is used verbatim.

2) A new option is added for all affected tool scripts not following the proposed precedence:

Name: modern
Description: This configuration determines whether to enable the configuration precedence proposed in this KIP. It will be deprecated in Kafka 5.
Name: modern
Description: This configuration determines whether to enable the configuration precedence proposed in this KIP. It will be deprecated in Kafka 5.0, after which the default behavior will always follow the approach proposed by this KIP. It will then be removed in Kafka 6.0.
Type: boolean
Default: false

...

Command line arguments
Properties passed as key=value pairs via command line (Command line properties)
Configuration Configuration files

An error should only be raised if the arguments are missing or invalid after checking all three sources.

...

Code Block

language	java
title	CommandLineUtils.javajava

    /**
     * Merge multiple configuration sources according to priority, from highest to lowest:
     * 1) Command line arguments
     * 2) Properties passed as key=value pairs via command line (Command line properties)
     * 3) Configuration files
     * 4) Default values set by tool scripts and default values of command line arguments
     *
     * @param commandLineMap       Map of configuration from command line arguments
     * @param commandLineKeyValMap Map of configuration from key=value pairs passed via command line (command line properties)
     * @param configMap            Map of configuration from configuration files
     * @param toolDefaultMap       Map of default values provided by tool scripts or default command line options
     * @return a merged Map of all configuration sources, where higher-priority sources override lower-priority ones
     */
	public static Map<String, Object> mergePropertiesWithPrecedence(
        Map<String, Object> commandLineMap,
        Map<String, Object> commandLineKeyValMap,
        Map<String, Object> configMap,
        Map<String, Object> toolDefaultMap
    ) {
		warnConfigFromMultipleSources(commandLineMap, commandLineKeyValMap, configMap);
        
		Map<String, Object> map = new HashMap<>();
        // Default values set by tool scripts and default values of command line arguments
        if (toolDefaultMap != null) {
            map.putAll(toolDefaultMap);
        }
        // Configuration file
        if (configMap != null) {
            map.putAll(configMap);
        }
        // Properties passed as key=value pairs via command line (Command line properties)
        if (commandLineKeyValMap != null) {
            map.putAll(commandLineKeyValMap);
        }
        // Command line arguments
        if (commandLineMap != null) {
            map.putAll(commandLineMap);
        }
        return map;
    }

    /**
     * Print warnings when the same configurations is defined in multiple sources.
     * - Command line arguments override command property values and configuration files.
     * - Command property values override configuration files if no command line argument exists.
     *
     * @param commandLineMap       Map of configuration from command line arguments
     * @param commandLineKeyValMap Map of configuration from command line properties
     * @param configMap            Map of configuration loaded from configuration files
     */
    public static void warnConfigFromMultipleSources(
        Map<String, Object> commandLineMap,
        Map<String, Object> commandLineKeyValMap,
        Map<String, Object> configMap
    ) {
        // Command line argument overrides command property and config file
        for (Map.Entry<String, Object> entry : commandLineMap.entrySet()) {
            String key = entry.getKey();
            Object value = entry.getValue();
            StringBuilder warning = new StringBuilder();

            if (commandLineKeyValMap.containsKey(key) && configMap.containsKey(key)) {
                warning.append(key).append("=").append(value)
                    .append(" from command line argument overrides command property value (")
                    .append(commandLineKeyValMap.get(key)).append(")")
                    .append(" and configuration file value (")
                    .append(configMap.get(key)).append(").");
            } else if (commandLineKeyValMap.containsKey(key)) {
                warning.append(key).append("=").append(value)
                    .append(" from command line argument overrides command property value (")
                    .append(commandLineKeyValMap.get(key)).append(").");
            } else if (configMap.containsKey(key)) {
                warning.append(key).append("=").append(value)
                    .append(" from command line overrides configuration file value (")
                    .append(configMap.get(key)).append(").");
            }

            if (!warning.isEmpty()) {    /**
     * Merge multiple configuration sources according to priority, from highest to lowest:
 System.out.println("Warning: " + warning);
            }
  *   1) Command line arguments
}

        // Command property *override 2)config Propertiesfile
 passed as key=value pairs via command line
 for (Map.Entry<String, Object> entry * 3: commandLineKeyValMap.entrySet()) Configuration files{
     * 4) Default values set by tool scriptsString andkey default values of command line arguments
= entry.getKey();
         */     
	public staticObject Map<String,value Object>= mergePropertiesWithPrecedenceentry.getValue();
         Map<String,   Object>if commandLineMap,(commandLineMap.containsKey(key)) continue;

        Map<String, Object> commandLineKeyValMap,
  if   (configMap.containsKey(key)) {
   Map<String, Object> configMap,
        Map<String, Object> toolDefaultMap
    ) {
    System.out.println("Warning: " + key + "=" + value
     Map<String, Object> map = new HashMap<>();
        // Default values+ set" byfrom toolcommand scriptsproperty andoverrides defaultconfiguration valuesfile of command line arguments
 value ("
       if (toolDefaultMap != null) {
         + configMap.get(key) + map.putAll(toolDefaultMap").");
        }
        // Configuration file}
        if}
 (configMap != null) {}

	// joptsimple-specific helper method
    public static <T> void  map.putAll(configMap);
        }putIfOption(OptionSet options, OptionSpec<T> spec, Map<String, Object> map, String key) {
        // Properties passed as key=value pairs via command line
if (options.has(spec)) {
            ifT (commandLineKeyValMapvalue != null) {options.valueOf(spec);
            map.putAll(commandLineKeyValMapput(key, value);
        }
    }

	// joptsimple-specific helper method
  // Command linepublic arguments
static <T> void putOption(OptionSet options, OptionSpec<T> spec, Map<String, ifObject> (commandLineMap != nullmap, String key) {
        T value =  mapoptions.putAllvalueOf(commandLineMapspec);
        }
        return mapmap.put(key, value);
    }

Example usage

Code Block

language	java
title	ConsoleProducer.java

        Map<String, Object> readerProps() throws IOException {
            Map<String, Object> commandLineMap = new HashMap<>();
            putOption(options, topicOpt, commandLineMap.put(, "topic", options.valueOf(topicOpt));

            Map<String, Object> configMap = new HashMap<>();
            if (options.has(readerConfigOpt)) {
                configMap.putAll(propsToStringMap(loadProps(options.valueOf(readerConfigOpt))));
            }

            Map<String, Object> commandLineKeyValMap = new HashMap<>();
            if (options.has(readerPropertyOpt)) {
                commandLineKeyValMap.putAll(propsToStringMap(parseKeyValueArgs(options.valuesOf(readerPropertyOpt))));
            }
            return mergePropertiesWithPrecedence(commandLineMap, commandLineKeyValMap, configMap, null);
        }

            
		Map<String, Object> producerProps() throws IOException {
            // Prepare the map from command line arguments
            Map<String, Object>  commandLineMap = new HashMap<>();
            putIfOption(options, bootstrapServerOpt, commandLineMap, BOOTSTRAP_SERVERS_CONFIG);
            commandLineMap.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
            putIfOption(options, sendTimeoutOpt,  commandLineMap.put(BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOpt))commandLineMap, LINGER_MS_CONFIG);
            commandLineMap.put(COMPRESSION_TYPE_CONFIG, compressionCodec()) putIfOption(options, requestRequiredAcksOpt, commandLineMap, ACKS_CONFIG);
            if putIfOption(options.has(sendTimeoutOpt)) commandLineMap.put(LINGER_MS_CONFIG, options.valueOf(sendTimeoutOpt).toString()), requestTimeoutMsOpt, commandLineMap, REQUEST_TIMEOUT_MS_CONFIG);
            if putIfOption(options.has(requestRequiredAcksOpt)) commandLineMap.put(ACKS_CONFIG, options.valueOf(requestRequiredAcksOpt).toString()), messageSendMaxRetriesOpt, commandLineMap, RETRIES_CONFIG);
            if putIfOption(options.has(requestTimeoutMsOpt)) commandLineMap.put(REQUEST_TIMEOUT_MS_CONFIG, options.valueOf(requestTimeoutMsOpt).toString()), retryBackoffMsOpt, commandLineMap, RETRY_BACKOFF_MS_CONFIG);
            if putIfOption(options.has(messageSendMaxRetriesOpt)) commandLineMap.put(RETRIES_CONFIG, options.valueOf(messageSendMaxRetriesOpt).toString()), socketBufferSizeOpt, commandLineMap, SEND_BUFFER_CONFIG);
            if putIfOption(options.has(retryBackoffMsOpt)) commandLineMap.put(RETRY_BACKOFF_MS_CONFIG, options.valueOf(retryBackoffMsOpt).toString());
 , maxMemoryBytesOpt, commandLineMap, BUFFER_MEMORY_CONFIG);
           if (options.has(socketBufferSizeOpt)) commandLineMap.put(SEND_BUFFER_CONFIG, options.valueOf(socketBufferSizeOpt).toString());
            if (options.has(maxMemoryBytesOpt)) commandLineMap.put(BUFFER_MEMORY_CONFIG, options.valueOf(maxMemoryBytesOpt).toString()); // We currently have 2 options to set the batch.size value. We'll deprecate/remove one of them in KIP-717.
            if putIfOption(options.has(batchSizeOpt)) commandLineMap.put(, batchSizeOpt, commandLineMap, BATCH_SIZE_CONFIG, options.valueOf(batchSizeOpt).toString());
            if putIfOption(options.has(maxPartitionMemoryBytesOpt)) commandLineMap.put(, maxPartitionMemoryBytesOpt, commandLineMap, BATCH_SIZE_CONFIG, options.valueOf(maxPartitionMemoryBytesOpt).toString())CONFIG);
            if putIfOption(options.has(metadataExpiryMsOpt)) commandLineMap.put(, metadataExpiryMsOpt, commandLineMap, METADATA_MAX_AGE_CONFIG, options.valueOf(metadataExpiryMsOpt).toString());
            if putIfOption(options.has(maxBlockMsOpt)) commandLineMap.put(, maxBlockMsOpt, commandLineMap, MAX_BLOCK_MS_CONFIG, options.valueOf(maxBlockMsOpt).toString());

            // Properties passed as key=value pairs via command line (Command line properties)
            Map<String, Object> commandLineKeyValMap = new HashMap<>();
            if (options.has(commandPropertyOpt)) {
                commandLineKeyValMap.putAll(Utils.propsToStringMap(
                        parseKeyValueArgs(options.valuesOf(commandPropertyOpt))
                ));
            }

            // Configuration files
            Map<String, Object> configMap = new HashMap<>();
            if (options.has(commandConfigOpt)) {
                configMap.putAll(Utils.propsToStringMap(
                        Utils.loadProps(options.valueOf(commandConfigOpt))
                ));
            }

            // Default values set by tool scripts and default values of command line arguments
            Map<String, Object> toolDefaultMap = new HashMap<>();
            toolDefaultMap.put(CLIENT_ID_CONFIG, "console-producer");
            toolDefaultMap.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
            toolDefaultMap.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
            // Since all the options below have default values, we don't need to check for null
            putOption(options, sendTimeoutOpt, toolDefaultMap, toolDefaultMap.put(LINGER_MS_CONFIG, options.valuesOf(sendTimeoutOpt).toString());
            putOption(options, requestRequiredAcksOpt, toolDefaultMap.put(, ACKS_CONFIG, options.valuesOf(requestRequiredAcksOpt).toString());
            toolDefaultMap.put(putOption(options, requestTimeoutMsOpt, toolDefaultMap, REQUEST_TIMEOUT_MS_CONFIG, options.valuesOf(requestTimeoutMsOpt).toString());
            putOption(options, messageSendMaxRetriesOpt, toolDefaultMap.put(, RETRIES_CONFIG, options.valuesOf(messageSendMaxRetriesOpt).toString());
            putOption(options, retryBackoffMsOpt, toolDefaultMap.put(, RETRY_BACKOFF_MS_CONFIG, options.valuesOf(retryBackoffMsOpt).toString());
            putOption(options, socketBufferSizeOpt, toolDefaultMap.put(, SEND_BUFFER_CONFIG, options.valuesOf(socketBufferSizeOpt).toString())CONFIG);
            putOption(options, maxMemoryBytesOpt, toolDefaultMap.put(, BUFFER_MEMORY_CONFIG, options.valuesOf(maxMemoryBytesOpt).toString());
            putOption(options, toolDefaultMap.put(batchSizeOpt, toolDefaultMap, BATCH_SIZE_CONFIG, options.valuesOf(batchSizeOpt).toString());
            toolDefaultMap.put(BATCH_SIZE_CONFIG, options.valuesOf(maxPartitionMemoryBytesOpt).toString())putOption(options, maxPartitionMemoryBytesOpt, toolDefaultMap, BATCH_SIZE_CONFIG);
            toolDefaultMap.put(putOption(options, metadataExpiryMsOpt, toolDefaultMap, METADATA_MAX_AGE_CONFIG, options.valuesOf(metadataExpiryMsOpt).toString());
            toolDefaultMap.put(putOption(options, maxBlockMsOpt, toolDefaultMap, MAX_BLOCK_MS_CONFIG, options.valuesOf(maxBlockMsOpt).toString());

            			return mergePropertiesWithPrecedence(commandLineMap, commandLineKeyValMap, configMap, toolDefaultMap);
        }

...

Tools

Need New "modern" Option and List properties that do not follow the proposed precedence

Validating Required Arguments from Multiple Sources

Previous Jira/Email Discussion

Note

kafka-acls.sh

--bootstrap-server

--bootstrap-controller

kafka-broker-api-versions.sh

--bootstrap-server

https://lists.apache.org/thread/yh21q5mm1kst0f1g8gommmp9hzy3dqzw

kafka-client-metrics.sh

--bootstrap-server

kafka-cluster.sh

--bootstrap-server

--bootstrap-controller

kafka-configs.sh

--bootstrap-server

--bootstrap-controller

kafka-console-consumer.sh

For formatter:

KEY_DESERIALIZER_CLASS_CONFIG

VALUE_DESERIALIZER_CLASS_CONFIG

--bootstrap-server

--from-beginning (AUTO_OFFSET_RESET_CONFIG)

--group (GROUP_ID_CONFIG) is validated but does not follow the proposed validation flow

kafka-console-producer.sh

KEY_SERIALIZER_CLASS_CONFIG

VALUE_SERIALIZER_CLASS_CONFIG

COMPRESSION_TYPE_CONFIG

--bootstrap-server

Jira

server	ASF JIRA
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-2526

kafka-console-share-consumer.sh

For formatter:

KEY_DESERIALIZER_CLASS_CONFIG

VALUE_DESERIALIZER_CLASS_CONFIG

--bootstrap-server

--group (GROUP_ID_CONFIG) is validated but does not follow the proposed validation flow

kafka-consumer-groups.sh

--bootstrap-server

kafka-consumer-perf-test.sh

GROUP_ID_CONFIG

RECEIVE_BUFFER_CONFIG

MAX_PARTITION_FETCH_BYTES_CONFIG

AUTO_OFFSET_RESET_CONFIG

KEY_DESERIALIZER_CLASS_CONFIG

VALUE_DESERIALIZER_CLASS_CONFIG

CHECK_CRCS_CONFIG

--bootstrap-server

Jira

server	ASF JIRA
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-10043

kafka-delegation-tokens.sh

--bootstrap-server

kafka-delete-records.sh

--bootstrap-server

kafka-e2e-latency.sh

For consumer:

GROUP_ID_CONFIG

ENABLE_AUTO_COMMIT_CONFIG

AUTO_OFFSET_RESET_CONFIG

KEY_DESERIALIZER_CLASS_CONFIG

VALUE_DESERIALIZER_CLASS_CONFIG

FETCH_MAX_WAIT_MS_CONFIG

For producer:

LINGER_MS_CONFIG

MAX_BLOCK_MS_CONFIG

ACKS_CONFIG

KEY_SERIALIZER_CLASS_CONFIG

VALUE_SERIALIZER_CLASS_CONFIG

--bootstrap-server

--producer-acks (ACKS_CONFIG)

kafka-features.sh

--bootstrap-server

--bootstrap-controller

kafka-get-offsets.sh

CLIENT_ID_CONFIG

--bootstrap-server

kafka-groups.sh

--bootstrap-server

kafka-leader-election.sh

--bootstrap-server

kafka-log-dirs.sh

--bootstrap-server

kafka-metadata-quorum.sh

--bootstrap-server

--bootstrap-controller

kafka-producer-perf-test.sh

KEY_SERIALIZER_CLASS_CONFIG

VALUE_SERIALIZER_CLASS_CONFIG

kafka-reassign-partitions.sh

--bootstrap-server

--bootstrap-controller

kafka-share-consumer-perf-test.sh

GROUP_ID_CONFIG

RECEIVE_BUFFER_CONFIG

MAX_PARTITION_FETCH_BYTES_CONFIG

AUTO_OFFSET_RESET_CONFIG

KEY_DESERIALIZER_CLASS_CONFIG

VALUE_DESERIALIZER_CLASS_CONFIG

CHECK_CRCS_CONFIG

--bootstrap-server

kafka-share-groups.sh

--bootstrap-server

kafka-streams-application-reset.sh

--bootstrap-server

This tool script is the only one with a default bootstrap-server; consider aligning it with the others.

kafka-streams-groups.sh

--bootstrap-server

kafka-topics.sh

--bootstrap-server

kafka-transactions.sh

--bootstrap-server

kafka-verifiable-consumer.sh

GROUP_PROTOCOL_CONFIG

GROUP_REMOTE_ASSIGNOR_CONFIG

PARTITION_ASSIGNMENT_STRATEGY_CONFIG

--group-id (GROUP_ID_CONFIG)

ENABLE_AUTO_COMMIT_CONFIG

AUTO_OFFSET_RESET_CONFIG

--bootstrap-server

--group-id (GROUP_ID_CONFIG)

kafka-verifiable-producer.sh

KEY_SERIALIZER_CLASS_CONFIG

VALUE_SERIALIZER_CLASS_CONFIG

ACKS_CONFIG

RETRIES_CONFIG

--bootstrap-server

kafka-verifiable-share-consumer.sh

--bootstrap-server

--group-id (GROUP_ID_CONFIG)

...

Space shortcuts

Child pages

Versions Compared

Old Version 25

New Version Current

Key

Public Interfaces

Example usage

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 25

New Version Current

Key

Public Interfaces

Example usage