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.
Deprecate:
--modern
The order of precedence for loading properties into Kafka configurations is as follows:
Tools that can be configured through both the command line and configuration files 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:
/**
* 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:
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;
} |
Impact: Existing users who might have configured their settings based on the tools, which implementing their own loading order, rather than the priority proposed here, would be impacted.
Unit tests will be added to ensure that the properties are loaded according to the priority order as proposed.
1. 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.
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.