AndrewJSchofield commented on code in PR #20385:
URL: https://github.com/apache/kafka/pull/20385#discussion_r2298444121
##########
tests/kafkatest/services/performance/share_consumer_performance.py:
##########
@@ -33,7 +33,7 @@ class ShareConsumerPerformanceService(PerformanceService):
"socket-buffer-size", "The size of the tcp RECV size."
- "consumer.config", "Consumer config properties file."
+ "command-config", "Share consumer config properties file."
Review Comment:
Given that we are trying in this KIP to get away from specific names for
producer/config config, maybe this could be "Config properties file".
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -58,7 +58,7 @@ public static void main(String[] args) {
try {
LOG.info("Starting consumer...");
ConsumerPerfOptions options = new ConsumerPerfOptions(args);
- AtomicLong totalMessagesRead = new AtomicLong(0);
+ AtomicLong totalRecordsRead = new AtomicLong(0);
Review Comment:
Thanks for the extra effort to align the terminology in the variable names
too.
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -310,7 +321,12 @@ public ConsumerPerfOptions(String[] args) {
.describedAs("milliseconds")
.ofType(Long.class)
.defaultsTo(10_000L);
- numMessagesOpt = parser.accepts("messages", "REQUIRED: The number
of messages to consume.")
+ numMessagesOpt = parser.accepts("messages", "(DEPRECATED)
REQUIRED: The number of messages to consume. " +
Review Comment:
nit: Personally, I would change the help message to say "The number of
records to consume." just so it matches the new `num-records` description.
Although we know that records and messages are interchangeable, maybe some
users would be confused.
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -349,8 +366,15 @@ public String brokerHostsAndPorts() {
}
public Properties props() throws IOException {
- Properties props = (options.has(consumerConfigOpt))
- ? Utils.loadProps(options.valueOf(consumerConfigOpt))
+ String commandConfigFile;
+ if (options.has(consumerConfigOpt)) {
+ System.out.println("Warning: --consumer.config is deprecated.
Use --command-config instead.");
+ commandConfigFile = options.valueOf(consumerConfigOpt);
+ } else {
+ commandConfigFile = options.valueOf(commandConfigOpt);
+ }
Review Comment:
Indeed. I'd check for both specified together first, then I'd use the new
one, then I'd use the old one with a deprecation warning, and then I'd default
to the empty Properties.
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -310,7 +321,12 @@ public ConsumerPerfOptions(String[] args) {
.describedAs("milliseconds")
.ofType(Long.class)
.defaultsTo(10_000L);
- numMessagesOpt = parser.accepts("messages", "REQUIRED: The number
of messages to consume.")
+ numMessagesOpt = parser.accepts("messages", "(DEPRECATED)
REQUIRED: The number of messages to consume. " +
Review Comment:
It's also a bit odd specifying "REQUIRED" when of course you just need one
or the other of `messages` or `num-records`. Maybe just remove "REQUIRED" from
the old option. I know that's not 100% correct, but it's perhaps the best
compromise.
##########
tools/src/main/java/org/apache/kafka/tools/ShareConsumerPerformance.java:
##########
@@ -366,7 +382,9 @@ public ShareConsumerPerfOptions(String[] args) {
}
if (options != null) {
CommandLineUtils.maybePrintHelpOrVersion(this, "This tool is
used to verify the share consumer performance.");
- CommandLineUtils.checkRequiredArgs(parser, options, topicOpt,
numMessagesOpt);
+ CommandLineUtils.checkRequiredArgs(parser, options, topicOpt,
numRecordsOpt);
Review Comment:
Doesn't this require `--num-records` when actually `--messages` is still
valid. The latter just produces a deprecation message if used.
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -292,13 +298,18 @@ public ConsumerPerfOptions(String[] args) {
.ofType(Integer.class)
.defaultsTo(1024 * 1024);
resetBeginningOffsetOpt = parser.accepts("from-latest", "If the
consumer does not already have an established " +
- "offset to consume from, start with the latest message present
in the log rather than the earliest message.");
+ "offset to consume from, start with the latest record present
in the log rather than the earliest record.");
socketBufferSizeOpt = parser.accepts("socket-buffer-size", "The
size of the tcp RECV size.")
.withRequiredArg()
.describedAs("size")
.ofType(Integer.class)
.defaultsTo(2 * 1024 * 1024);
- consumerConfigOpt = parser.accepts("consumer.config", "Consumer
config properties file.")
+ consumerConfigOpt = parser.accepts("consumer.config",
"(DEPRECATED) Consumer config properties file. " +
+ "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", "Consumer
config properties file")
Review Comment:
Given that we are trying in this KIP to get away from specific names for
producer/config config, maybe this could be "Config properties file".
##########
tests/kafkatest/services/performance/consumer_performance.py:
##########
@@ -40,7 +40,7 @@ class ConsumerPerformanceService(PerformanceService):
"socket-buffer-size", "The size of the tcp RECV size."
"new-consumer", "Use the new consumer implementation."
- "consumer.config", "Consumer config properties file."
+ "command-config", "Consumer config properties file."
Review Comment:
Given that we are trying in this KIP to get away from specific names for
producer/config config, maybe this could be "Config properties file".
##########
tools/src/main/java/org/apache/kafka/tools/ShareConsumerPerformance.java:
##########
@@ -322,7 +328,12 @@ public ShareConsumerPerfOptions(String[] args) {
.describedAs("size")
.ofType(Integer.class)
.defaultsTo(2 * 1024 * 1024);
- consumerConfigOpt = parser.accepts("consumer.config", "Share
consumer config properties file.")
+ consumerConfigOpt = parser.accepts("consumer.config",
"(DEPRECATED) Share consumer config properties file. " +
+ "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", "Share
consumer config properties file.")
Review Comment:
Given that we are trying in this KIP to get away from specific names for
producer/config config, maybe this could be "Config properties file".
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -292,13 +298,18 @@ public ConsumerPerfOptions(String[] args) {
.ofType(Integer.class)
.defaultsTo(1024 * 1024);
resetBeginningOffsetOpt = parser.accepts("from-latest", "If the
consumer does not already have an established " +
- "offset to consume from, start with the latest message present
in the log rather than the earliest message.");
+ "offset to consume from, start with the latest record present
in the log rather than the earliest record.");
socketBufferSizeOpt = parser.accepts("socket-buffer-size", "The
size of the tcp RECV size.")
.withRequiredArg()
.describedAs("size")
.ofType(Integer.class)
.defaultsTo(2 * 1024 * 1024);
- consumerConfigOpt = parser.accepts("consumer.config", "Consumer
config properties file.")
+ consumerConfigOpt = parser.accepts("consumer.config",
"(DEPRECATED) Consumer config properties file. " +
+ "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", "Consumer
config properties file")
Review Comment:
Good catch. You can't use both. They are optional.
##########
tools/src/main/java/org/apache/kafka/tools/ShareConsumerPerformance.java:
##########
@@ -366,7 +382,9 @@ public ShareConsumerPerfOptions(String[] args) {
}
if (options != null) {
CommandLineUtils.maybePrintHelpOrVersion(this, "This tool is
used to verify the share consumer performance.");
- CommandLineUtils.checkRequiredArgs(parser, options, topicOpt,
numMessagesOpt);
+ CommandLineUtils.checkRequiredArgs(parser, options, topicOpt,
numRecordsOpt);
+ CommandLineUtils.checkOneOfArgs(parser, options,
numMessagesOpt, numRecordsOpt);
+ CommandLineUtils.checkOneOfArgs(parser, options,
consumerConfigOpt, commandConfigOpt);
Review Comment:
This is incorrect. It is not required to specify one of these.
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -378,8 +402,10 @@ public Optional<Pattern> include() {
: Optional.empty();
}
- public long numMessages() {
- return options.valueOf(numMessagesOpt);
+ public long numRecords() {
+ return options.has(numMessagesOpt)
+ ? options.valueOf(numMessagesOpt)
+ : options.valueOf(numRecordsOpt);
Review Comment:
We should.
##########
tools/src/main/java/org/apache/kafka/tools/ShareConsumerPerformance.java:
##########
@@ -398,8 +423,10 @@ public Set<String> topic() {
return Set.of(options.valueOf(topicOpt));
}
- public long numMessages() {
- return options.valueOf(numMessagesOpt);
+ public long numRecords() {
+ return options.has(numMessagesOpt)
+ ? options.valueOf(numMessagesOpt)
Review Comment:
Deprecation warning?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
