KAFKA-19498: Add include argument to ConsumerPerformance tool

[fvaleri]

This patch adds the include argument to ConsumerPerformance tool.

ConsoleConsumer and ConsumerPerformance serve different purposes but
share common functionality for message consumption. Currently, there's
an inconsistency in their command-line interfaces:

• ConsoleConsumer supports an --include argument that allows users to
  specify a regular expression pattern to filter topics for consumption
• ConsumerPerformance lacks this topic filtering capability, requiring
  users to specify a single topic explicitly via --topic argument

This inconsistency creates two problems:

• Similar tools should provide similar topic selection capabilities for
  better user experience
• Users cannot test consumer performance across multiple topics or
  dynamically matching topic sets, making it difficult to test realistic
  scenarios

Reviewers: Chia-Ping Tsai [email protected]

[fvaleri]

Usage example:

$ bin/kafka-topics.sh --bootstrap-server :9092 --create --topic t1
Created topic t1.
$ bin/kafka-topics.sh --bootstrap-server :9092 --create --topic t2
Created topic t2.
$ bin/kafka-topics.sh --bootstrap-server :9092 --create --topic t3
Created topic t3.

$ bin/kafka-console-producer.sh --bootstrap-server :9092 --topic t1
>foo
>^C
$ bin/kafka-console-producer.sh --bootstrap-server :9092 --topic t2
>bar
>^C
$ bin/kafka-console-producer.sh --bootstrap-server :9092 --topic t3
>baz
>^C

$ bin/kafka-consumer-perf-test.sh --bootstrap-server :9092 --include t.* --messages 3
start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
2025-07-22 17:07:21:762, 2025-07-22 17:07:25:272, 0.0000, 0.0000, 3, 0.8547, 3464, 46, 0.0002, 65.2174

$ bin/kafka-consumer-perf-test.sh --bootstrap-server :9092 --messages 4
Exactly one of the following arguments is required: [topic], [include]
...

$ bin/kafka-consumer-perf-test.sh --bootstrap-server :9092 --topic t1 --include t.* --messages 3
Exactly one of the following arguments is required: [topic], [include]
...

Relevant system test run:

================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.12.0
session_id:       2025-07-22--001
run time:         1 minute 14.771 seconds
tests run:        4
passed:           4
flaky:            0
failed:           0
ignored:          0
================================================================================
test_id:    kafkatest.sanity_checks.test_performance_services.PerformanceServiceTest.test_version.version=2.1.1.metadata_quorum=COMBINED_KRAFT
status:     PASS
run time:   25.090 seconds
{"producer_performance": {"records_per_sec": 26041.7, "mb_per_sec": 2.48}, "end_to_end_latency": {"latency_50th_ms": 0.0, "latency_99th_ms": 1.0, "latency_999th_ms": 3.0}, "consumer_performance": {"records_per_sec": 38625.4826, "mb_per_sec": 3.6824}}
--------------------------------------------------------------------------------
test_id:    kafkatest.sanity_checks.test_performance_services.PerformanceServiceTest.test_version.version=2.1.1.metadata_quorum=ISOLATED_KRAFT
status:     PASS
run time:   31.330 seconds
{"producer_performance": {"records_per_sec": 42372.9, "mb_per_sec": 4.04}, "end_to_end_latency": {"latency_50th_ms": 0.0, "latency_99th_ms": 2.0, "latency_999th_ms": 3.0}, "consumer_performance": {"records_per_sec": 46752.3364, "mb_per_sec": 4.4569}}
--------------------------------------------------------------------------------
test_id:    kafkatest.sanity_checks.test_performance_services.PerformanceServiceTest.test_version.version=dev.metadata_quorum=COMBINED_KRAFT.use_share_groups=True
status:     PASS
run time:   37.767 seconds
{"producer_performance": {"records_per_sec": 25188.9, "mb_per_sec": 2.4}, "end_to_end_latency": {"latency_50th_ms": 0.0, "latency_99th_ms": 3.0, "latency_999th_ms": 5.0}, "consumer_performance": {"records_per_sec": 26714.2857, "mb_per_sec": 2.457}, "share_consumer_performance": {"records_per_sec": 10342.0, "mb_per_sec": 0.1732}}
--------------------------------------------------------------------------------
test_id:    kafkatest.sanity_checks.test_performance_services.PerformanceServiceTest.test_version.version=dev.metadata_quorum=ISOLATED_KRAFT.use_share_groups=True
status:     PASS
run time:   43.212 seconds
{"producer_performance": {"records_per_sec": 20202.0, "mb_per_sec": 1.93}, "end_to_end_latency": {"latency_50th_ms": 0.0, "latency_99th_ms": 2.0, "latency_999th_ms": 3.0}, "consumer_performance": {"records_per_sec": 28150.5376, "mb_per_sec": 2.5891}, "share_consumer_performance": {"records_per_sec": 10342.0, "mb_per_sec": 0.1737}}
--------------------------------------------------------------------------------

[chia7712]

please fix the build error:

Execution failed for task ':server-common:spotlessJavaCheck'.
> The following files had format violations:
      src/test/java/org/apache/kafka/server/util/CommandLineUtilsTest.java
          @@ -17,6 +17,7 @@
           package·org.apache.kafka.server.util;
           
           import·org.apache.kafka.common.utils.Exit;
          +
           import·org.junit.jupiter.api.Test;
           
           import·java.util.List;
  Run './gradlew :server-common:spotlessApply' to fix these violations.

[fvaleri]

@chia7712 done.

[fvaleri]

The only failed test is DescribeStreamsGroupTest.testDescribeWithoutGroupOption that is fixed by #20225. Rebased.

[chia7712]

@fvaleri thanks for this patch. one major comment is left.

[fvaleri]

@chia7712 gentle reminder about this patch

[chia7712]

@fvaleri thanks for this patch. I have one last question. PTAL

[chia7712]

@fvaleri Could you please rebase the code to include the CI changes? Otherwise, the build under JDK 24 will always fail

[fvaleri]

@fvaleri Could you please rebase the code to include the CI changes? Otherwise, the build under JDK 24 will always fail

Sure, just pushed the rebase. Thanks.

[chia7712]

open https://issues.apache.org/jira/browse/KAFKA-19641 to trace the flaky

[chia7712]

@fvaleri could you please update upgrade.html to include this new feature?

[fvaleri]

@chia7712 upgrade note added. Thanks for opening the flaky test issue.

[chia7712]

LGTM