You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "DL1231 (via GitHub)" <gi...@apache.org> on 2025/08/21 07:49:34 UTC
[PR] KAFKA-19617: ConsumerPerformance#ConsumerPerfRebListener get corrupted value when the number of partitions is increased [kafka]
DL1231 opened a new pull request, #20388:
URL: https://github.com/apache/kafka/pull/20388
With changes to the consumer protocol, rebalance may not necessarily result in a "stop the world".
Thus, the method for calculating pause time in `ConsumerPerformance#ConsumerPerfRebListener` needs to be modified.
Stop time is only recorded if `assignedPartitions` is empty.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19617: ConsumerPerformance#ConsumerPerfRebListener get corrupted value when the number of partitions is increased [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20388:
URL: https://github.com/apache/kafka/pull/20388#issuecomment-3235588781
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19617: ConsumerPerformance#ConsumerPerfRebListener get corrupted value when the number of partitions is increased [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #20388:
URL: https://github.com/apache/kafka/pull/20388#issuecomment-3336557261
@chia7712 @AndrewJSchofield @apoorvmittal10 PTAL when you get a chance, thanks in advance.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19617: ConsumerPerformance#ConsumerPerfRebListener get corrupted value when the number of partitions is increased [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20388:
URL: https://github.com/apache/kafka/pull/20388#discussion_r2381595668
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -230,24 +231,32 @@ private static void printExtendedProgress(long bytesRead,
public static class ConsumerPerfRebListener implements ConsumerRebalanceListener {
private final AtomicLong joinTimeMs;
private final AtomicLong joinTimeMsInSingleRound;
+ private final Collection<TopicPartition> assignedPartitions;
private long joinStartMs;
public ConsumerPerfRebListener(AtomicLong joinTimeMs, long joinStartMs, AtomicLong joinTimeMsInSingleRound) {
this.joinTimeMs = joinTimeMs;
this.joinStartMs = joinStartMs;
this.joinTimeMsInSingleRound = joinTimeMsInSingleRound;
+ this.assignedPartitions = new HashSet<>();
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
- joinStartMs = System.currentTimeMillis();
+ assignedPartitions.removeAll(partitions);
+ if (assignedPartitions.isEmpty()) {
+ joinStartMs = System.currentTimeMillis();
+ }
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
- long elapsedMs = System.currentTimeMillis() - joinStartMs;
- joinTimeMs.addAndGet(elapsedMs);
- joinTimeMsInSingleRound.addAndGet(elapsedMs);
+ if (assignedPartitions.isEmpty()) {
+ long elapsedMs = System.currentTimeMillis() - joinStartMs;
+ joinTimeMs.addAndGet(elapsedMs);
+ joinTimeMsInSingleRound.addAndGet(elapsedMs);
Review Comment:
I know this isn't strictly fixing the same problem, but isn't `joinTimeMsInSingleRound` flawed. The problem is that the rebalance listener is passed a reference to an `AtomicLong` in its constructor, but the loop in the `consume()` method creates a new `AtomicLong` in each reporting interval. It probably should set the current value to zero on each reporting interval instead. wdyt?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19617: ConsumerPerformance#ConsumerPerfRebListener get corrupted value when the number of partitions is increased [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #20388:
URL: https://github.com/apache/kafka/pull/20388
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19617: ConsumerPerformance#ConsumerPerfRebListener get corrupted value when the number of partitions is increased [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20388:
URL: https://github.com/apache/kafka/pull/20388#discussion_r2386189192
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -230,24 +231,32 @@ private static void printExtendedProgress(long bytesRead,
public static class ConsumerPerfRebListener implements ConsumerRebalanceListener {
private final AtomicLong joinTimeMs;
private final AtomicLong joinTimeMsInSingleRound;
+ private final Collection<TopicPartition> assignedPartitions;
private long joinStartMs;
public ConsumerPerfRebListener(AtomicLong joinTimeMs, long joinStartMs, AtomicLong joinTimeMsInSingleRound) {
this.joinTimeMs = joinTimeMs;
this.joinStartMs = joinStartMs;
this.joinTimeMsInSingleRound = joinTimeMsInSingleRound;
+ this.assignedPartitions = new HashSet<>();
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
- joinStartMs = System.currentTimeMillis();
+ assignedPartitions.removeAll(partitions);
+ if (assignedPartitions.isEmpty()) {
Review Comment:
Should we change the word "rebalance.time.ms" to match the change?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19617: ConsumerPerformance#ConsumerPerfRebListener get corrupted value when the number of partitions is increased [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20388:
URL: https://github.com/apache/kafka/pull/20388#discussion_r2384437256
##########
tools/src/main/java/org/apache/kafka/tools/ConsumerPerformance.java:
##########
@@ -230,24 +231,32 @@ private static void printExtendedProgress(long bytesRead,
public static class ConsumerPerfRebListener implements ConsumerRebalanceListener {
private final AtomicLong joinTimeMs;
private final AtomicLong joinTimeMsInSingleRound;
+ private final Collection<TopicPartition> assignedPartitions;
private long joinStartMs;
public ConsumerPerfRebListener(AtomicLong joinTimeMs, long joinStartMs, AtomicLong joinTimeMsInSingleRound) {
this.joinTimeMs = joinTimeMs;
this.joinStartMs = joinStartMs;
this.joinTimeMsInSingleRound = joinTimeMsInSingleRound;
+ this.assignedPartitions = new HashSet<>();
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
- joinStartMs = System.currentTimeMillis();
+ assignedPartitions.removeAll(partitions);
+ if (assignedPartitions.isEmpty()) {
+ joinStartMs = System.currentTimeMillis();
+ }
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
- long elapsedMs = System.currentTimeMillis() - joinStartMs;
- joinTimeMs.addAndGet(elapsedMs);
- joinTimeMsInSingleRound.addAndGet(elapsedMs);
+ if (assignedPartitions.isEmpty()) {
+ long elapsedMs = System.currentTimeMillis() - joinStartMs;
+ joinTimeMs.addAndGet(elapsedMs);
+ joinTimeMsInSingleRound.addAndGet(elapsedMs);
Review Comment:
Good catch! Fixed now. 👍
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org