dajac commented on code in PR #13963:
URL: https://github.com/apache/kafka/pull/13963#discussion_r1259657044
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1059,4 +1205,44 @@ public void onNewMetadataImage(MetadataImage newImage,
MetadataDelta delta) {
}
});
}
+
+ /**
+ * The coordinator has been loaded. Session timeouts are registered
+ * for all members.
+ */
+ public void onLoaded() {
+ groups.forEach((groupId, group) -> {
+ switch (group.type()) {
+ case CONSUMER:
+ ConsumerGroup consumerGroup = (ConsumerGroup) group;
+ log.info("Loaded consumer group {} with {} members.",
groupId, consumerGroup.members().size());
+ consumerGroup.members().forEach((memberId, member) -> {
+ log.debug("Loaded member {} in consumer group {}.",
memberId, groupId);
+ scheduleConsumerGroupSessionTimeout(groupId, memberId);
+ if (member.state() ==
ConsumerGroupMember.MemberState.REVOKING) {
+ scheduleConsumerGroupRevocationTimeout(
+ groupId,
+ memberId,
+ member.rebalanceTimeoutMs(),
+ member.memberEpoch()
+ );
+ }
+ });
+ break;
+
+ case GENERIC:
+ GenericGroup genericGroup = (GenericGroup) group;
+ log.info("Loaded generic group {} with {} members.",
groupId, genericGroup.allMembers().size());
Review Comment:
I will you complete this part. I think that we need to log each members as I
did. We also need to schedule the session timeouts here.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
