dongnuo123 commented on code in PR #15442:
URL: https://github.com/apache/kafka/pull/15442#discussion_r1508436563
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3500,6 +3501,42 @@ public void maybeDeleteGroup(String groupId,
List<Record> records) {
}
}
+ /**
+ * A group can be upgraded offline if it's a classic group and empty.
+ *
+ * @param groupId The group to be validated.
+ * @return true if the offline upgrade is valid.
+ */
+ private boolean validateOfflineUpgrade(String groupId) {
+ Group group = groups.get(groupId);
+ return group != null && group.type() == CLASSIC && group.isEmpty();
+ }
+
+ /**
+ * A group can be downgraded offline if it's a consumer group and empty.
+ *
+ * @param groupId The group to be validated.
+ * @return true if the offline downgrade is valid.
+ */
+ private boolean validateOfflineDowngrade(String groupId) {
+ Group group = groups.get(groupId);
+ return group != null && group.type() == CONSUMER && group.isEmpty();
+ }
+
+ /**
+ * Upgrade/Downgrade the empty group if it's valid.
+ *
+ * @param groupId The group id to be migrated.
+ * @param records The list of records to delete the previous group.
+ */
+ public void maybeMigrateEmptyGroup(String groupId, List<Record> records,
boolean isUpgrade) {
+ if ((isUpgrade && validateOfflineUpgrade(groupId)) ||
+ (!isUpgrade && validateOfflineDowngrade(groupId))) {
+ deleteGroup(groupId, records);
+ removeGroup(groupId);
Review Comment:
There is a bug for downgrade. The sequence of downgrading now is 1) consumer
group gets deleted 2) classicGroupJoin creates a classic group and put it into
`groupMetadataManager.groups`3) replaying consumer group tombstone from
`deleteGroup` where `groups.get(groupId)` should be a consumer group.
--
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
