Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Kafka allows users to configure its behavior through multiple methods:

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

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 priority order for property loading.

Public Interfaces


Deprecate:

--modern

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

  1. Command line arguments
  2. Properties passed as key=value pairs via command line
  3. Configuration files
  4. Default values

...

  1. provided by tool scripts
  2. Default values set by ConfigDef 

Proposed Changes

Tools that can be configured through both the command - line arguments and configuration files , or that do not follow this precedence, should be adjusted accordingly. Add the helper methods in CommandLineUtils.java, for all class inheritant CommandDefaultOptions can more easily honor the proposed property loading priority.

The following code is a demonstration of the key helper method:

Code Block
languagejava
titleCommandLineUtils.java
    /**
     * Merges multiple configuration sources by priority. The merge order is as follows, from high to low:
     * 1) Command line arguments
     * 2) Properties passed as key=value pairs via command line
     * 3) Configuration files
     * 4) Default values provided by tool scripts
     */
    public static Map<String, Object> mergeByPriority(OptionSet options, OptionSpec<String> configOpt, OptionSpec<String> propertyOpt, Map<String, Object> overrides, Map<String, Object> defaultIfMissing) throws IOException {
        Map<String, Object> map = new HashMap<>();

        if (defaultIfMissing != null) {
            map.putAll(defaultIfMissing);
        }
        if (configOpt != null && options.has(configOpt)) {
            map.putAll(Utils.propsToMap(
                Utils.loadProps(options.valueOf(configOpt)))
            );
        }
        if (propertyOpt != null && options.has(propertyOpt)) {
            map.putAll(Utils.propsToMap(
                parseKeyValueArgs(options.valuesOf(propertyOpt)))
            );
        }
        if (overrides != null) {
            map.putAll(overrides);
        }

        return map;
    }

    public static void maybeMergeOption(OptionSet options, Map<String, Object> map, String key, OptionSpec<?> spec, Object defaultValue)
    {
        Object value = null;
        if (options.has(spec)) {
            value = options.valueOf(spec);
            // This can also be null. For example, --compression-codec specified without a value and with no default value set in the option
			if (value == null) {
				System.err.println(key);
			}
        }

        if (value == null) {
            value = defaultValue;
        }

        if (value != null) {
            map.put(key, value);
        }
    }


The following tools will be updated:

  • ConsoleConsumer, ConsoleProducer, ConsoleShareConsumer
  • ConsumerPerformance, ShareConsumerPerformance, ProducerPerformance
  • DelegationTokenCommand

Example usage

Code Block
languagejava
titleConsoleProducer.java
        Map<String, Object> producerProps() throws IOException {
            // Prepare the map from command line arguments
			Map<String, Object> commandlineMap = new HashMap<>();
            CommandLineUtils.maybeMergeOption(options, commandlineMap, BOOTSTRAP_SERVERS_CONFIG, bootstrapServerOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, COMPRESSION_TYPE_CONFIG, compressionCodecOpt, CompressionType.GZIP.name);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, LINGER_MS_CONFIG, sendTimeoutOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, ACKS_CONFIG, requestRequiredAcksOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMsOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, RETRIES_CONFIG, messageSendMaxRetriesOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, RETRY_BACKOFF_MS_CONFIG, retryBackoffMsOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, SEND_BUFFER_CONFIG, socketBufferSizeOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, BUFFER_MEMORY_CONFIG, maxMemoryBytesOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, BATCH_SIZE_CONFIG, batchSizeOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, BATCH_SIZE_CONFIG, maxPartitionMemoryBytesOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, METADATA_MAX_AGE_CONFIG, metadataExpiryMsOpt);
            CommandLineUtils.maybeMergeOption(options, commandlineMap, MAX_BLOCK_MS_CONFIG, maxBlockMsOpt);

			// The map of default values set by tool scripts
            Map<String, Object> defaultMap = Map.of(
                BOOTSTRAP_SERVERS_CONFIG, CompressionType.NONE.name,
                CLIENT_ID_CONFIG, "console-producer",
                KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer",
                VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"
            );

            Map<String, Object> map = CommandLineUtils.mergeByPriority(this, producerConfigOpt, producerPropertyOpt, commandlineMap, defaultMap);

            return map;
        }


Compatibility, Deprecation, and Migration Plan

...