However, there is no standardized or consistent approach for determining which source takes precedence when multiple are provided simultaneously. This inconsistency can cause user confusion and make the system more bug-prone. This KIP aims to establish a clear and uniform precedence order for property loading.

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

Public Interfaces

1) The precedence order for loading properties into Kafka configurations is as follows:

...

All the tool scripts that can be configured through either the command line or configuration files, or both, will be adjusted accordingly.

Validation of required arguments

Required arguments (i.e., --bootstrap-server and --bootstrap-controller for now) should be validated after considering values from the following sources:

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

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

Proposed Changes

Add To ensure that all tool scripts correctly honor the proposed property loading precedence, introduce the helper methods to in CommandLineUtils.java so developers can more easily honor the proposed property loading precedence. All the tool scripts that can be configured through either the command line or configuration files, or both, will be adjusted to call the helper method to follow the proposed property loading precedence.and enforce their use across all tools. This prevents each script from implementing its own logic and ensures consistent behavior.

Since the codebase currently uses three different argument parsers (joptsimple, argparse4j, and AbstractConnectCli), the helper methods are designed to be simple and generic, without depending on any specific external library.

Helper methods

The following code demonstrates the key helper methods:

...

Code Block

language	java
title	ConsoleProducer.java

         Map<String, Object> readerProps() throws IOException {
            Map<String, Object> commandLineMap = new HashMap<>();
            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>  commandlineMapcommandLineMap = new HashMap<>();
            CommandLineUtilscommandLineMap.maybeMergeOption(options, commandlineMap, put(BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOpt));
            CommandLineUtilscommandLineMap.maybeMergeOption(options, commandlineMap, put(COMPRESSION_TYPE_CONFIG, compressionCodecOptcompressionCodec());
            CommandLineUtils.maybeMergeOption(options, commandlineMap, if (options.has(sendTimeoutOpt)) commandLineMap.put(LINGER_MS_CONFIG, options.valueOf(sendTimeoutOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(requestRequiredAcksOpt)) commandLineMap.put(ACKS_CONFIG, options.valueOf(requestRequiredAcksOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(requestTimeoutMsOpt)) commandLineMap.put(REQUEST_TIMEOUT_MS_CONFIG, options.valueOf(requestTimeoutMsOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(messageSendMaxRetriesOpt)) commandLineMap.put(RETRIES_CONFIG, options.valueOf(messageSendMaxRetriesOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(retryBackoffMsOpt)) commandLineMap.put(RETRY_BACKOFF_MS_CONFIG, options.valueOf(retryBackoffMsOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(socketBufferSizeOpt)) commandLineMap.put(SEND_BUFFER_CONFIG, options.valueOf(socketBufferSizeOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(maxMemoryBytesOpt)) commandLineMap.put(BUFFER_MEMORY_CONFIG, options.valueOf(maxMemoryBytesOpt).toString());
            // We currently have 2 options to set the batch.size value.  CommandLineUtils.maybeMergeOption(options, commandlineMap, We'll deprecate/remove one of them in KIP-717.
            if (options.has(batchSizeOpt)) commandLineMap.put(BATCH_SIZE_CONFIG, options.valueOf(batchSizeOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(maxPartitionMemoryBytesOpt)) commandLineMap.put(BATCH_SIZE_CONFIG, options.valueOf(maxPartitionMemoryBytesOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(metadataExpiryMsOpt)) commandLineMap.put(METADATA_MAX_AGE_CONFIG, options.valueOf(metadataExpiryMsOpt).toString());
            if CommandLineUtils(options.maybeMergeOption(options, commandlineMap, has(maxBlockMsOpt)) commandLineMap.put(MAX_BLOCK_MS_CONFIG, options.valueOf(maxBlockMsOpt).toString());

			// The map of default values set by tool scripts            // Properties passed as key=value pairs via command line
            Map<String, Object> commandLineKeyValMap = new HashMap<>();
            if (options.has(commandPropertyOpt)) {
                commandLineKeyValMap.putAll(Utils.propsToStringMap(
                        parseKeyValueArgs(options.valuesOf(commandPropertyOpt))
                ));
            }
            // Configuration files
            Map<String, Object> defaultMapconfigMap = Map.of(
 new HashMap<>();
            if (options.has(commandConfigOpt)) {
                configMap.putAll(Utils.propsToStringMap(
                   BOOTSTRAP_SERVERS_CONFIG, CompressionType.NONE.name,
         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
            toolDefaultMap.put(LINGER_MS_CONFIG, options.valuesOf(sendTimeoutOpt).toString());

			// producerConfigOpt - Configuration files
			// producerPropertyOpt - Properties passed as key=value pairs via command line
		 	// commandlineMap - Command line arguments
			// defaultMap - Default values set by tool scripts
			Map<String, Object> map = CommandLineUtils.mergeByPriority(options, producerConfigOpt, producerPropertyOpt, commandlineMap, defaultMap            toolDefaultMap.put(ACKS_CONFIG, options.valuesOf(requestRequiredAcksOpt).toString());
            toolDefaultMap.put(REQUEST_TIMEOUT_MS_CONFIG, options.valuesOf(requestTimeoutMsOpt).toString());
            toolDefaultMap.put(RETRIES_CONFIG, options.valuesOf(messageSendMaxRetriesOpt).toString());
            toolDefaultMap.put(RETRY_BACKOFF_MS_CONFIG, options.valuesOf(retryBackoffMsOpt).toString());
            toolDefaultMap.put(SEND_BUFFER_CONFIG, options.valuesOf(socketBufferSizeOpt).toString());
            toolDefaultMap.put(BUFFER_MEMORY_CONFIG, options.valuesOf(maxMemoryBytesOpt).toString());
            toolDefaultMap.put(BATCH_SIZE_CONFIG, options.valuesOf(batchSizeOpt).toString());
            toolDefaultMap.put(BATCH_SIZE_CONFIG, options.valuesOf(maxPartitionMemoryBytesOpt).toString());
            toolDefaultMap.put(METADATA_MAX_AGE_CONFIG, options.valuesOf(metadataExpiryMsOpt).toString());
            toolDefaultMap.put(MAX_BLOCK_MS_CONFIG, options.valuesOf(maxBlockMsOpt).toString());

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

Required Updates to Existing Tools

The following table shows the existing and proposed changes. The changes are in GREEN.

Tools

*New "modern" Option for Deprecation

Delaye validation of required arguments

Existing Jira/Discussion

kafka-console-producer.sh

Jira

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

Yes

kafka-consumer-perf-test.sh

Yes

Jira

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

kafka-broker-api-versions.sh

Yes

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

*New "modern" Option for Deprecation - This means the tool script has properties that don’t follow the proposed precedence.

Compatibility, Deprecation, and Migration Plan

...

Space shortcuts

Child pages

Versions Compared

Old Version 15

New Version 16

Key

Public Interfaces

Validation of required arguments

Proposed Changes

Helper methods

Required Updates to Existing Tools

Compatibility, Deprecation, and Migration Plan

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 15

New Version 16

Key

Public Interfaces

Validation of required arguments

Proposed Changes

Helper methods

Required Updates to Existing Tools

Compatibility, Deprecation, and Migration Plan