KAFKA-18428: Measure share consumers performance #18415
Conversation
| @@ -0,0 +1,20 @@ | |||
| #!/bin/bash | |||
There was a problem hiding this comment.
Please can you also create bin/windows/kafka-share-consumer-perf-test.bat.
| ShareConsumerPerfOptions options = new ShareConsumerPerfOptions(args); | ||
| AtomicLong totalMessagesRead = new AtomicLong(0); | ||
| AtomicLong totalBytesRead = new AtomicLong(0); | ||
| AtomicLong joinTimeMs = new AtomicLong(0); |
There was a problem hiding this comment.
joinTimeMs and joinTimeMsInSingleRound don't make sense for a share group consumer. In the regular consumer perf test, they are used to see how long it took to join the group. A share consumer cannot work this out because there is no onPartitionsAssigned callback. Please remove these variables throughout the code.
| printStats(totalBytesRead.get(), totalMessagesRead.get(), elapsedSec, fetchTimeInMs, startMs, endMs, | ||
| options.dateFormat(), -1); | ||
|
|
||
| if (!shareConsumersMetrics.isEmpty()) { |
There was a problem hiding this comment.
I think this would be neater as shareConsumerMetrics.forEach.
| totalBytesRead.set(bytesRead.get()); | ||
| } | ||
|
|
||
| private static void consumeMessagesForSingleShareConsumer(long currentTimeMs, |
There was a problem hiding this comment.
This could be more neatly initialised by having lastConsumedTimeMs and currentTimeMs as local variables, and then initialising with
long currentTimeMs = System.currentTimeMills();
long lastConsumedTimeMs = currentTimeMs;
and then going into the while loop.
| props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerHostsAndPorts()); | ||
| props.put(ConsumerConfig.GROUP_ID_CONFIG, options.valueOf(groupIdOpt)); | ||
| props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, options.valueOf(socketBufferSizeOpt).toString()); | ||
| props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, options.valueOf(fetchSizeOpt).toString()); |
There was a problem hiding this comment.
This would be FETCH_MAX_BYTES_CONFIG for a share consumer I think.
About
Added code to measure performance of share consumer/consumers in a share group. Added the following files -
ShareConsumerPerformance.java- Code which measures the performance of share consumer/consumers.ShareConsumerPerformanceTest.java- Contains unit tests for individual functionalities inShareConsumerPerformance.javakafka-share-consumer-perf-test.sh- CLI utility to runShareConsumerPerformance.java