Versions Compared

Key

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

Table of Contents

Status

Current state: Under Discussion

...

JIRA:

Jira
serverASF JIRA
serverId5aa69414-a9e9-3523-82ec-879b028fb15b
keyKAFKA-10043

Motivation

Kafka allows users to configure its behavior through multiple methods:

...

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

Public Interfaces

1) The precedence order 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 set by tool scripts and default values of command line arguments
  5. Default values set by ConfigDef

...

  • Name: modern
  • Description: This configuration determines whether to enable the property loading priority proposed 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

...

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

Proposed Changes

Add the helper methods to CommandLineUtils.java so developers can more easily honor the proposed property loading priorityprecedence. 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 priorityprecedence.

The following code demonstrates the key helper methods:

Code Block
languagejava
titleCommandLineUtils.java
    /**
     * 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
     * 3) Configuration files
     * 4) Default values set by tool scripts
     */
      
	public static Map<String, Object> mergeByPrioritymergePropertiesWithPrecedence(OptionSet
 options, OptionSpec<String> configOpt, OptionSpec<String> propertyOpt, Map<String, Object> overrides, Map<String, Object> defaultMap) throws IOException {commandLineMap,
        Map<String, Object> map = new HashMap<>();
		// Default values set by tool scripts
        if (defaultMap != null) {commandLineKeyValMap,
        Map<String, Object> configMap,
        Map<String,    map.putAll(defaultMap);
        }
		// Configuration file
        if (configOpt != null && options.has(configOpt))Object> toolDefaultMap
    ) {
        Map<String, Object>   map.putAll(Utils.propsToMap(
                Utils.loadProps(options.valueOf(configOpt))) = new HashMap<>();
        // Default values set );
by tool scripts and default    }
		// Properties passed as key=value pairs viavalues of command line arguments
        if (propertyOpttoolDefaultMap != null && options.has(propertyOpt)) {
            map.putAll(Utils.propsToMap(toolDefaultMap);
                parseKeyValueArgs(options.valuesOf(propertyOpt)))}
        // Configuration   );
  file
      }
		// Command line arguments
		if (overridesconfigMap != null) {
            map.putAll(overridesconfigMap);
        }

        return map;
    }      

	/**
     * Merge the option into {@code map} for the given {@code key} using the following logic:
     * 1) Merge the option value into the map if the option is present and has a value or a default value provided.// Properties passed as key=value pairs via command line
     * 2) Use {@code valueIfNoRequiredArg} if the(commandLineKeyValMap option is present but its value is null.
     * 3) Otherwise, do nothing
     */
!= null) {
     public static void maybeMergeOption(OptionSet options, Map<String, Object> map, String key, OptionSpec<?> spec, Object valueIfNoRequiredArg)
    {
        if (!options.has(spec)) {
            return.putAll(commandLineKeyValMap);
        }

        Object value = options.valueOf(spec);
        if (value == null) {
            // ThisCommand can also be null if the option is meant to be used without any argument (e.g., --from-latest)
    line arguments
        // In that case, valueIfNoRequiredArg acts as the value
            if (valueIfNoRequiredArgcommandLineMap !== null) {
                System.err.println("No value specified for option \"" + key + "\" and no value provided.");
                Exit.exit(1);
    map.putAll(commandLineMap);
        }
            value = valueIfNoRequiredArg;
        }

        map.put(key, value)return map;
    }

    public static void maybeMergeOption(OptionSet options, Map<String, Object> map, String key, OptionSpec<?> spec)
    {
        maybeMergeOption(options, map, key, spec, null);
    }


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);
            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"
            );

			// 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);

            return map;
        }


Compatibility, Deprecation, and Migration Plan

Impact: Existing users who have configured their settings based on the current tool implementations, rather than the proposed priorityprecedence, may be affected.

Deprecation: Adjusting property loading priority precedence might break existing users. For this reason, we have an option added to each affected tool script to enable or disable this change. The plan is to always use the proposed priority in precedence in the next major release, Kafka 5.0. Concurrently, the option will be deprecated in Kafka 5.0 and subsequently removed in Kafka 6.0.

Test Plan

Unit tests will be added to ensure that the properties are loaded according to the priority order precedence order as proposed.

Rejected Alternatives

1. Deprecate the "modern" option during the proposed rollout and remove it in the next major release, Kafka 5.0

...