jeffkbkim commented on code in PR #15446:
URL: https://github.com/apache/kafka/pull/15446#discussion_r1539809862
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1920,6 +1921,49 @@ public void onLoaded() {
});
}
+ /**
+ * Called when the partition is unloaded.
+ * ClassicGroup: Complete all awaiting join 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={}] Unloaded group metadata for group
epoch {}.",
+ consumerGroup.groupId(), consumerGroup.groupEpoch());
+ 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())
+ .setErrorCode(NOT_COORDINATOR.code()));
+ });
+
+ break;
+ case COMPLETING_REBALANCE:
+ case STABLE:
+ classicGroup.allMembers().forEach(member -> {
+ classicGroup.completeSyncFuture(member, new
SyncGroupResponseData()
+ .setErrorCode(NOT_COORDINATOR.code()));
+ });
+
+ removeSyncExpiration(classicGroup);
Review Comment:
i missed that. thanks for the catch!
--
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]
