dajac commented on code in PR #15446:
URL: https://github.com/apache/kafka/pull/15446#discussion_r1515878176
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -9607,6 +9609,104 @@ public void
testOnConsumerGroupStateTransitionOnLoading() {
verify(context.metrics,
times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.EMPTY,
null);
}
+ @Test
+ public void testClassicGroupOnUnloadedEmptyAndPreparingRebalance() throws
Exception {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ ClassicGroup emptyGroup = context.createClassicGroup("empty-group");
+ assertTrue(emptyGroup.isInState(EMPTY));
+
+ ClassicGroup preparingGroup =
context.createClassicGroup("preparing-group");
+ JoinGroupRequestData request = new
GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
Review Comment:
Is it worth having two members in the group?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -639,6 +639,7 @@ public void onLoaded(MetadataImage newImage) {
public void onUnloaded() {
timer.cancel(GROUP_EXPIRATION_KEY);
coordinatorMetrics.deactivateMetricsShard(metricsShard);
+ groupMetadataManager.onUnloaded();
Review Comment:
Should we extend unit test to cover this change?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1830,6 +1831,54 @@ 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={}] 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()));
+
+
timer.cancel(classicGroupHeartbeatKey(classicGroup.groupId(),
member.memberId()));
Review Comment:
We can also remove the cancellations for the classic group case.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1830,6 +1831,54 @@ public void onLoaded() {
});
}
+ /**
+ * Called when the partition is unloaded. Cancel all existing timers for
the group.
Review Comment:
nit: We could remove `Cancel all existing timers for the group.`
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1830,6 +1832,64 @@ public void onLoaded() {
});
}
+ // TODO: this part is done asynchronously in the old coordinator via a
KafkaScheduler#schedule
+ /**
Review Comment:
That's a good point. We need to observe how it works. If it causes
performance issues, we could move this to the loader and use the loader to load
and unload.
--
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]
