dajac commented on code in PR #15446:
URL: https://github.com/apache/kafka/pull/15446#discussion_r1515870996
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1830,6 +1832,63 @@ public void onLoaded() {
});
}
+ /**
+ * Called when the partition is unloaded. Cancel all existing timers for
the group.
+ * ClassicGroup: Complete all awaiting join futures and sync futures.
Transition group to Dead.
+ */
+ public void onUnloaded() {
+ groups.values().forEach(group -> {
+ switch (group.type()) {
+ case CONSUMER:
+ ConsumerGroup consumerGroup = (ConsumerGroup) group;
+ log.info("[GroupId={}] Unloading group metadata for group
epoch {}.",
+ consumerGroup.groupId(), consumerGroup.groupEpoch());
+
+ consumerGroup.members().values().forEach(member -> {
+
timer.cancel(consumerGroupSessionTimeoutKey(consumerGroup.groupId(),
member.memberId()));
+
timer.cancel(consumerGroupRevocationTimeoutKey(consumerGroup.groupId(),
member.memberId()));
Review Comment:
Yes.
--
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]
