KAFKA-19624: Improving consistency of command-line arguments for consumer performance tests

[aheev]

resolves https://issues.apache.org/jira/browse/KAFKA-19624

Reviewers: @brandboat, @AndrewJSchofield, @m1a2st

[aheev]

@AndrewJSchofield can you please review this?

[brandboat]

ping @aheev, could you please fix the conflicts?

[aheev]

ping @aheev, could you please fix the conflicts?

done

[brandboat]

Thanks for the patch, left some comments below.

[aheev]

FAILED ❌ RestoreIntegrationTest > "shouldInvokeUserDefinedGlobalStateRestoreListener(boolean).useNewProtocol=false"

This test seems to be flaky. It succeeds on my machine and fails sometimes. Also the changes are not even related to the test

Just tested. Same issue on trunk too

[m1a2st]

Thanks for this patch, some comments left

[AndrewJSchofield]

Thanks for the PR. Please add some tests for the variation combinations of new/old options to make sure the rules in the KIP are correctly implemented.

[brandboat]

Thanks for the update, some minor comments.

[brandboat]

LGTM, thanks!

[AndrewJSchofield]

Apart from adding --command-property to both kafka-consumer-perf-test.sh and kafka-share-consumer-perf-test.sh, this looks good to me.

[aheev]

Overall, LGTM. Could you run the e2e test and share the results?

What do you mean by e2e test?

[m1a2st]

Since you have modified the e2e test file consumer_performance.py, you can validate it by running the following command to ensure the test still works:
TC_PATHS="tests/kafkatest/benchmarks/core/benchmark_test.py" bash tests/docker/run_tests.sh

FYI: https://github.com/apache/kafka/blob/trunk/tests/README.md

[aheev]

Since you have modified the e2e test file consumer_performance.py, you can validate it by running the following command to ensure the test still works: TC_PATHS="tests/kafkatest/benchmarks/core/benchmark_test.py" bash tests/docker/run_tests.sh

FYI: https://github.com/apache/kafka/blob/trunk/tests/README.md

I will run them after andrew's comments are resolved

[AndrewJSchofield]

@aheev Please resolve conflicts

[AndrewJSchofield]

Thanks for the PR. Once we have a green build, I'm ready to merge this.

[aheev]

Thanks for the PR. Once we have a green build, I'm ready to merge this.

Shouldn't we wait for this? I am trying to run the tests, but running into some issues

Since you have modified the e2e test file consumer_performance.py, you can validate it by running the following command to ensure the test still works:
TC_PATHS="tests/kafkatest/benchmarks/core/benchmark_test.py" bash tests/docker/run_tests.sh

[AndrewJSchofield]

Thanks for the PR. Once we have a green build, I'm ready to merge this.

Shouldn't we wait for this? I am trying to run the tests, but running into some issues

Since you have modified the e2e test file consumer_performance.py, you can validate it by running the following command to ensure the test still works:
TC_PATHS="tests/kafkatest/benchmarks/core/benchmark_test.py" bash tests/docker/run_tests.sh

Yes, we should. I'll wait then.

[aheev]

After a lot of struggle I managed to run the test by using a different JDK(will file a ticket for this later). Here are the results @m1a2st . Three tests failed, all of which are related to producer-perf
benchmark_test.log

[m1a2st]

Here are the results @m1a2st . Three tests failed, all of which are related to producer-perf

Thanks, Could you file a ticket to trace this issue?

[aheev]

Here are the results @m1a2st . Three tests failed, all of which are related to producer-perf

Thanks, Could you file a ticket to trace this issue?

https://issues.apache.org/jira/browse/KAFKA-19672

[aheev]

This test seems to be flaky and not related to the changes. It succeeded in a couple of runs in my local. @AndrewJSchofield should we trigger a CI re-run?

FAILED ❌ SmokeTestDriverIntegrationTest > "shouldWorkWithRebalance(boolean, boolean, boolean).stateUpdaterEnabled=false, processingThreadsEnabled=false, streamsProtocolEnabled=true"
FAILED ❌ SmokeTestDriverIntegrationTest > "shouldWorkWithRebalance(boolean, boolean, boolean).stateUpdaterEnabled=true, processingThreadsEnabled=false, streamsProtocolEnabled=true"
FAILED ❌ SmokeTestDriverIntegrationTest > "shouldWorkWithRebalance(boolean, boolean, boolean).stateUpdaterEnabled=true, processingThreadsEnabled=true, streamsProtocolEnabled=true"
Found 2 flaky test failures:
FLAKY ⚠️  SmokeTestDriverIntegrationTest > "shouldWorkWithRebalance(boolean, boolean, boolean).stateUpdaterEnabled=true, processingThreadsEnabled=true, streamsProtocolEnabled=false"
FLAKY ⚠️  SmokeTestDriverIntegrationTest > "shouldWorkWithRebalance(boolean, boolean, boolean).stateUpdaterEnabled=true, processingThreadsEnabled=false, streamsProtocolEnabled=false"

[AndrewJSchofield]

@aheev I would merge latest changes into this branch. I find the test fails on your branch, and passes on trunk. I can't see that this PR would cause the tests to fail, so I'd update the branch and let the CI run again.

[aheev]

@Yunyung ConsumerPerformanceService and ShareConsumerPerformanceService are used in benchmark_test.py(which uses DEV_BRANCH and test_performance_services.py(which uses LATEST_2_1 and DEV_BRANCH)

• ShareConsumerPerformanceService runs only in test_performance_services.py runs on branches >= 4.1. Hence, we can ignore this
• As for ConsumerPerformanceService, I have added changes, but, while running the test, ProducerPerformanceService, which is used to produce data in turn consumed by ConsumerPerformanceService, fails, thereby blocking the ConsumerPerformanceService to consume on LATEST_2_1. It fails on trunk too

[INFO:2025-09-07 09:01:58,715]: RunnerClient: kafkatest.sanity_checks.test_performance_services.PerformanceServiceTest.test_version.version=2.1.1.metadata_quorum=ISOLATED_KRAFT: FAIL: Exception('ProducerPerformanceService-0-125869363841104-worker-1: Traceback (most recent call last):\n  File "/usr/local/lib/python3.10/dist-packages/ducktape/services/background_thread.py", line 36, in _protected_worker\n    self._worker(idx, node)\n  File "/opt/kafka-dev/tests/kafkatest/services/performance/producer_performance.py", line 130, in _worker\n    wait_until(lambda: self.alive(node), timeout_sec=20, err_msg="ProducerPerformance failed to start")\n  File "/usr/local/lib/python3.10/dist-packages/ducktape/utils/util.py", line 58, in wait_until\n    raise TimeoutError(err_msg() if callable(err_msg) else err_msg) from last_exception\nducktape.errors.TimeoutError: ProducerPerformance failed to start\n')
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/ducktape/tests/runner_client.py", line 351, in _do_run
    data = self.run_test()
  File "/usr/local/lib/python3.10/dist-packages/ducktape/tests/runner_client.py", line 411, in run_test
    return self.test_context.function(self.test)
  File "/usr/local/lib/python3.10/dist-packages/ducktape/mark/_mark.py", line 438, in wrapper
    return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
  File "/opt/kafka-dev/tests/kafkatest/sanity_checks/test_performance_services.py", line 58, in test_version
    self.producer_perf.run()
  File "/usr/local/lib/python3.10/dist-packages/ducktape/services/service.py", line 345, in run
    self.wait()
  File "/usr/local/lib/python3.10/dist-packages/ducktape/services/background_thread.py", line 72, in wait
    self._propagate_exceptions()
  File "/usr/local/lib/python3.10/dist-packages/ducktape/services/background_thread.py", line 103, in _propagate_exceptions
    raise Exception(self.errors)
Exception: ProducerPerformanceService-0-125869363841104-worker-1: Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/ducktape/services/background_thread.py", line 36, in _protected_worker
    self._worker(idx, node)
  File "/opt/kafka-dev/tests/kafkatest/services/performance/producer_performance.py", line 130, in _worker
    wait_until(lambda: self.alive(node), timeout_sec=20, err_msg="ProducerPerformance failed to start")
  File "/usr/local/lib/python3.10/dist-packages/ducktape/utils/util.py", line 58, in wait_until
    raise TimeoutError(err_msg() if callable(err_msg) else err_msg) from last_exception
ducktape.errors.TimeoutError: ProducerPerformance failed to start

I think we need to fix our test suite first
@AndrewJSchofield what is the way forward here?