DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
Status
Current state: Under Discussion
...
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 bugerror-prone. This KIP aims to establish a clear and uniform precedence order for property loading.
Furthermore, the validation of required arguments - currently , such as --bootstrap-server and --bootstrap-controller - , should consider multiple sources. For existing tool scripts, these arguments are only validated when provided via the command line "--bootstrap-server" or "--bootstrap-controller". However, they could also be specified through properties passed as key=value pairs via the command line or configuration files.
...
- 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 ConfigDefKafka components
2) A new option is added for all affected tool scripts not following the proposed precedence:
- Name: modern
- Description: This configuration determines whether to enable the property loading 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.
Validation of required arguments
...
Validating Required Arguments from Multiple Sources
Required arguments (e.g., --bootstrap-server and --bootstrap-controller for now) should be validated after considering values from the following sources:
- Command line arguments
- Properties passed as key=value pairs via command line
- Configuration files
...
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 toolsaffected tools. This prevents each script from implementing its own logic and ensures consistent behavior.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
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()); // We currently have 2 options to set the batch.size value. We'll deprecate/remove one of them in KIP-717. if (options.has(batchSizeOpt)) commandLineMap.put(BATCH_SIZE_CONFIG, options.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); } |
...
Required Updates to Existing Tools
The following table shows the existing and proposed changes. The changes are in GREENlists 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 |
| Jira | ||||||
|---|---|---|---|---|---|---|
|
--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) |
| Jira | ||||||
|---|---|---|---|---|---|---|
|
kafka-broker-api-versions.sh
*New "modern" Option for Deprecation - This means Indicates that the tool script currently has properties that don’t do not follow the proposed precedence.
...
Reason for Rejection: Users who have enabled the modern option in scripts will face the fatal error errors once Kafka 5.0 is released.
...