Skip to content

KAFKA-19570: Implement offline migration for streams groups #20288

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lucasbru merged 3 commits into from
Aug 26, 2025
Merged

KAFKA-19570: Implement offline migration for streams groups #20288

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8db6c8c
KAFKA-19570: Implement offline migration for streams groups
lucasbru Aug 1, 2025
79ac3f0
comments
lucasbru Aug 25, 2025
613be0c
Merge remote-tracking branch 'apache/trunk' into offline_migration
lucasbru Aug 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 45 additions & 3 deletions group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -833,19 +833,28 @@ ConsumerGroup getOrMaybeCreateConsumerGroup(
* Gets or creates a streams group without updating the groups map.
* The group will be materialized during the replay.
*
* If there is an empty classic consumer group of the same name, it will be deleted and a new streams
* group will be created.
*
* @param groupId The group ID.
* @param records The record list to which the group tombstones are written
* if the group is empty and is a classic group.
*
* @return A StreamsGroup.
*
* Package private for testing.
*/
StreamsGroup getOrCreateStreamsGroup(
String groupId
String groupId,
List<CoordinatorRecord> records
) {
Group group = groups.get(groupId);

if (group == null) {
return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
} else if (maybeDeleteEmptyClassicGroup(group, records)) {
Comment thread
lucasbru marked this conversation as resolved.
log.info("[GroupId {}] Converted the empty classic group to a streams group.", groupId);
return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
} else {
return castToStreamsGroup(group);
}
Expand Down Expand Up @@ -1871,7 +1880,7 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
boolean isJoining = memberEpoch == 0;
StreamsGroup group;
if (isJoining) {
group = getOrCreateStreamsGroup(groupId);
group = getOrCreateStreamsGroup(groupId, records);
throwIfStreamsGroupIsFull(group);
} else {
group = getStreamsGroupOrThrow(groupId);
Expand Down Expand Up @@ -6066,7 +6075,11 @@ public CoordinatorResult<Void, CoordinatorRecord> classicGroupJoin(
// classicGroupJoinToConsumerGroup takes the join requests to non-empty consumer groups.
// The empty consumer groups should be converted to classic groups in classicGroupJoinToClassicGroup.
return classicGroupJoinToConsumerGroup((ConsumerGroup) group, context, request, responseFuture);
} else if (group.type() == CONSUMER || group.type() == CLASSIC) {
} else if (group.type() == CONSUMER || group.type() == CLASSIC || group.type() == STREAMS && group.isEmpty()) {
// classicGroupJoinToClassicGroup accepts:
// - classic groups
// - empty streams groups
// - empty consumer groups
return classicGroupJoinToClassicGroup(context, request, responseFuture);

@dongnuo123 dongnuo123 Aug 3, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we add a comment to state the groups accepted by classicGroupJoinToClassicGroup here?

  • existing classic group
  • existing empty consumer group (will be converted to classic)
  • existing empty streams group (will be converted to classic)

@lucasbru lucasbru Aug 25, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

} else {
// Group exists but it's not a consumer group
Expand Down Expand Up @@ -6107,6 +6120,8 @@ CoordinatorResult<Void, CoordinatorRecord> classicGroupJoinToClassicGroup(
ClassicGroup group;
if (maybeDeleteEmptyConsumerGroup(groupId, records)) {
log.info("[GroupId {}] Converted the empty consumer group to a classic group.", groupId);
} else if (maybeDeleteEmptyStreamsGroup(groupId, records)) {
log.info("[GroupId {}] Converted the empty streams group to a classic group.", groupId);
}
boolean isNewGroup = !groups.containsKey(groupId);
try {
Expand Down Expand Up @@ -8398,6 +8413,13 @@ private static boolean isEmptyConsumerGroup(Group group) {
return group != null && group.type() == CONSUMER && group.isEmpty();
}

/**
* @return true if the group is an empty streams group.
*/
private static boolean isEmptyStreamsGroup(Group group) {
return group != null && group.type() == STREAMS && group.isEmpty();
}

/**
* Write tombstones for the group if it's empty and is a classic group.
*
Expand Down Expand Up @@ -8435,6 +8457,26 @@ private boolean maybeDeleteEmptyConsumerGroup(String groupId, List<CoordinatorRe
}
return false;
}

/**
* Delete and write tombstones for the group if it's empty and is a streams group.
*
* @param groupId The group id to be deleted.
* @param records The list of records to delete the group.
*
* @return true if the group is an empty streams group.
*/
private boolean maybeDeleteEmptyStreamsGroup(String groupId, List<CoordinatorRecord> records) {
Group group = groups.get(groupId, Long.MAX_VALUE);
if (isEmptyStreamsGroup(group)) {
// Add tombstones for the previous streams group. The tombstones won't actually be
// replayed because its coordinator result has a non-null appendFuture.
createGroupTombstoneRecords(group, records);
removeGroup(groupId);
return true;
}
return false;
}

/**
* Checks whether the given protocol type or name in the request is inconsistent with the group's.
Expand Down
150 changes: 150 additions & 0 deletions ...oordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18633,6 +18633,156 @@ public void testStreamsGroupEndpointInformationOnlyWhenEpochGreater() {
assertNull(result.response().data().partitionsByUserEndpoint());
}

@Test
public void testStreamsGroupHeartbeatWithNonEmptyClassicGroup() {
String classicGroupId = "classic-group-id";
String memberId = Uuid.randomUuid().toString();

String subtopology1 = "subtopology1";
String fooTopicName = "foo";
Topology topology = new Topology().setSubtopologies(List.of(
new Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
));

GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
ClassicGroup classicGroup = new ClassicGroup(
new LogContext(),
classicGroupId,
EMPTY,
context.time
);
context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(classicGroup, classicGroup.groupAssignment()));

context.groupMetadataManager.getOrMaybeCreateClassicGroup(classicGroupId, false).transitionTo(PREPARING_REBALANCE);
assertThrows(GroupIdNotFoundException.class, () ->
context.streamsGroupHeartbeat(
new StreamsGroupHeartbeatRequestData()
.setGroupId(classicGroupId)
.setMemberId(memberId)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(12000)
.setTopology(topology)
.setActiveTasks(List.of())
.setStandbyTasks(List.of())
.setWarmupTasks(List.of())));
}

@Test
public void testStreamsGroupHeartbeatWithEmptyClassicGroup() {
String classicGroupId = "classic-group-id";
String memberId = Uuid.randomUuid().toString();
String fooTopicName = "foo";
String subtopology1 = "subtopology1";
Topology topology = new Topology().setSubtopologies(List.of(
new Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
));

MockTaskAssignor assignor = new MockTaskAssignor("sticky");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
.withStreamsGroupTaskAssignors(List.of(assignor))
.build();
ClassicGroup classicGroup = new ClassicGroup(
new LogContext(),
classicGroupId,
EMPTY,
context.time
);
context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(classicGroup, classicGroup.groupAssignment()));

CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
new StreamsGroupHeartbeatRequestData()
.setGroupId(classicGroupId)
.setMemberId(memberId)
.setMemberEpoch(0)
.setRebalanceTimeoutMs(12000)
.setTopology(topology)
.setActiveTasks(List.of())
.setStandbyTasks(List.of())
.setWarmupTasks(List.of()));

StreamsGroupMember expectedMember = StreamsGroupMember.Builder.withDefaults(memberId)
.setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
.setMemberEpoch(1)
.setPreviousMemberEpoch(0)
.setRebalanceTimeoutMs(5000)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
.setAssignedTasks(TasksTuple.EMPTY)
.setTasksPendingRevocation(TasksTuple.EMPTY)
.setRebalanceTimeoutMs(12000)
.setTopologyEpoch(0)
.build();

assertEquals(Errors.NONE.code(), result.response().data().errorCode());
assertEquals(
List.of(
GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(classicGroupId),
StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(classicGroupId, expectedMember),
StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecord(classicGroupId, topology),
StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord(classicGroupId, 1, 0),
StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord(classicGroupId, memberId, TasksTuple.EMPTY),
StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord(classicGroupId, 1),
StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord(classicGroupId, expectedMember)
),
result.records()
);
assertEquals(
Group.GroupType.STREAMS,
context.groupMetadataManager.streamsGroup(classicGroupId).type()
);
}

@Test
public void testClassicGroupJoinWithEmptyStreamsGroup() throws Exception {
String streamsGroupId = "streams-group-id";
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
.withStreamsGroup(new StreamsGroupBuilder(streamsGroupId, 10))
.build();

JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
.withGroupId(streamsGroupId)
.withMemberId(UNKNOWN_MEMBER_ID)
.withDefaultProtocolTypeAndProtocols()
.build();
GroupMetadataManagerTestContext.JoinResult joinResult = context.sendClassicGroupJoin(request, true);

List<CoordinatorRecord> expectedRecords = List.of(
StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochTombstoneRecord(streamsGroupId),
StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord(streamsGroupId),
StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecordTombstone(streamsGroupId)
);

assertEquals(Errors.MEMBER_ID_REQUIRED.code(), joinResult.joinFuture.get().errorCode());
assertEquals(expectedRecords, joinResult.records.subList(0, expectedRecords.size()));
assertEquals(
Group.GroupType.CLASSIC,
context.groupMetadataManager.getOrMaybeCreateClassicGroup(streamsGroupId, false).type()
);
}

@Test
public void testClassicGroupJoinWithNonEmptyStreamsGroup() throws Exception {
String streamsGroupId = "streams-group-id";
String memberId = Uuid.randomUuid().toString();
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
.withStreamsGroup(new StreamsGroupBuilder(streamsGroupId, 10)
.withMember(StreamsGroupMember.Builder.withDefaults(memberId)
.setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
.setMemberEpoch(10)
.setPreviousMemberEpoch(10)
.build()))
.build();

JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
.withGroupId(streamsGroupId)
.withMemberId(UNKNOWN_MEMBER_ID)
.withDefaultProtocolTypeAndProtocols()
.build();

GroupMetadataManagerTestContext.JoinResult joinResult = context.sendClassicGroupJoin(request);
assertEquals(Errors.INCONSISTENT_GROUP_PROTOCOL.code(), joinResult.joinFuture.get().errorCode());
}

@Test
public void testConsumerGroupDynamicConfigs() {
String groupId = "fooup";
Expand Down