DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
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
1) The precedence order of precedence for loading properties into Kafka configurations is as follows:
...
2) A new option added for each affacted all affected tool scrtipsscripts:
- Name: modern
- Description: This configuration relates to determines whether to use enable the property loading priority proposed by 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
...
Proposed Changes
Add the helper methods in 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 is demonstrates the demonstration of the key helper methods:
| Code Block | ||||
|---|---|---|---|---|
| ||||
/**
* MergesMerge multiple configuration sources according byto priority. The merge order is as follows, from highhighest to lowlowest:
* 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 precedence, from highest to lowestlogic:
* 1) Merge anthe option value into the map if the option is present and not null has a value or a default value provided.
* 2) Use {@code valueIfNoRequiredArg} if the default value {@code defaultValue} if specifiedoption 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 defaultValuevalueIfNoRequiredArg)
{
Object value = null;
if (!options.has(spec)) {
if (options.has(spec)) {
return;
}
Object value = options.valueOf(spec);
if (value == null) {
// This can also be null.
// For example if the option is meant to be used without any argument (e.g., --compression-codec specified without a value, and with no default value set in the option as we as for default value passed to the method
if (value == null && defaultValuefrom-latest)
// In that case, valueIfNoRequiredArg acts as the value
if (valueIfNoRequiredArg == null) {
System.err.println("No value specified for option \"" + key + "\" and no default value provided.");
Exit.exit(1);
}
}
Exit.exit(1);
if (value == null) {}
value = defaultValuevalueIfNoRequiredArg;
}
if (value != null) {map.put(key, value);
}
public static void maybeMergeOption(OptionSet options, Map<String, Object> map, String map.put(key, value);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(thisoptions, producerConfigOpt, producerPropertyOpt, commandlineMap, defaultMap);
return map;
} |
...
Impact: Existing users who might have configured their settings based on the implementation of the toolscurrent tool implementations, rather than the proposed priority proposed here, would may be impactedaffected.
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.
Rejected Alternatives
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.
23. Allow Each Tool to Have Its Own Property Loading Precedence
...