Current state: Under Discussion
Discussion thread: WIP
JIRA:
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 for property loading.
1) The precedence order for loading properties into Kafka configurations is as follows:
2) A new option added for all affected tool scripts:
All the tool scripts that can be configured through either the command line or configuration files, or both, will be adjusted accordingly.
Add the helper methods to CommandLineUtils.java so developers can more easily honor the proposed property loading priority. 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 priority.
The following code demonstrates the key helper methods:
/**
* 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> mergeByPriority(OptionSet options, OptionSpec<String> configOpt, OptionSpec<String> propertyOpt, Map<String, Object> overrides, Map<String, Object> defaultMap) throws IOException {
Map<String, Object> map = new HashMap<>();
// Default values set by tool scripts
if (defaultMap != null) {
map.putAll(defaultMap);
}
// Configuration file
if (configOpt != null && options.has(configOpt)) {
map.putAll(Utils.propsToMap(
Utils.loadProps(options.valueOf(configOpt)))
);
}
// Properties passed as key=value pairs via command line
if (propertyOpt != null && options.has(propertyOpt)) {
map.putAll(Utils.propsToMap(
parseKeyValueArgs(options.valuesOf(propertyOpt)))
);
}
// Command line arguments
if (overrides != null) {
map.putAll(overrides);
}
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.
* 2) Use {@code valueIfNoRequiredArg} if the option is present but its value is null.
* 3) Otherwise, do nothing
*/
public static void maybeMergeOption(OptionSet options, Map<String, Object> map, String key, OptionSpec<?> spec, Object valueIfNoRequiredArg)
{
if (!options.has(spec)) {
return;
}
Object value = options.valueOf(spec);
if (value == null) {
// This can also be null if the option is meant to be used without any argument (e.g., --from-latest)
// In that case, valueIfNoRequiredArg acts as the value
if (valueIfNoRequiredArg == null) {
System.err.println("No value specified for option \"" + key + "\" and no value provided.");
Exit.exit(1);
}
value = valueIfNoRequiredArg;
}
map.put(key, value);
}
public static void maybeMergeOption(OptionSet options, Map<String, Object> map, String key, OptionSpec<?> spec)
{
maybeMergeOption(options, map, key, spec, null);
} |
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;
} |
Impact: Existing users who have configured their settings based on the current tool implementations, rather than the proposed priority, may be affected.
Deprecation: Adjusting property loading priority 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 the next major release, Kafka 5.0. Concurrently, the option will be deprecated in Kafka 5.0 and subsequently removed in Kafka 6.0.
Unit tests will be added to ensure that the properties are loaded according to the priority order as proposed.
1. Deprecate the "modern" option during the proposed rollout and remove it in the next major release, Kafka 5.0
Reason for Rejection: Users who have enabled the modern option in scripts will face the fatal error once Kafka 5.0 is released.
2. Preferring Configuration Files Over Command Line Arguments
Reason for Rejection: Intuitively, users are more likely to expect properties set via command line arguments to take precedence over those defined in configuration files.
3. Allow Each Tool to Have Its Own Property Loading Precedence
Reason for Rejection: While this would provide more flexibility for Kafka developers, it would introduce undesirable complexity to the configuration logic, be confusing to users, and increase the potential for bugs.