KAFKA-19623: Implement KIP-1147 for console producer/consumer/share-consumer. (#20479)
*What*
https://issues.apache.org/jira/browse/KAFKA-19623
- The PR implements KIP-1147
(https://cwiki.apache.org/confluence/display/KAFKA/KIP-1147%3A+Improve+consistency+of+command-line+arguments)
for the console tools i.e. `ConsoleProducer`, `ConsoleConsumer` and
`ConsoleShareConsumer`.
- Currently the previous names for the options are still usable but
there will be warning message stating those are deprecated and will be
removed in a future version.
- I have added unit tests and also manually verified using the console
tools that things are working as expected.
Reviewers: Andrew Schofield <aschofield@confluent.io>, Jhen-Yung Hsu
<jhenyunghsu@gmail.com>, Jimmy Wang
<48462172+JimmyWang6@users.noreply.github.com>
diff --git a/docker/examples/README.md b/docker/examples/README.md
index e76247b..162e27c 100644
--- a/docker/examples/README.md
+++ b/docker/examples/README.md
@@ -147,7 +147,7 @@
- To produce messages using client scripts (Ensure that java version >= 17):
```
# Run from root of the repo
- $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9093 --producer.config ./docker/examples/fixtures/client-secrets/client-ssl.properties
+ $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9093 --command-config ./docker/examples/fixtures/client-secrets/client-ssl.properties
```
- File Input:
- Here ssl configs are provided via file input.
@@ -167,7 +167,7 @@
- To produce messages using client scripts (Ensure that java version >= 17):
```
# Run from root of the repo
- $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9093 --producer.config ./docker/examples/fixtures/client-secrets/client-ssl.properties
+ $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9093 --command-config ./docker/examples/fixtures/client-secrets/client-ssl.properties
```
Multi Node Cluster
@@ -219,7 +219,7 @@
- To produce messages using client scripts (Ensure that java version >= 17):
```
# Run from root of the repo
- $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:29093 --producer.config ./docker/examples/fixtures/client-secrets/client-ssl.properties
+ $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:29093 --command-config ./docker/examples/fixtures/client-secrets/client-ssl.properties
```
- Isolated:
- Examples are present in `docker-compose-files/cluster/isolated` directory.
@@ -258,7 +258,7 @@
- To produce messages using client scripts (Ensure that java version >= 17):
```
# Run from root of the repo
- $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:29093 --producer.config ./docker/examples/fixtures/client-secrets/client-ssl.properties
+ $ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:29093 --command-config ./docker/examples/fixtures/client-secrets/client-ssl.properties
```
- Note that the examples are meant to be tried one at a time, make sure you close an example server before trying out the other to avoid conflicts.
diff --git a/docker/test/docker_sanity_test.py b/docker/test/docker_sanity_test.py
index d2135fb..0d21bf4 100644
--- a/docker/test/docker_sanity_test.py
+++ b/docker/test/docker_sanity_test.py
@@ -65,7 +65,7 @@
subprocess.run(["bash", "-c", " ".join(command)])
def consume_message(self, topic, consumer_config):
- command = [f"{self.FIXTURES_DIR}/{constants.KAFKA_CONSOLE_CONSUMER}", "--topic", topic, "--property", "'print.key=true'", "--property", "'key.separator=:'", "--from-beginning", "--max-messages", "1", "--timeout-ms", f"{constants.CLIENT_TIMEOUT}"]
+ command = [f"{self.FIXTURES_DIR}/{constants.KAFKA_CONSOLE_CONSUMER}", "--topic", topic, "--formatter-property", "'print.key=true'", "--formatter-property", "'key.separator=:'", "--from-beginning", "--max-messages", "1", "--timeout-ms", f"{constants.CLIENT_TIMEOUT}"]
command.extend(consumer_config)
message = subprocess.check_output(["bash", "-c", " ".join(command)])
return message.decode("utf-8").strip()
@@ -93,9 +93,9 @@
errors.append(constants.BROKER_METRICS_ERROR_PREFIX + str(e))
return errors
- producer_config = ["--bootstrap-server", "localhost:9092", "--property", "client.id=host"]
+ producer_config = ["--bootstrap-server", "localhost:9092", "--command-property", "client.id=host"]
self.produce_message(constants.BROKER_METRICS_TEST_TOPIC, producer_config, "key", "message")
- consumer_config = ["--bootstrap-server", "localhost:9092", "--property", "auto.offset.reset=earliest"]
+ consumer_config = ["--bootstrap-server", "localhost:9092", "--command-property", "auto.offset.reset=earliest"]
message = self.consume_message(constants.BROKER_METRICS_TEST_TOPIC, consumer_config)
try:
self.assertEqual(message, "key:message")
@@ -129,13 +129,13 @@
return errors
producer_config = ["--bootstrap-server", ssl_broker_port,
- "--producer.config", f"{self.FIXTURES_DIR}/{constants.SSL_CLIENT_CONFIG}"]
+ "--command-config", f"{self.FIXTURES_DIR}/{constants.SSL_CLIENT_CONFIG}"]
self.produce_message(topic, producer_config, "key", "message")
consumer_config = [
"--bootstrap-server", ssl_broker_port,
- "--property", "auto.offset.reset=earliest",
- "--consumer.config", f"{self.FIXTURES_DIR}/{constants.SSL_CLIENT_CONFIG}",
+ "--command-property", "auto.offset.reset=earliest",
+ "--command-config", f"{self.FIXTURES_DIR}/{constants.SSL_CLIENT_CONFIG}",
]
message = self.consume_message(topic, consumer_config)
try:
@@ -155,7 +155,7 @@
errors.append(constants.BROKER_RESTART_ERROR_PREFIX + str(e))
return errors
- producer_config = ["--bootstrap-server", "localhost:9092", "--property", "client.id=host"]
+ producer_config = ["--bootstrap-server", "localhost:9092", "--command-property", "client.id=host"]
self.produce_message(constants.BROKER_RESTART_TEST_TOPIC, producer_config, "key", "message")
print("Stopping Container")
@@ -163,7 +163,7 @@
print("Resuming Container")
self.resume_container()
- consumer_config = ["--bootstrap-server", "localhost:9092", "--property", "auto.offset.reset=earliest"]
+ consumer_config = ["--bootstrap-server", "localhost:9092", "--command-property", "auto.offset.reset=earliest"]
message = self.consume_message(constants.BROKER_RESTART_TEST_TOPIC, consumer_config)
try:
self.assertEqual(message, "key:message")
diff --git a/docs/security.html b/docs/security.html
index 5940fc3..9364a05 100644
--- a/docs/security.html
+++ b/docs/security.html
@@ -510,8 +510,8 @@
</ol>
<br>
Examples using console-producer and console-consumer:
- <pre><code class="language-bash">$ bin/kafka-console-producer.sh --bootstrap-server localhost:9093 --topic test --producer.config client-ssl.properties
-$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test --consumer.config client-ssl.properties</code></pre>
+ <pre><code class="language-bash">$ bin/kafka-console-producer.sh --bootstrap-server localhost:9093 --topic test --command-config client-ssl.properties
+$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test --command-config client-ssl.properties</code></pre>
</li>
</ol>
<h3 class="anchor-heading"><a id="security_sasl" class="anchor-link"></a><a href="#security_sasl">7.4 Authentication using SASL</a></h3>
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
index 7037ce1..aa28372 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
@@ -1122,15 +1122,15 @@
final String[] args = new String[] {
"--bootstrap-server", CLUSTER.bootstrapServers(),
"--from-beginning",
- "--property", "print.key=true",
- "--property", "print.timestamp=" + printTimestamp,
+ "--formatter-property", "print.key=true",
+ "--formatter-property", "print.timestamp=" + printTimestamp,
"--topic", outputTopic,
"--max-messages", String.valueOf(numMessages),
- "--property", "key.deserializer=" + keyDeserializer.getClass().getName(),
- "--property", "value.deserializer=" + valueDeserializer.getClass().getName(),
- "--property", "key.separator=" + keySeparator,
- "--property", "key.deserializer." + TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS + "=" + Serdes.serdeFrom(innerClass).getClass().getName(),
- "--property", "key.deserializer.window.size.ms=500",
+ "--formatter-property", "key.deserializer=" + keyDeserializer.getClass().getName(),
+ "--formatter-property", "value.deserializer=" + valueDeserializer.getClass().getName(),
+ "--formatter-property", "key.separator=" + keySeparator,
+ "--formatter-property", "key.deserializer." + TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS + "=" + Serdes.serdeFrom(innerClass).getClass().getName(),
+ "--formatter-property", "key.deserializer.window.size.ms=500",
};
ConsoleConsumer.run(new ConsoleConsumerOptions(args));
diff --git a/tests/kafkatest/services/console_consumer.py b/tests/kafkatest/services/console_consumer.py
index 9755faa..0f95a0f 100644
--- a/tests/kafkatest/services/console_consumer.py
+++ b/tests/kafkatest/services/console_consumer.py
@@ -88,7 +88,7 @@
jaas_override_variables A dict of variables to be used in the jaas.conf template file
kafka_opts_override Override parameters of the KAFKA_OPTS environment variable
client_prop_file_override Override client.properties file used by the consumer
- consumer_properties A dict of values to pass in as --consumer-property key=value
+ consumer_properties A dict of values to pass in as --command-property key=value. For versions older than KAFKA_4_2_0, these will be passed as --consumer-property key=value
"""
JmxMixin.__init__(self, num_nodes=num_nodes, jmx_object_names=jmx_object_names, jmx_attributes=(jmx_attributes or []),
root=ConsoleConsumer.PERSISTENT_ROOT)
@@ -163,8 +163,11 @@
"export KAFKA_LOG4J_OPTS=\"%(log4j_param)s%(log4j_config)s\"; " \
"export KAFKA_OPTS=%(kafka_opts)s; " \
"%(console_consumer)s " \
- "--topic %(topic)s " \
- "--consumer.config %(config_file)s " % args
+ "--topic %(topic)s " % args
+
+ version = get_version(node)
+ command_config_arg = "--command-config" if version.supports_command_config() else "--consumer.config"
+ cmd += "%s %s" % (command_config_arg, args['config_file'])
cmd += " --bootstrap-server %(broker_list)s" % args
cmd += " --isolation-level %s" % self.isolation_level
@@ -176,14 +179,15 @@
# This will be added in the properties file instead
cmd += " --timeout-ms %s" % self.consumer_timeout_ms
+ formatter_property_arg = "--formatter-property" if version.supports_formatter_property else "--property"
if self.print_timestamp:
- cmd += " --property print.timestamp=true"
+ cmd += " %s print.timestamp=true" % formatter_property_arg
if self.print_key:
- cmd += " --property print.key=true"
+ cmd += " %s print.key=true" % formatter_property_arg
if self.print_partition:
- cmd += " --property print.partition=true"
+ cmd += " %s print.partition=true" % formatter_property_arg
# LoggingMessageFormatter was introduced after 0.9
if node.version > LATEST_3_7:
@@ -194,9 +198,10 @@
if self.enable_systest_events:
cmd += " --enable-systest-events"
+ command_property_arg = "--command-property" if version.supports_command_property() else "--consumer-property"
if self.consumer_properties is not None:
for k, v in self.consumer_properties.items():
- cmd += " --consumer-property %s=%s" % (k, v)
+ cmd += " %s %s=%s" % (command_property_arg, k, v)
cmd += " 2>> %(stderr)s | tee -a %(stdout)s &" % args
return cmd
diff --git a/tests/kafkatest/services/console_share_consumer.py b/tests/kafkatest/services/console_share_consumer.py
index 03fbaea..2d7da50 100644
--- a/tests/kafkatest/services/console_share_consumer.py
+++ b/tests/kafkatest/services/console_share_consumer.py
@@ -21,7 +21,7 @@
from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
from kafkatest.services.monitor.jmx import JmxMixin
-from kafkatest.version import DEV_BRANCH, LATEST_4_1
+from kafkatest.version import DEV_BRANCH, LATEST_4_1, get_version
from kafkatest.services.kafka.util import fix_opts_for_new_jvm, get_log4j_config_param, get_log4j_config_for_tools
"""
@@ -84,7 +84,7 @@
jaas_override_variables A dict of variables to be used in the jaas.conf template file
kafka_opts_override Override parameters of the KAFKA_OPTS environment variable
client_prop_file_override Override client.properties file used by the consumer
- share_consumer_properties A dict of values to pass in as --consumer-property key=value
+ share_consumer_properties A dict of values to pass in as --command-property key=value. For versions older than KAFKA_4_2_0, these will be passed as --consumer-property key=value
"""
JmxMixin.__init__(self, num_nodes=num_nodes, jmx_object_names=jmx_object_names, jmx_attributes=(jmx_attributes or []),
root=ConsoleShareConsumer.PERSISTENT_ROOT)
@@ -156,31 +156,36 @@
"export KAFKA_LOG4J_OPTS=\"%(log4j_param)s%(log4j_config)s\"; " \
"export KAFKA_OPTS=%(kafka_opts)s; " \
"%(console_share_consumer)s " \
- "--topic %(topic)s " \
- "--consumer-config %(config_file)s " % args
+ "--topic %(topic)s " % args
+
+ version = get_version(node)
+ command_config_arg = "--command-config" if version.supports_command_config() else "--consumer-config"
+ cmd += "%s %s" % (command_config_arg, args['config_file'])
cmd += " --bootstrap-server %(broker_list)s" % args
if self.share_consumer_timeout_ms is not None:
# This will be added in the properties file instead
cmd += " --timeout-ms %s" % self.share_consumer_timeout_ms
+ formatter_property_arg = "--formatter-property" if version.supports_formatter_property else "--property"
if self.print_timestamp:
- cmd += " --property print.timestamp=true"
+ cmd += " %s print.timestamp=true" % formatter_property_arg
if self.print_key:
- cmd += " --property print.key=true"
+ cmd += " %s print.key=true" % formatter_property_arg
if self.print_partition:
- cmd += " --property print.partition=true"
+ cmd += " %s print.partition=true" % formatter_property_arg
cmd += " --formatter org.apache.kafka.tools.consumer.LoggingMessageFormatter"
if self.enable_systest_events:
cmd += " --enable-systest-events"
+ command_property_arg = "--command-property" if version.supports_command_property() else "--consumer-property"
if self.share_consumer_properties is not None:
for k, v in self.share_consumer_properties.items():
- cmd += " --consumer-property %s=%s" % (k, v)
+ cmd += " %s %s=%s" % (command_property_arg, k, v)
cmd += " 2>> %(stderr)s | tee -a %(stdout)s &" % args
return cmd
diff --git a/tests/kafkatest/version.py b/tests/kafkatest/version.py
index e4d600a..8114d51 100644
--- a/tests/kafkatest/version.py
+++ b/tests/kafkatest/version.py
@@ -104,6 +104,20 @@
# - For older versions, continue using --producer.config or --consumer.config
return self >= V_4_2_0
+ def supports_command_property(self):
+ # According to KIP-1147, --producer-property and --consumer-property have been deprecated and will be removed in future versions
+ # For backward compatibility, we select the configuration based on node version:
+ # - For versions 4.2.0 and above, use --command-property
+ # - For older versions, continue using --producer-property or --consumer-property
+ return self >= V_4_2_0
+
+ def supports_formatter_property(self):
+ # According to KIP-1147, --property has been deprecated and will be removed in future versions
+ # For backward compatibility, we select the configuration based on node version:
+ # - For versions 4.2.0 and above, use --formatter-property
+ # - For older versions, continue using --property
+ return self >= V_4_2_0
+
def get_version(node=None):
"""Return the version attached to the given node.
Default to DEV_BRANCH if node or node.version is undefined (aka None)
diff --git a/tools/src/main/java/org/apache/kafka/tools/ConsoleProducer.java b/tools/src/main/java/org/apache/kafka/tools/ConsoleProducer.java
index 93e81bd..3e9e34f 100644
--- a/tools/src/main/java/org/apache/kafka/tools/ConsoleProducer.java
+++ b/tools/src/main/java/org/apache/kafka/tools/ConsoleProducer.java
@@ -131,8 +131,12 @@
private final OptionSpec<Integer> socketBufferSizeOpt;
private final OptionSpec<String> propertyOpt;
private final OptionSpec<String> readerConfigOpt;
+ @Deprecated(since = "4.2", forRemoval = true)
private final OptionSpec<String> producerPropertyOpt;
+ private OptionSpec<String> commandPropertyOpt;
+ @Deprecated(since = "4.2", forRemoval = true)
private final OptionSpec<String> producerConfigOpt;
+ private OptionSpec<String> commandConfigOpt;
public ConsoleProducerOptions(String[] args) {
super(args);
@@ -250,11 +254,20 @@
.withRequiredArg()
.describedAs("config file")
.ofType(String.class);
- producerPropertyOpt = parser.accepts("producer-property", "A mechanism to pass user-defined properties in the form key=value to the producer. ")
+ producerPropertyOpt = parser.accepts("producer-property", "(DEPRECATED) A mechanism to pass user-defined properties in the form key=value to the producer." +
+ "This option will be removed in a future version. Use --command-property instead.")
.withRequiredArg()
.describedAs("producer_prop")
.ofType(String.class);
- producerConfigOpt = parser.accepts("producer.config", "Producer config properties file. Note that " + producerPropertyOpt + " takes precedence over this config.")
+ commandPropertyOpt = parser.accepts("command-property", "A mechanism to pass user-defined properties in the form key=value to the producer.")
+ .withRequiredArg()
+ .describedAs("producer_prop")
+ .ofType(String.class);
+ producerConfigOpt = parser.accepts("producer.config", "(DEPRECATED) Producer config properties file. Note that " + commandPropertyOpt + " takes precedence over this config. This option will be removed in a future version. Use --command-config instead.")
+ .withRequiredArg()
+ .describedAs("config file")
+ .ofType(String.class);
+ commandConfigOpt = parser.accepts("command-config", "Producer config properties file. Note that " + commandPropertyOpt + " takes precedence over this config.")
.withRequiredArg()
.describedAs("config file")
.ofType(String.class);
@@ -273,6 +286,23 @@
CommandLineUtils.checkRequiredArgs(parser, options, topicOpt);
+ if (options.has(commandConfigOpt) && options.has(producerConfigOpt)) {
+ CommandLineUtils.printUsageAndExit(parser, "Options --command-config and --producer.config cannot be specified together.");
+ }
+ if (options.has(commandPropertyOpt) && options.has(producerPropertyOpt)) {
+ CommandLineUtils.printUsageAndExit(parser, "Options --command-property and --producer-property cannot be specified together.");
+ }
+
+ if (options.has(producerPropertyOpt)) {
+ System.out.println("Warning: --producer-property is deprecated and will be removed in a future version. Use --command-property instead.");
+ commandPropertyOpt = producerPropertyOpt;
+ }
+
+ if (options.has(producerConfigOpt)) {
+ System.out.println("Warning: --producer.config is deprecated and will be removed in a future version. Use --command-config instead.");
+ commandConfigOpt = producerConfigOpt;
+ }
+
try {
ToolsUtils.validateBootstrapServer(options.valueOf(bootstrapServerOpt));
} catch (IllegalArgumentException e) {
@@ -314,11 +344,11 @@
Properties producerProps() throws IOException {
Properties props = new Properties();
- if (options.has(producerConfigOpt)) {
- props.putAll(loadProps(options.valueOf(producerConfigOpt)));
+ if (options.has(commandConfigOpt)) {
+ props.putAll(loadProps(options.valueOf(commandConfigOpt)));
}
- props.putAll(parseKeyValueArgs(options.valuesOf(producerPropertyOpt)));
+ props.putAll(parseKeyValueArgs(options.valuesOf(commandPropertyOpt)));
props.put(BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOpt));
props.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptions.java b/tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptions.java
index fe33bfe..abe6322 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptions.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptions.java
@@ -48,7 +48,9 @@
private final OptionSpec<Integer> partitionIdOpt;
private final OptionSpec<String> offsetOpt;
private final OptionSpec<String> messageFormatterOpt;
- private final OptionSpec<String> messageFormatterArgOpt;
+ @Deprecated(since = "4.2", forRemoval = true)
+ private final OptionSpec<String> messageFormatterArgOptDeprecated;
+ private OptionSpec<String> messageFormatterArgOpt;
private final OptionSpec<String> messageFormatterConfigOpt;
private final OptionSpec<?> resetBeginningOpt;
private final OptionSpec<Integer> maxMessagesOpt;
@@ -66,6 +68,7 @@
private final long timeoutMs;
private final MessageFormatter formatter;
+ @SuppressWarnings("deprecation")
public ConsoleConsumerOptions(String[] args) throws IOException {
super(args);
topicOpt = parser.accepts("topic", "The topic to consume on.")
@@ -87,11 +90,23 @@
.describedAs("consume offset")
.ofType(String.class)
.defaultsTo("latest");
- OptionSpec<String> consumerPropertyOpt = parser.accepts("consumer-property", "A mechanism to pass user-defined properties in the form key=value to the consumer.")
+ @Deprecated(since = "4.2", forRemoval = true)
+ OptionSpec<String> consumerPropertyOpt = parser.accepts("consumer-property", "(DEPRECATED) A mechanism to pass user-defined properties in the form key=value to the consumer. " +
+ "This option will be removed in a future version. Use --command-property instead.")
.withRequiredArg()
.describedAs("consumer_prop")
.ofType(String.class);
- OptionSpec<String> consumerConfigOpt = parser.accepts("consumer.config", "Consumer config properties file. Note that " + consumerPropertyOpt + " takes precedence over this config.")
+ OptionSpec<String> commandPropertyOpt = parser.accepts("command-property", "A mechanism to pass user-defined properties in the form key=value to the consumer.")
+ .withRequiredArg()
+ .describedAs("consumer_prop")
+ .ofType(String.class);
+ @Deprecated(since = "4.2", forRemoval = true)
+ OptionSpec<String> consumerConfigOpt = parser.accepts("consumer.config", "(DEPRECATED) Consumer config properties file. Note that " + commandPropertyOpt + " takes precedence over this config. " +
+ "This option will be removed in a future version. Use --command-config instead.")
+ .withRequiredArg()
+ .describedAs("config file")
+ .ofType(String.class);
+ OptionSpec<String> commandConfigOpt = parser.accepts("command-config", "Consumer config properties file. Note that " + commandPropertyOpt + " takes precedence over this config.")
.withRequiredArg()
.describedAs("config file")
.ofType(String.class);
@@ -100,7 +115,28 @@
.describedAs("class")
.ofType(String.class)
.defaultsTo(DefaultMessageFormatter.class.getName());
- messageFormatterArgOpt = parser.accepts("property",
+ messageFormatterArgOptDeprecated = parser.accepts("property",
+ "(DEPRECATED) The properties to initialize the message formatter. Default properties include: \n" +
+ " print.timestamp=true|false\n" +
+ " print.key=true|false\n" +
+ " print.offset=true|false\n" +
+ " print.epoch=true|false\n" +
+ " print.partition=true|false\n" +
+ " print.headers=true|false\n" +
+ " print.value=true|false\n" +
+ " key.separator=<key.separator>\n" +
+ " line.separator=<line.separator>\n" +
+ " headers.separator=<line.separator>\n" +
+ " null.literal=<null.literal>\n" +
+ " key.deserializer=<key.deserializer>\n" +
+ " value.deserializer=<value.deserializer>\n" +
+ " header.deserializer=<header.deserializer>\n" +
+ "\nUsers can also pass in customized properties for their formatter; more specifically, users can pass in properties keyed with 'key.deserializer.', 'value.deserializer.' and 'headers.deserializer.' prefixes to configure their deserializers. " +
+ "\nThis option will be removed in a future version. Use --formatter-property instead.")
+ .withRequiredArg()
+ .describedAs("prop")
+ .ofType(String.class);
+ messageFormatterArgOpt = parser.accepts("formatter-property",
"The properties to initialize the message formatter. Default properties include: \n" +
" print.timestamp=true|false\n" +
" print.key=true|false\n" +
@@ -170,11 +206,25 @@
CommandLineUtils.maybePrintHelpOrVersion(this, "This tool helps to read data from Kafka topics and outputs it to standard output.");
checkRequiredArgs();
+ if (options.has(consumerPropertyOpt) && options.has(commandPropertyOpt)) {
+ CommandLineUtils.printUsageAndExit(parser, "Options --consumer-property and --command-property cannot be specified together.");
+ }
+ if (options.has(consumerConfigOpt) && options.has(commandConfigOpt)) {
+ CommandLineUtils.printUsageAndExit(parser, "Options --consumer.config and --command-config cannot be specified together.");
+ }
- Properties consumerPropsFromFile = options.has(consumerConfigOpt)
- ? Utils.loadProps(options.valueOf(consumerConfigOpt))
+ if (options.has(consumerPropertyOpt)) {
+ System.out.println("Option --consumer-property is deprecated and will be removed in a future version. Use --command-property instead.");
+ commandPropertyOpt = consumerPropertyOpt;
+ }
+ if (options.has(consumerConfigOpt)) {
+ System.out.println("Option --consumer.config is deprecated and will be removed in a future version. Use --command-config instead.");
+ commandConfigOpt = consumerConfigOpt;
+ }
+ Properties consumerPropsFromFile = options.has(commandConfigOpt)
+ ? Utils.loadProps(options.valueOf(commandConfigOpt))
: new Properties();
- Properties extraConsumerProps = CommandLineUtils.parseKeyValueArgs(options.valuesOf(consumerPropertyOpt));
+ Properties extraConsumerProps = CommandLineUtils.parseKeyValueArgs(options.valuesOf(commandPropertyOpt));
Set<String> groupIdsProvided = checkConsumerGroup(consumerPropsFromFile, extraConsumerProps);
consumerProps = buildConsumerProps(consumerPropsFromFile, extraConsumerProps, groupIdsProvided);
offset = parseOffset();
@@ -323,6 +373,13 @@
Class<?> messageFormatterClass = Class.forName(options.valueOf(messageFormatterOpt));
formatter = (MessageFormatter) messageFormatterClass.getDeclaredConstructor().newInstance();
+ if (options.has(messageFormatterArgOpt) && options.has(messageFormatterArgOptDeprecated)) {
+ CommandLineUtils.printUsageAndExit(parser, "Options --property and --formatter-property cannot be specified together.");
+ }
+ if (options.has(messageFormatterArgOptDeprecated)) {
+ System.out.println("Option --property is deprecated and will be removed in a future version. Use --formatter-property instead.");
+ messageFormatterArgOpt = messageFormatterArgOptDeprecated;
+ }
Properties formatterArgs = formatterArgs();
Map<String, String> formatterConfigs = new HashMap<>();
for (final String name : formatterArgs.stringPropertyNames()) {
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptions.java b/tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptions.java
index f00407b..3472d07 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptions.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptions.java
@@ -37,7 +37,9 @@
public final class ConsoleShareConsumerOptions extends CommandDefaultOptions {
private final OptionSpec<String> messageFormatterOpt;
private final OptionSpec<String> messageFormatterConfigOpt;
- private final OptionSpec<String> messageFormatterArgOpt;
+ @Deprecated(since = "4.2", forRemoval = true)
+ private final OptionSpec<String> messageFormatterArgOptDeprecated;
+ private OptionSpec<String> messageFormatterArgOpt;
private final OptionSpec<String> keyDeserializerOpt;
private final OptionSpec<String> valueDeserializerOpt;
private final OptionSpec<Integer> maxMessagesOpt;
@@ -52,17 +54,30 @@
private final MessageFormatter formatter;
private final OptionSpec<?> enableSystestEventsLoggingOpt;
+ @SuppressWarnings("deprecation")
public ConsoleShareConsumerOptions(String[] args) throws IOException {
super(args);
topicOpt = parser.accepts("topic", "The topic to consume from.")
.withRequiredArg()
.describedAs("topic")
.ofType(String.class);
- OptionSpec<String> consumerPropertyOpt = parser.accepts("consumer-property", "A mechanism to pass user-defined properties in the form key=value to the consumer.")
+ @Deprecated(since = "4.2", forRemoval = true)
+ OptionSpec<String> consumerPropertyOpt = parser.accepts("consumer-property", "(DEPRECATED) A mechanism to pass user-defined properties in the form key=value to the consumer. " +
+ "This option will be removed in a future version. Use --command-property instead.")
.withRequiredArg()
.describedAs("consumer_prop")
.ofType(String.class);
- OptionSpec<String> consumerConfigOpt = parser.accepts("consumer-config", "Consumer config properties file. Note that " + consumerPropertyOpt + " takes precedence over this config.")
+ OptionSpec<String> commandPropertyOpt = parser.accepts("command-property", "A mechanism to pass user-defined properties in the form key=value to the consumer.")
+ .withRequiredArg()
+ .describedAs("consumer_prop")
+ .ofType(String.class);
+ @Deprecated(since = "4.2", forRemoval = true)
+ OptionSpec<String> consumerConfigOpt = parser.accepts("consumer-config", "(DEPRECATED) Consumer config properties file. Note that " + commandPropertyOpt + " takes precedence over this config. " +
+ "This option will be removed in a future version. Use --command-config instead.")
+ .withRequiredArg()
+ .describedAs("config file")
+ .ofType(String.class);
+ OptionSpec<String> commandConfigOpt = parser.accepts("command-config", "Consumer config properties file. Note that " + commandPropertyOpt + " takes precedence over this config.")
.withRequiredArg()
.describedAs("config file")
.ofType(String.class);
@@ -71,7 +86,29 @@
.describedAs("class")
.ofType(String.class)
.defaultsTo(DefaultMessageFormatter.class.getName());
- messageFormatterArgOpt = parser.accepts("property",
+ messageFormatterArgOptDeprecated = parser.accepts("property",
+ "(DEPRECATED) The properties to initialize the message formatter. Default properties include: \n" +
+ " print.timestamp=true|false\n" +
+ " print.key=true|false\n" +
+ " print.offset=true|false\n" +
+ " print.delivery=true|false\n" +
+ " print.epoch=true|false\n" +
+ " print.partition=true|false\n" +
+ " print.headers=true|false\n" +
+ " print.value=true|false\n" +
+ " key.separator=<key.separator>\n" +
+ " line.separator=<line.separator>\n" +
+ " headers.separator=<line.separator>\n" +
+ " null.literal=<null.literal>\n" +
+ " key.deserializer=<key.deserializer>\n" +
+ " value.deserializer=<value.deserializer>\n" +
+ " header.deserializer=<header.deserializer>\n" +
+ "\nUsers can also pass in customized properties for their formatter; more specifically, users can pass in properties keyed with 'key.deserializer.', 'value.deserializer.' and 'headers.deserializer.' prefixes to configure their deserializers. " +
+ "\nThis option will be removed in a future version. Use --formatter-property instead.")
+ .withRequiredArg()
+ .describedAs("prop")
+ .ofType(String.class);
+ messageFormatterArgOpt = parser.accepts("formatter-property",
"The properties to initialize the message formatter. Default properties include: \n" +
" print.timestamp=true|false\n" +
" print.key=true|false\n" +
@@ -141,10 +178,26 @@
CommandLineUtils.printUsageAndExit(parser, "At most one of --reject and --release may be specified.");
}
- Properties consumerPropsFromFile = options.has(consumerConfigOpt)
- ? Utils.loadProps(options.valueOf(consumerConfigOpt))
+ if (options.has(consumerPropertyOpt) && options.has(commandPropertyOpt)) {
+ CommandLineUtils.printUsageAndExit(parser, "Options --consumer-property and --command-property cannot be specified together.");
+ }
+ if (options.has(consumerConfigOpt) && options.has(commandConfigOpt)) {
+ CommandLineUtils.printUsageAndExit(parser, "Options --consumer-config and --command-config cannot be specified together.");
+ }
+
+ if (options.has(consumerPropertyOpt)) {
+ System.out.println("Option --consumer-property is deprecated and will be removed in a future version. Use --command-property instead.");
+ commandPropertyOpt = consumerPropertyOpt;
+ }
+ if (options.has(consumerConfigOpt)) {
+ System.out.println("Option --consumer-config is deprecated and will be removed in a future version. Use --command-config instead.");
+ commandConfigOpt = consumerConfigOpt;
+ }
+
+ Properties consumerPropsFromFile = options.has(commandConfigOpt)
+ ? Utils.loadProps(options.valueOf(commandConfigOpt))
: new Properties();
- Properties extraConsumerProps = CommandLineUtils.parseKeyValueArgs(options.valuesOf(consumerPropertyOpt));
+ Properties extraConsumerProps = CommandLineUtils.parseKeyValueArgs(options.valuesOf(commandPropertyOpt));
Set<String> groupIdsProvided = checkShareGroup(consumerPropsFromFile, extraConsumerProps);
consumerProps = buildConsumerProps(consumerPropsFromFile, extraConsumerProps, groupIdsProvided);
@@ -203,6 +256,13 @@
Class<?> messageFormatterClass = Class.forName(options.valueOf(messageFormatterOpt));
formatter = (MessageFormatter) messageFormatterClass.getDeclaredConstructor().newInstance();
+ if (options.has(messageFormatterArgOpt) && options.has(messageFormatterArgOptDeprecated)) {
+ CommandLineUtils.printUsageAndExit(parser, "Options --property and --formatter-property cannot be specified together.");
+ }
+ if (options.has(messageFormatterArgOptDeprecated)) {
+ System.out.println("Option --property is deprecated and will be removed in a future version. Use --formatter-property instead.");
+ messageFormatterArgOpt = messageFormatterArgOptDeprecated;
+ }
Properties formatterArgs = formatterArgs();
Map<String, String> formatterConfigs = new HashMap<>();
for (final String name : formatterArgs.stringPropertyNames()) {
diff --git a/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java b/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
index 6752aef..e1a519a 100644
--- a/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java
@@ -34,6 +34,7 @@
import java.io.OutputStream;
import java.nio.file.Files;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -57,11 +58,16 @@
"--bootstrap-server", "localhost:1002",
"--topic", "t3",
};
- private static final String[] CLIENT_ID_OVERRIDE = new String[]{
+ private static final String[] CLIENT_ID_OVERRIDE_DEPRECATED = new String[]{
"--bootstrap-server", "localhost:1001",
"--topic", "t3",
"--producer-property", "client.id=producer-1"
};
+ private static final String[] CLIENT_ID_OVERRIDE = new String[]{
+ "--bootstrap-server", "localhost:1001",
+ "--topic", "t3",
+ "--command-property", "client.id=producer-1"
+ };
private static final String[] BATCH_SIZE_OVERRIDDEN_BY_MAX_PARTITION_MEMORY_BYTES_VALUE = new String[]{
"--bootstrap-server", "localhost:1002",
"--topic", "t3",
@@ -151,8 +157,8 @@
}
@Test
- public void testClientIdOverride() throws IOException {
- ConsoleProducerOptions opts = new ConsoleProducerOptions(CLIENT_ID_OVERRIDE);
+ public void testClientIdOverrideDeprecated() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(CLIENT_ID_OVERRIDE_DEPRECATED);
ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
assertEquals("producer-1", producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG));
@@ -222,6 +228,107 @@
assertEquals(1, reader.closeCount());
}
+ @Test
+ public void shouldExitOnBothProducerPropertyAndCommandProperty() {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--producer-property", "acks=all",
+ "--command-property", "batch.size=16384"
+ };
+
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleProducerOptions(args));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void shouldExitOnBothProducerConfigAndCommandConfig() throws IOException {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ Map<String, String> configs = new HashMap<>();
+ configs.put("acks", "all");
+ File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+
+ Map<String, String> configs2 = new HashMap<>();
+ configs2.put("batch.size", "16384");
+ File propsFile2 = ToolsTestUtils.tempPropertiesFile(configs2);
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--producer.config", propsFile.getAbsolutePath(),
+ "--command-config", propsFile2.getAbsolutePath()
+ };
+
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleProducerOptions(args));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void testClientIdOverrideUsingCommandProperty() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(CLIENT_ID_OVERRIDE);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("producer-1", producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG));
+ }
+
+ @Test
+ public void testProducerConfigFromFileUsingCommandConfig() throws IOException {
+ Map<String, String> configs = new HashMap<>();
+ configs.put("acks", "all");
+ configs.put("batch.size", "32768");
+ File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ // "all" gets converted to "-1" internally by ProducerConfig
+ assertEquals("-1", producerConfig.getString(ProducerConfig.ACKS_CONFIG));
+ assertEquals(32768, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
+ }
+
+ @Test
+ public void testCommandPropertyOverridesConfig() throws IOException {
+ Map<String, String> configs = new HashMap<>();
+ configs.put("acks", "1");
+ configs.put("batch.size", "16384");
+ File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-config", propsFile.getAbsolutePath(),
+ "--command-property", "acks=all"
+ };
+
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ // Command property should override the config file value
+ // "all" gets converted to "-1" internally by ProducerConfig
+ assertEquals("-1", producerConfig.getString(ProducerConfig.ACKS_CONFIG));
+ // Config file value should still be present
+ assertEquals(16384, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
+ }
+
public static class TestRecordReader implements RecordReader {
private int configureCount = 0;
private int closeCount = 0;
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptionsTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptionsTest.java
index 02ded3d..4639ff6 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptionsTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptionsTest.java
@@ -172,7 +172,7 @@
}
@Test
- public void shouldParseValidConsumerConfigWithAutoOffsetResetLatest() throws IOException {
+ public void shouldParseValidConsumerConfigWithAutoOffsetResetLatestDeprecated() throws IOException {
String[] args = new String[]{
"--bootstrap-server", "localhost:9092",
"--topic", "test",
@@ -189,7 +189,7 @@
}
@Test
- public void shouldParseValidConsumerConfigWithAutoOffsetResetEarliest() throws IOException {
+ public void shouldParseValidConsumerConfigWithAutoOffsetResetEarliestDeprecated() throws IOException {
String[] args = new String[]{
"--bootstrap-server", "localhost:9092",
"--topic", "test",
@@ -206,7 +206,7 @@
}
@Test
- public void shouldParseValidConsumerConfigWithAutoOffsetResetAndMatchingFromBeginning() throws IOException {
+ public void shouldParseValidConsumerConfigWithAutoOffsetResetAndMatchingFromBeginningDeprecated() throws IOException {
String[] args = new String[]{
"--bootstrap-server", "localhost:9092",
"--topic", "test",
@@ -240,7 +240,7 @@
}
@Test
- public void shouldExitOnInvalidConfigWithAutoOffsetResetAndConflictingFromBeginning() {
+ public void shouldExitOnInvalidConfigWithAutoOffsetResetAndConflictingFromBeginningDeprecated() {
Exit.setExitProcedure((code, message) -> {
throw new IllegalArgumentException(message);
});
@@ -259,7 +259,7 @@
}
@Test
- public void shouldParseConfigsFromFile() throws IOException {
+ public void shouldParseConfigsFromFileDeprecated() throws IOException {
Map<String, String> configs = new HashMap<>();
configs.put("request.timeout.ms", "1000");
configs.put("group.id", "group1");
@@ -276,80 +276,82 @@
}
@Test
- public void groupIdsProvidedInDifferentPlacesMustMatch() throws IOException {
+ public void groupIdsProvidedInDifferentPlacesMustMatchDeprecated() throws IOException {
Exit.setExitProcedure((code, message) -> {
throw new IllegalArgumentException(message);
});
+ try {
- // different in all three places
- File propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
- final String[] args = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "group-from-arguments",
- "--consumer-property", "group.id=group-from-properties",
- "--consumer.config", propsFile.getAbsolutePath()
- };
+ // different in all three places
+ File propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--consumer-property", "group.id=group-from-properties",
+ "--consumer.config", propsFile.getAbsolutePath()
+ };
- assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args));
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args));
- // the same in all three places
- propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "test-group"));
- final String[] args1 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "test-group",
- "--consumer-property", "group.id=test-group",
- "--consumer.config", propsFile.getAbsolutePath()
- };
+ // the same in all three places
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "test-group"));
+ final String[] args1 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "test-group",
+ "--consumer-property", "group.id=test-group",
+ "--consumer.config", propsFile.getAbsolutePath()
+ };
- ConsoleConsumerOptions config = new ConsoleConsumerOptions(args1);
- Properties props = config.consumerProps();
- assertEquals("test-group", props.getProperty("group.id"));
+ ConsoleConsumerOptions config = new ConsoleConsumerOptions(args1);
+ Properties props = config.consumerProps();
+ assertEquals("test-group", props.getProperty("group.id"));
- // different via --consumer-property and --consumer.config
- propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
- final String[] args2 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--consumer-property", "group.id=group-from-properties",
- "--consumer.config", propsFile.getAbsolutePath()
- };
+ // different via --consumer-property and --consumer.config
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args2 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--consumer-property", "group.id=group-from-properties",
+ "--consumer.config", propsFile.getAbsolutePath()
+ };
- assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args2));
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args2));
- // different via --consumer-property and --group
- final String[] args3 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "group-from-arguments",
- "--consumer-property", "group.id=group-from-properties"
- };
+ // different via --consumer-property and --group
+ final String[] args3 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--consumer-property", "group.id=group-from-properties"
+ };
- assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args3));
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args3));
- // different via --group and --consumer.config
- propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
- final String[] args4 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "group-from-arguments",
- "--consumer.config", propsFile.getAbsolutePath()
- };
- assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args4));
+ // different via --group and --consumer.config
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args4 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--consumer.config", propsFile.getAbsolutePath()
+ };
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args4));
- // via --group only
- final String[] args5 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "group-from-arguments"
- };
+ // via --group only
+ final String[] args5 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments"
+ };
- config = new ConsoleConsumerOptions(args5);
- props = config.consumerProps();
- assertEquals("group-from-arguments", props.getProperty("group.id"));
-
- Exit.resetExitProcedure();
+ config = new ConsoleConsumerOptions(args5);
+ props = config.consumerProps();
+ assertEquals("group-from-arguments", props.getProperty("group.id"));
+ } finally {
+ Exit.resetExitProcedure();
+ }
}
@Test
@@ -508,7 +510,7 @@
}
@Test
- public void testClientIdOverride() throws IOException {
+ public void testClientIdOverrideDeprecated() throws IOException {
String[] args = new String[]{
"--bootstrap-server", "localhost:9092",
"--topic", "test",
@@ -618,4 +620,234 @@
"--formatter", formatter,
};
}
+
+ @Test
+ public void shouldExitOnBothConsumerPropertyAndCommandProperty() {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--consumer-property", "auto.offset.reset=latest",
+ "--command-property", "session.timeout.ms=10000"
+ };
+
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void shouldExitOnBothConsumerConfigAndCommandConfig() throws IOException {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ Map<String, String> configs = new HashMap<>();
+ configs.put("request.timeout.ms", "1000");
+ File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+
+ Map<String, String> configs2 = new HashMap<>();
+ configs2.put("session.timeout.ms", "10000");
+ File propsFile2 = ToolsTestUtils.tempPropertiesFile(configs2);
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--consumer.config", propsFile.getAbsolutePath(),
+ "--command-config", propsFile2.getAbsolutePath()
+ };
+
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void shouldParseValidConsumerConfigWithAutoOffsetResetLatestUsingCommandProperty() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "auto.offset.reset=latest"
+ };
+
+ ConsoleConsumerOptions config = new ConsoleConsumerOptions(args);
+ Properties consumerProperties = config.consumerProps();
+
+ assertEquals("localhost:9092", config.bootstrapServer());
+ assertEquals("test", config.topicArg().orElse(""));
+ assertFalse(config.fromBeginning());
+ assertEquals("latest", consumerProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
+ }
+
+ @Test
+ public void shouldParseValidConsumerConfigWithAutoOffsetResetEarliestUsingCommandProperty() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "auto.offset.reset=earliest"
+ };
+
+ ConsoleConsumerOptions config = new ConsoleConsumerOptions(args);
+ Properties consumerProperties = config.consumerProps();
+
+ assertEquals("localhost:9092", config.bootstrapServer());
+ assertEquals("test", config.topicArg().orElse(""));
+ assertFalse(config.fromBeginning());
+ assertEquals("earliest", consumerProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
+ }
+
+ @Test
+ public void shouldParseValidConsumerConfigWithAutoOffsetResetAndMatchingFromBeginningUsingCommandProperty() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "auto.offset.reset=earliest",
+ "--from-beginning"
+ };
+
+ ConsoleConsumerOptions config = new ConsoleConsumerOptions(args);
+ Properties consumerProperties = config.consumerProps();
+
+ assertEquals("localhost:9092", config.bootstrapServer());
+ assertEquals("test", config.topicArg().orElse(""));
+ assertTrue(config.fromBeginning());
+ assertEquals("earliest", consumerProperties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
+ }
+
+ @Test
+ public void shouldExitOnInvalidConfigWithAutoOffsetResetAndConflictingFromBeginningUsingCommandProperty() {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "auto.offset.reset=latest",
+ "--from-beginning"
+ };
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void shouldParseConfigsFromFileUsingCommandConfig() throws IOException {
+ Map<String, String> configs = new HashMap<>();
+ configs.put("request.timeout.ms", "1000");
+ configs.put("group.id", "group1");
+ File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleConsumerOptions config = new ConsoleConsumerOptions(args);
+ assertEquals("1000", config.consumerProps().get("request.timeout.ms"));
+ assertEquals("group1", config.consumerProps().get("group.id"));
+ }
+
+ @Test
+ public void groupIdsProvidedInDifferentPlacesMustMatchUsingCommandConfig() throws IOException {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ try {
+ // different in all three places
+ File propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--command-property", "group.id=group-from-properties",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args));
+
+ // the same in all three places
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "test-group"));
+ final String[] args1 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "test-group",
+ "--command-property", "group.id=test-group",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleConsumerOptions config = new ConsoleConsumerOptions(args1);
+ Properties props = config.consumerProps();
+ assertEquals("test-group", props.getProperty("group.id"));
+
+ // different via --command-property and --command-config
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args2 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "group.id=group-from-properties",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args2));
+
+ // different via --command-property and --group
+ final String[] args3 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--command-property", "group.id=group-from-properties"
+ };
+
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args3));
+
+ // different via --group and --command-config
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args4 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleConsumerOptions(args4));
+
+ // via --group only
+ final String[] args5 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments"
+ };
+
+ config = new ConsoleConsumerOptions(args5);
+ props = config.consumerProps();
+ assertEquals("group-from-arguments", props.getProperty("group.id"));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void testClientIdOverrideUsingCommandProperty() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--from-beginning",
+ "--command-property", "client.id=consumer-1"
+ };
+
+ ConsoleConsumerOptions config = new ConsoleConsumerOptions(args);
+ Properties consumerProperties = config.consumerProps();
+
+ assertEquals("consumer-1", consumerProperties.getProperty(ConsumerConfig.CLIENT_ID_CONFIG));
+ }
}
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java
index c2b7b73..fecf53d 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java
@@ -20,6 +20,7 @@
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.test.MockDeserializer;
import org.apache.kafka.tools.ToolsTestUtils;
import org.junit.jupiter.api.Test;
@@ -32,7 +33,9 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class ConsoleShareConsumerOptionsTest {
@@ -72,7 +75,7 @@
}
@Test
- public void shouldParseValidConsumerConfigWithSessionTimeout() throws IOException {
+ public void shouldParseValidConsumerConfigWithSessionTimeoutDeprecated() throws IOException {
String[] args = new String[]{
"--bootstrap-server", "localhost:9092",
"--topic", "test",
@@ -88,7 +91,7 @@
}
@Test
- public void shouldParseConfigsFromFile() throws IOException {
+ public void shouldParseConfigsFromFileDeprecated() throws IOException {
Map<String, String> configs = new HashMap<>();
configs.put("request.timeout.ms", "1000");
configs.put("group.id", "group1");
@@ -109,80 +112,82 @@
}
@Test
- public void groupIdsProvidedInDifferentPlacesMustMatch() throws IOException {
+ public void groupIdsProvidedInDifferentPlacesMustMatchDeprecated() throws IOException {
Exit.setExitProcedure((code, message) -> {
throw new IllegalArgumentException(message);
});
- // different in all three places
- File propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
- final String[] args = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "group-from-arguments",
- "--consumer-property", "group.id=group-from-properties",
- "--consumer-config", propsFile.getAbsolutePath()
- };
+ try {
+ // different in all three places
+ File propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--consumer-property", "group.id=group-from-properties",
+ "--consumer-config", propsFile.getAbsolutePath()
+ };
- assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args));
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args));
- // the same in all three places
- propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "test-group"));
- final String[] args1 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "test-group",
- "--consumer-property", "group.id=test-group",
- "--consumer-config", propsFile.getAbsolutePath()
- };
+ // the same in all three places
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "test-group"));
+ final String[] args1 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "test-group",
+ "--consumer-property", "group.id=test-group",
+ "--consumer-config", propsFile.getAbsolutePath()
+ };
- ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args1);
- Properties props = config.consumerProps();
- assertEquals("test-group", props.getProperty("group.id"));
+ ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args1);
+ Properties props = config.consumerProps();
+ assertEquals("test-group", props.getProperty("group.id"));
- // different via --consumer-property and --consumer-config
- propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
- final String[] args2 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--consumer-property", "group.id=group-from-properties",
- "--consumer-config", propsFile.getAbsolutePath()
- };
+ // different via --consumer-property and --consumer-config
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args2 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--consumer-property", "group.id=group-from-properties",
+ "--consumer-config", propsFile.getAbsolutePath()
+ };
- assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args2));
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args2));
- // different via --consumer-property and --group
- final String[] args3 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "group-from-arguments",
- "--consumer-property", "group.id=group-from-properties"
- };
+ // different via --consumer-property and --group
+ final String[] args3 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--consumer-property", "group.id=group-from-properties"
+ };
- assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args3));
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args3));
- // different via --group and --consumer-config
- propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
- final String[] args4 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "group-from-arguments",
- "--consumer-config", propsFile.getAbsolutePath()
- };
- assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args4));
+ // different via --group and --consumer-config
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args4 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--consumer-config", propsFile.getAbsolutePath()
+ };
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args4));
- // via --group only
- final String[] args5 = new String[]{
- "--bootstrap-server", "localhost:9092",
- "--topic", "test",
- "--group", "group-from-arguments"
- };
+ // via --group only
+ final String[] args5 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments"
+ };
- config = new ConsoleShareConsumerOptions(args5);
- props = config.consumerProps();
- assertEquals("group-from-arguments", props.getProperty("group.id"));
-
- Exit.resetExitProcedure();
+ config = new ConsoleShareConsumerOptions(args5);
+ props = config.consumerProps();
+ assertEquals("group-from-arguments", props.getProperty("group.id"));
+ } finally {
+ Exit.resetExitProcedure();
+ }
}
@Test
@@ -203,7 +208,7 @@
}
@Test
- public void testClientIdOverride() throws IOException {
+ public void testClientIdOverrideDeprecated() throws IOException {
String[] args = new String[]{
"--bootstrap-server", "localhost:9092",
"--topic", "test",
@@ -217,6 +222,56 @@
}
@Test
+ public void testCustomPropertyShouldBePassedToConfigureMethod() throws Exception {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--property", "print.key=true",
+ "--property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
+ "--property", "key.deserializer.my-props=abc"
+ };
+
+ ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args);
+
+ assertInstanceOf(DefaultMessageFormatter.class, config.formatter());
+ assertTrue(config.formatterArgs().containsKey("key.deserializer.my-props"));
+ DefaultMessageFormatter formatter = (DefaultMessageFormatter) config.formatter();
+ assertTrue(formatter.keyDeserializer().isPresent());
+ assertInstanceOf(MockDeserializer.class, formatter.keyDeserializer().get());
+ MockDeserializer keyDeserializer = (MockDeserializer) formatter.keyDeserializer().get();
+ assertEquals(1, keyDeserializer.configs.size());
+ assertEquals("abc", keyDeserializer.configs.get("my-props"));
+ assertTrue(keyDeserializer.isKey);
+ }
+
+ @Test
+ public void testCustomConfigShouldBePassedToConfigureMethod() throws Exception {
+ Map<String, String> configs = new HashMap<>();
+ configs.put("key.deserializer.my-props", "abc");
+ configs.put("print.key", "false");
+ File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--property", "print.key=true",
+ "--property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
+ "--formatter-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args);
+
+ assertInstanceOf(DefaultMessageFormatter.class, config.formatter());
+ assertTrue(config.formatterArgs().containsKey("key.deserializer.my-props"));
+ DefaultMessageFormatter formatter = (DefaultMessageFormatter) config.formatter();
+ assertTrue(formatter.keyDeserializer().isPresent());
+ assertInstanceOf(MockDeserializer.class, formatter.keyDeserializer().get());
+ MockDeserializer keyDeserializer = (MockDeserializer) formatter.keyDeserializer().get();
+ assertEquals(1, keyDeserializer.configs.size());
+ assertEquals("abc", keyDeserializer.configs.get("my-props"));
+ assertTrue(keyDeserializer.isKey);
+ }
+
+ @Test
public void testDefaultClientId() throws IOException {
String[] args = new String[]{
"--bootstrap-server", "localhost:9092",
@@ -271,4 +326,182 @@
Exit.resetExitProcedure();
}
}
+
+ @Test
+ public void shouldExitOnBothConsumerPropertyAndCommandProperty() {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--consumer-property", "session.timeout.ms=10000",
+ "--command-property", "request.timeout.ms=30000"
+ };
+
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void shouldExitOnBothConsumerConfigAndCommandConfig() throws IOException {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ Map<String, String> configs = new HashMap<>();
+ configs.put("request.timeout.ms", "1000");
+ File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+
+ Map<String, String> configs2 = new HashMap<>();
+ configs2.put("session.timeout.ms", "10000");
+ File propsFile2 = ToolsTestUtils.tempPropertiesFile(configs2);
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--consumer-config", propsFile.getAbsolutePath(),
+ "--command-config", propsFile2.getAbsolutePath()
+ };
+
+ try {
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void shouldParseValidConsumerConfigWithSessionTimeout() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "session.timeout.ms=10000"
+ };
+
+ ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args);
+ Properties consumerProperties = config.consumerProps();
+
+ assertEquals("localhost:9092", config.bootstrapServer());
+ assertEquals("test", config.topicArg());
+ assertEquals("10000", consumerProperties.getProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG));
+ }
+
+ @Test
+ public void shouldParseConfigsFromFile() throws IOException {
+ Map<String, String> configs = new HashMap<>();
+ configs.put("request.timeout.ms", "1000");
+ configs.put("group.id", "group1");
+ File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args);
+
+ // KafkaShareConsumer uses Utils.propsToMap to convert the properties to a map,
+ // so using the same method to check the map has the expected values
+ Map<String, Object> configMap = Utils.propsToMap(config.consumerProps());
+ assertEquals("1000", configMap.get("request.timeout.ms"));
+ assertEquals("group1", configMap.get("group.id"));
+ }
+
+ @Test
+ public void groupIdsProvidedInDifferentPlacesMustMatch() throws IOException {
+ Exit.setExitProcedure((code, message) -> {
+ throw new IllegalArgumentException(message);
+ });
+
+ try {
+ // different in all three places
+ File propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--command-property", "group.id=group-from-properties",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args));
+
+ // the same in all three places
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "test-group"));
+ final String[] args1 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "test-group",
+ "--command-property", "group.id=test-group",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args1);
+ Properties props = config.consumerProps();
+ assertEquals("test-group", props.getProperty("group.id"));
+
+ // different via --command-property and --command-config
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args2 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "group.id=group-from-properties",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args2));
+
+ // different via --command-property and --group
+ final String[] args3 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--command-property", "group.id=group-from-properties"
+ };
+
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args3));
+
+ // different via --group and --command-config
+ propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("group.id", "group-from-file"));
+ final String[] args4 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+ assertThrows(IllegalArgumentException.class, () -> new ConsoleShareConsumerOptions(args4));
+
+ // via --group only
+ final String[] args5 = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--group", "group-from-arguments"
+ };
+
+ config = new ConsoleShareConsumerOptions(args5);
+ props = config.consumerProps();
+ assertEquals("group-from-arguments", props.getProperty("group.id"));
+ } finally {
+ Exit.resetExitProcedure();
+ }
+ }
+
+ @Test
+ public void testClientIdOverride() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "client.id=consumer-1"
+ };
+
+ ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args);
+ Properties consumerProperties = config.consumerProps();
+
+ assertEquals("consumer-1", consumerProperties.getProperty(ConsumerConfig.CLIENT_ID_CONFIG));
+ }
}
