squah-confluent commented on code in PR #21721:
URL: https://github.com/apache/kafka/pull/21721#discussion_r2929492850
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2727,9 +2727,15 @@ private
CoordinatorResult<Map.Entry<ShareGroupHeartbeatResponseData, Optional<In
member,
updatedMember,
records
- ) || initializedAssignmentPending(group);
+ );
int groupEpoch = group.groupEpoch();
+ if (group.assignmentEpoch() >= groupEpoch) {
+ // A new assignment is needed if no assignment is pending and
there are unassigned
+ // initialized partitions.
+ bumpGroupEpoch |= initializedAssignmentPending(group);
+ }
+
Review Comment:
@smjn Could you check if this change is reasonable?
When assignments are delayed, the original code would bump the epoch
continuously on every heartbeat. After this change, we only bump the epoch due
to `initializedAssignmentPending` when there isn't a pending assignment.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
