KAFKA-18655: Implement the consumer group size counter with scheduled task #18717
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
dongnuo123
commented
Jan 27, 2025
| @@ -71,7 +71,7 @@ public TimelineGaugeCounter(TimelineLong timelineLong, AtomicLong atomicLong) { | |||
| /** | |||
| * Consumer group size gauge counters keyed by the metric name. | |||
| */ | |||
| private final Map<ConsumerGroupState, TimelineGaugeCounter> consumerGroupGauges; | |||
| private volatile Map<ConsumerGroupState, Long> consumerGroupGauges; | |||
There was a problem hiding this comment.
Removed the timeline gauge counter because we recompute the metric from scratch every 60 second, so the result will eventually be consistent despite any rollbacks.
There was a problem hiding this comment.
confirming the concurrency pattern here:
- we have only one reader (metrics collector) and writer (runtime) thread for a shard at any given time.
- we are atomically updating the map to a new map
does this sound right?
jeffkbkim
added
the
KIP-848
jeffkbkim
reviewed
Jan 29, 2025
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
Outdated
Show resolved
Hide resolved
| @@ -71,7 +71,7 @@ public TimelineGaugeCounter(TimelineLong timelineLong, AtomicLong atomicLong) { | |||
| /** | |||
| * Consumer group size gauge counters keyed by the metric name. | |||
| */ | |||
| private final Map<ConsumerGroupState, TimelineGaugeCounter> consumerGroupGauges; | |||
| private volatile Map<ConsumerGroupState, Long> consumerGroupGauges; | |||
There was a problem hiding this comment.
confirming the concurrency pattern here:
- we have only one reader (metrics collector) and writer (runtime) thread for a shard at any given time.
- we are atomically updating the map to a new map
does this sound right?
...r/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java
Outdated
Show resolved
Hide resolved
jeffkbkim
reviewed
Jan 30, 2025
Comment on lines
+136
to
+138
| * Set the number of consumer groups. The method is the only way to update | ||
| * the map and is called by the scheduled task that updates the metrics | ||
| * in {@link org.apache.kafka.coordinator.group.GroupCoordinatorShard}. |
There was a problem hiding this comment.
nit: "This method should be the only way.." and add "..breaking this will result in inconsistent behavior" as well as for setClassicGroupGauges
| * | ||
| * @param state the consumer group state. | ||
| * @param consumerGroupGauges The map counting the number of consumer groups in each state. |
There was a problem hiding this comment.
nit: can we update the classicGroupGauges parameter similarly to this?
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
During testing we discovered that the empty group count is not updated in group conversion, but when the new group is transition to other state, the empty group count is decremented. This could result in negative empty group count.
We can have a new consumer group count implementation that follows the pattern we did for the classic group count. The timeout task periodically refreshes the metrics based on the current groups soft state.
Committer Checklist (excluded from commit message)