Current state: Under Discussion
Discussion thread: WIP
JIRA:
Kafka allows users to configure how tool scripts behave 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 error-prone.
This KIP aims to establish a clear and uniform precedence order for property loading.
Furthermore, the validation of required arguments, such as --bootstrap-server and --bootstrap-controller, should consider multiple sources. For most existing tool scripts, these arguments are only validated when provided via the command line. However, they could also be specified through properties passed as key=value pairs via the command line or configuration files.
1) The precedence order for loading properties into Kafka configurations is as follows:
2) A new option is added for all affected tool scripts not following the proposed precedence:
Required arguments (e.g., --bootstrap-server and --bootstrap-controller for now) should be validated after considering values from the following sources:
An error should only be raised if the arguments are missing or invalid after checking all three sources.
To ensure that all tool scripts correctly honor the proposed property loading precedence, introduce the helper methods in CommandLineUtils.java and enforce their use across all affected tools. This prevents each script from implementing its own logic and ensures consistent behavior.
Since the codebase currently uses three different argument parsers (joptsimple, argparse4j, and AbstractConnectCli), the helper methods are designed to be simple and generic, without depending on any specific external library.
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> mergePropertiesWithPrecedence(
Map<String, Object> commandLineMap,
Map<String, Object> commandLineKeyValMap,
Map<String, Object> configMap,
Map<String, Object> toolDefaultMap
) {
Map<String, Object> map = new HashMap<>();
// Default values set by tool scripts and default values of command line arguments
if (toolDefaultMap != null) {
map.putAll(toolDefaultMap);
}
// Configuration file
if (configMap != null) {
map.putAll(configMap);
}
// Properties passed as key=value pairs via command line
if (commandLineKeyValMap != null) {
map.putAll(commandLineKeyValMap);
}
// Command line arguments
if (commandLineMap != null) {
map.putAll(commandLineMap);
}
return map;
} |
Map<String, Object> readerProps() throws IOException {
Map<String, Object> commandLineMap = new HashMap<>();
commandLineMap.put("topic", options.valueOf(topicOpt));
Map<String, Object> configMap = new HashMap<>();
if (options.has(readerConfigOpt)) {
configMap.putAll(propsToStringMap(loadProps(options.valueOf(readerConfigOpt))));
}
Map<String, Object> commandLineKeyValMap = new HashMap<>();
if (options.has(readerPropertyOpt)) {
commandLineKeyValMap.putAll(propsToStringMap(parseKeyValueArgs(options.valuesOf(readerPropertyOpt))));
}
return mergePropertiesWithPrecedence(commandLineMap, commandLineKeyValMap, configMap, null);
}
Map<String, Object> producerProps() throws IOException {
// Prepare the map from command line arguments
Map<String, Object> commandLineMap = new HashMap<>();
commandLineMap.put(BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOpt));
commandLineMap.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
if (options.has(sendTimeoutOpt)) commandLineMap.put(LINGER_MS_CONFIG, options.valueOf(sendTimeoutOpt).toString());
if (options.has(requestRequiredAcksOpt)) commandLineMap.put(ACKS_CONFIG, options.valueOf(requestRequiredAcksOpt).toString());
if (options.has(requestTimeoutMsOpt)) commandLineMap.put(REQUEST_TIMEOUT_MS_CONFIG, options.valueOf(requestTimeoutMsOpt).toString());
if (options.has(messageSendMaxRetriesOpt)) commandLineMap.put(RETRIES_CONFIG, options.valueOf(messageSendMaxRetriesOpt).toString());
if (options.has(retryBackoffMsOpt)) commandLineMap.put(RETRY_BACKOFF_MS_CONFIG, options.valueOf(retryBackoffMsOpt).toString());
if (options.has(socketBufferSizeOpt)) commandLineMap.put(SEND_BUFFER_CONFIG, options.valueOf(socketBufferSizeOpt).toString());
if (options.has(maxMemoryBytesOpt)) commandLineMap.put(BUFFER_MEMORY_CONFIG, options.valueOf(maxMemoryBytesOpt).toString());
if (options.has(batchSizeOpt)) commandLineMap.put(BATCH_SIZE_CONFIG, options.valueOf(batchSizeOpt).toString());
if (options.has(maxPartitionMemoryBytesOpt)) commandLineMap.put(BATCH_SIZE_CONFIG, options.valueOf(maxPartitionMemoryBytesOpt).toString());
if (options.has(metadataExpiryMsOpt)) commandLineMap.put(METADATA_MAX_AGE_CONFIG, options.valueOf(metadataExpiryMsOpt).toString());
if (options.has(maxBlockMsOpt)) commandLineMap.put(MAX_BLOCK_MS_CONFIG, options.valueOf(maxBlockMsOpt).toString());
// Properties passed as key=value pairs via command line
Map<String, Object> commandLineKeyValMap = new HashMap<>();
if (options.has(commandPropertyOpt)) {
commandLineKeyValMap.putAll(Utils.propsToStringMap(
parseKeyValueArgs(options.valuesOf(commandPropertyOpt))
));
}
// Configuration files
Map<String, Object> configMap = new HashMap<>();
if (options.has(commandConfigOpt)) {
configMap.putAll(Utils.propsToStringMap(
Utils.loadProps(options.valueOf(commandConfigOpt))
));
}
// Default values set by tool scripts and default values of command line arguments
Map<String, Object> toolDefaultMap = new HashMap<>();
toolDefaultMap.put(CLIENT_ID_CONFIG, "console-producer");
toolDefaultMap.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
toolDefaultMap.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
// Since all the options below have default values, we don't need to check for null
toolDefaultMap.put(LINGER_MS_CONFIG, options.valuesOf(sendTimeoutOpt).toString());
toolDefaultMap.put(ACKS_CONFIG, options.valuesOf(requestRequiredAcksOpt).toString());
toolDefaultMap.put(REQUEST_TIMEOUT_MS_CONFIG, options.valuesOf(requestTimeoutMsOpt).toString());
toolDefaultMap.put(RETRIES_CONFIG, options.valuesOf(messageSendMaxRetriesOpt).toString());
toolDefaultMap.put(RETRY_BACKOFF_MS_CONFIG, options.valuesOf(retryBackoffMsOpt).toString());
toolDefaultMap.put(SEND_BUFFER_CONFIG, options.valuesOf(socketBufferSizeOpt).toString());
toolDefaultMap.put(BUFFER_MEMORY_CONFIG, options.valuesOf(maxMemoryBytesOpt).toString());
toolDefaultMap.put(BATCH_SIZE_CONFIG, options.valuesOf(batchSizeOpt).toString());
toolDefaultMap.put(BATCH_SIZE_CONFIG, options.valuesOf(maxPartitionMemoryBytesOpt).toString());
toolDefaultMap.put(METADATA_MAX_AGE_CONFIG, options.valuesOf(metadataExpiryMsOpt).toString());
toolDefaultMap.put(MAX_BLOCK_MS_CONFIG, options.valuesOf(maxBlockMsOpt).toString());
return mergePropertiesWithPrecedence(commandLineMap, commandLineKeyValMap, configMap, toolDefaultMap);
} |
The following table lists the tools that need to be updated based on the proposed changes.
All of the tools listed below should use the helper methods to ensure they honor the proposed precedence.
| Tools | *New "modern" Option for Deprecation | Validating Required Arguments from Multiple Sources | Previous Jira/Email Discussion | Note |
|---|---|---|---|---|
| kafka-acls.sh | --bootstrap-server --bootstrap-controller | |||
kafka-broker-api-versions.sh | --bootstrap-server | |||
| kafka-client-metrics.sh | --bootstrap-server | |||
| kafka-cluster.sh | --bootstrap-server --bootstrap-controller | |||
| kafka-configs.sh | --bootstrap-server --bootstrap-controller | |||
| kafka-console-consumer.sh | For formatter: KEY_DESERIALIZER_CLASS_CONFIG VALUE_DESERIALIZER_CLASS_CONFIG | --bootstrap-server --from-beginning (AUTO_OFFSET_RESET_CONFIG) --group (GROUP_ID_CONFIG) is validated, but not following the proposed flow | ||
| kafka-console-producer.sh | KEY_SERIALIZER_CLASS_CONFIG VALUE_SERIALIZER_CLASS_CONFIG COMPRESSION_TYPE_CONFIG | --bootstrap-server | ||
| kafka-console-share-consumer.sh | For formatter: KEY_DESERIALIZER_CLASS_CONFIG VALUE_DESERIALIZER_CLASS_CONFIG | --bootstrap-server --group (GROUP_ID_CONFIG) is validated, but not following the proposed flow | ||
| kafka-consumer-groups.sh | --bootstrap-server | |||
| kafka-consumer-perf-test.sh | GROUP_ID_CONFIG RECEIVE_BUFFER_CONFIG MAX_PARTITION_FETCH_BYTES_CONFIG AUTO_OFFSET_RESET_CONFIG KEY_DESERIALIZER_CLASS_CONFIG VALUE_DESERIALIZER_CLASS_CONFIG CHECK_CRCS_CONFIG | --bootstrap-server | ||
| kafka-delegation-tokens.sh | --bootstrap-server | |||
| kafka-delete-records.sh | --bootstrap-server | |||
kafka-e2e-latency.sh | For consumer: GROUP_ID_CONFIG ENABLE_AUTO_COMMIT_CONFIG AUTO_OFFSET_RESET_CONFIG KEY_DESERIALIZER_CLASS_CONFIG VALUE_DESERIALIZER_CLASS_CONFIG FETCH_MAX_WAIT_MS_CONFIG For producer: LINGER_MS_CONFIG MAX_BLOCK_MS_CONFIG ACKS_CONFIG KEY_SERIALIZER_CLASS_CONFIG VALUE_SERIALIZER_CLASS_CONFIG | --bootstrap-server --producer-acks (ACKS_CONFIG) | ||
| kafka-features.sh | --bootstrap-server --bootstrap-controller | |||
kafka-get-offsets.sh | CLIENT_ID_CONFIG | --bootstrap-server | ||
kafka-groups.sh | --bootstrap-server | |||
kafka-leader-election.sh | --bootstrap-server | |||
kafka-log-dirs.sh | --bootstrap-server | |||
kafka-metadata-quorum.sh | --bootstrap-server --bootstrap-controller | |||
kafka-producer-perf-test.sh | KEY_SERIALIZER_CLASS_CONFIG VALUE_SERIALIZER_CLASS_CONFIG | |||
kafka-reassign-partitions.sh | --bootstrap-server --bootstrap-controller | |||
kafka-share-consumer-perf-test.sh | GROUP_ID_CONFIG RECEIVE_BUFFER_CONFIG MAX_PARTITION_FETCH_BYTES_CONFIG AUTO_OFFSET_RESET_CONFIG KEY_DESERIALIZER_CLASS_CONFIG VALUE_DESERIALIZER_CLASS_CONFIG CHECK_CRCS_CONFIG | --bootstrap-server | ||
kafka-share-groups.sh | --bootstrap-server | |||
kafka-streams-application-reset.sh | --bootstrap-server | This tool script is the only one with a default bootstrap-server; consider aligning it with the others. | ||
kafka-streams-groups.sh | --bootstrap-server | |||
kafka-topics.sh | --bootstrap-server | |||
kafka-transactions.sh | --bootstrap-server | |||
kafka-verifiable-consumer.sh | GROUP_PROTOCOL_CONFIG GROUP_REMOTE_ASSIGNOR_CONFIG PARTITION_ASSIGNMENT_STRATEGY_CONFIG --group-id (GROUP_ID_CONFIG) ENABLE_AUTO_COMMIT_CONFIG AUTO_OFFSET_RESET_CONFIG | --bootstrap-server --group-id (GROUP_ID_CONFIG) | ||
kafka-verifiable-producer.sh | KEY_SERIALIZER_CLASS_CONFIG VALUE_SERIALIZER_CLASS_CONFIG ACKS_CONFIG RETRIES_CONFIG | --bootstrap-server | ||
kafka-verifiable-share-consumer.sh | --bootstrap-server --group-id (GROUP_ID_CONFIG) |
*New "modern" Option for Deprecation - Indicates that the tool script currently has properties that do not follow the proposed precedence.
Impact: Existing users who have configured their settings based on the current tool implementations, rather than the proposed precedence, may be affected.
Deprecation: Adjusting property loading 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 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.
Unit tests will be added to ensure that the properties are loaded according to the precedence 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 fatal errors 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.