DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
2) A new option added for each affacted tool scrtips:
- Name: modern
- Description: Deprecated. This configuration relates to whether to use the property loading priority proposed by this KIP. It will be removed 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
...
Add the helper methods in 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 a the demonstration of the key helper methodmethods:
| Code Block | ||||
|---|---|---|---|---|
| ||||
/**
* 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 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<>();
if (defaultMap != null) {
map.putAll(defaultMap);
}
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;
}
/**
* Merge the option into {@code map} for the given {@code key}, using the following precedence, from highest to lowest:
* 1) Merge an option value into the map if present and not null
* 2) Use the default value {@code defaultValue} if specified
* 3) Otherwise, do nothing
*/
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 as we as for default value passed to the method
if (value == null && defaultValue == null) {
System.err.println("No value specified for option \"" + key + "\" and no default value provided.");
Exit.exit(1);
}
}
if (value == null) {
value = defaultValue;
}
if (value != null) {
map.put(key, value);
}
} |
...
| 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(this, producerConfigOpt, producerPropertyOpt, commandlineMap, defaultMap);
return map;
} |
...
Impact: Existing users who might have configured their settings based on the implementation of the tools, rather than the priority proposed here, would be impacted.
DeprecateDeprecation: 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, and to remove 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 as proposed.
...