dajac commented on code in PR #15446:
URL: https://github.com/apache/kafka/pull/15446#discussion_r1508764392
##########
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:
Cancelling the timers is not necessary here because all the timers of the
shard are already cancelled when we reach this point. Check the unload method
in the coordinator context.
##########
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()));
+ });
+
+ break;
+ case CLASSIC:
+ ClassicGroup classicGroup = (ClassicGroup) group;
+ log.info("[GroupId={}] Unloading group metadata for
generation {}.",
+ classicGroup.groupId(), classicGroup.generationId());
+
+ classicGroup.transitionTo(DEAD);
+ switch (classicGroup.previousState()) {
+ case EMPTY:
+ case DEAD:
+ break;
+ case PREPARING_REBALANCE:
+ classicGroup.allMembers().forEach(member -> {
+ classicGroup.completeJoinFuture(member, new
JoinGroupResponseData()
+ .setMemberId(member.memberId())
+ .setMembers(Collections.emptyList())
+ .setGenerationId(NO_GENERATION)
+ .setErrorCode(NOT_COORDINATOR.code()));
+
+
timer.cancel(classicGroupHeartbeatKey(classicGroup.groupId(),
member.memberId()));
+ });
+
+
timer.cancel(classicGroupJoinKey(classicGroup.groupId()));
+ break;
+ case COMPLETING_REBALANCE:
+ case STABLE:
+ classicGroup.allMembers().forEach(member -> {
+ classicGroup.completeSyncFuture(member, new
SyncGroupResponseData()
+ .setAssignment(EMPTY_ASSIGNMENT)
Review Comment:
The assignment is empty by default too.
##########
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()));
+ });
+
+ break;
+ case CLASSIC:
+ ClassicGroup classicGroup = (ClassicGroup) group;
+ log.info("[GroupId={}] Unloading group metadata for
generation {}.",
+ classicGroup.groupId(), classicGroup.generationId());
+
+ classicGroup.transitionTo(DEAD);
+ switch (classicGroup.previousState()) {
+ case EMPTY:
+ case DEAD:
+ break;
+ case PREPARING_REBALANCE:
+ classicGroup.allMembers().forEach(member -> {
+ classicGroup.completeJoinFuture(member, new
JoinGroupResponseData()
+ .setMemberId(member.memberId())
+ .setMembers(Collections.emptyList())
+ .setGenerationId(NO_GENERATION)
Review Comment:
Members are empty by default and the generation id is -1 by default too. I
wonder if we could just remove those two.
--
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
