DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
Status
Current state: Under Discussion
...
JIRA:
| Jira | ||||||
|---|---|---|---|---|---|---|
|
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:
- Command line arguments
- Properties passed as key=value pairs via command line
- Configuration files
- Default values set by tool scripts and default values of command line arguments
- 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 | ||||
|---|---|---|---|---|
| ||||
/**
* 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 | ||||
|---|---|---|---|---|
| ||||
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
...