Skip to content

auto create share group when alter offset #155

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
showuon merged 1 commit into from
May 8, 2026
Merged

auto create share group when alter offset #155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
3 changes: 2 additions & 1 deletion ...p-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -800,7 +800,8 @@ public CoordinatorResult<Map.Entry<AlterShareGroupOffsetsResponseData, Initializ
Map.Entry<AlterShareGroupOffsetsResponseData, InitializeShareGroupStateParameters> response = groupMetadataManager.completeAlterShareGroupOffsets(
groupId,
alterShareGroupOffsetsRequestData,
records
records,
group
);
return new CoordinatorResult<>(
records,
Expand Down
11 changes: 6 additions & 5 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 @@ -1171,10 +1171,11 @@ public ShareGroup shareGroup(
String groupId,
long committedOffset
) throws GroupIdNotFoundException {
Group group = group(groupId, committedOffset);
// Get or create the share group. If the group exists, check that it's empty. If it is created, it is empty.
final ShareGroup group = getOrMaybeCreateShareGroup(groupId, true);

@gaurav-narula gaurav-narula May 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

I think this is causing test failures in CI. Can you please have a look @showuon when you get a chance?


if (group.type() == SHARE) {
return (ShareGroup) group;
return group;
} else {
// We don't support upgrading/downgrading between protocols at the moment so
// we throw an exception if a group exists with the wrong type.
Expand Down Expand Up @@ -8190,10 +8191,10 @@ public List<DeleteShareGroupStateRequestData.DeleteStateData> sharePartitionsEli
public Map.Entry<AlterShareGroupOffsetsResponseData, InitializeShareGroupStateParameters> completeAlterShareGroupOffsets(
String groupId,
AlterShareGroupOffsetsRequestData alterShareGroupOffsetsRequest,
List<CoordinatorRecord> records
List<CoordinatorRecord> records,
ShareGroup group
) {
final long currentTimeMs = time.milliseconds();
Group group = groups.get(groupId);
AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection alterShareGroupOffsetsResponseTopics = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection();

Map<Uuid, InitMapValue> initializingTopics = new HashMap<>();
Expand Down Expand Up @@ -8256,7 +8257,7 @@ public Map.Entry<AlterShareGroupOffsetsResponseData, InitializeShareGroupStatePa
return Map.entry(
new AlterShareGroupOffsetsResponseData()
.setResponses(alterShareGroupOffsetsResponseTopics),
buildInitializeShareGroupState(groupId, ((ShareGroup) group).groupEpoch(), offsetByTopicPartitions)
buildInitializeShareGroupState(groupId, group.groupEpoch(), offsetByTopicPartitions)
);
}

Expand Down
Loading