You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "squah-confluent (via GitHub)" <gi...@apache.org> on 2026/09/24 12:02:22 UTC
[PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
squah-confluent opened a new pull request, #23575:
URL: https://github.com/apache/kafka/pull/23575
...and handle membership changes during offloaded streams assignments.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23575:
URL: https://github.com/apache/kafka/pull/23575#discussion_r4105394892
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4627,7 +4625,14 @@ private UpdateTargetAssignmentResult<Map<String, TasksTuple>> maybeUpdateStreams
group.groupId(), groupEpoch, assignor, assignorTimeMs);
}
- records.addAll(assignmentResult.records());
+ new TargetAssignmentRecordsBuilder.StreamsTargetAssignmentRecordsBuilder(log, group.groupId())
+ .withTargetAssignmentMetadata(assignmentResult.targetAssignmentMetadata())
+ .withCurrentMemberIds(updatedMembersAndTargetAssignment.members().keySet())
+ .withPreviousStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
+ .withCurrentStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
Review Comment:
The distinction between previous and current is meant for the async assignor path. The previous static members are the static members the assignment was computed off. The current static members are the current state of the group, which can have changed after the assignment was computed but before it was applied.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23575:
URL: https://github.com/apache/kafka/pull/23575#discussion_r4106987366
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4627,7 +4625,14 @@ private UpdateTargetAssignmentResult<Map<String, TasksTuple>> maybeUpdateStreams
group.groupId(), groupEpoch, assignor, assignorTimeMs);
}
- records.addAll(assignmentResult.records());
+ new TargetAssignmentRecordsBuilder.StreamsTargetAssignmentRecordsBuilder(log, group.groupId())
+ .withTargetAssignmentMetadata(assignmentResult.targetAssignmentMetadata())
+ .withCurrentMemberIds(updatedMembersAndTargetAssignment.members().keySet())
+ .withPreviousStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
+ .withCurrentStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
Review Comment:
Ah, so the distinction is yet to come. Got it, thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23575:
URL: https://github.com/apache/kafka/pull/23575#discussion_r4104809094
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4627,7 +4625,14 @@ private UpdateTargetAssignmentResult<Map<String, TasksTuple>> maybeUpdateStreams
group.groupId(), groupEpoch, assignor, assignorTimeMs);
}
- records.addAll(assignmentResult.records());
+ new TargetAssignmentRecordsBuilder.StreamsTargetAssignmentRecordsBuilder(log, group.groupId())
+ .withTargetAssignmentMetadata(assignmentResult.targetAssignmentMetadata())
+ .withCurrentMemberIds(updatedMembersAndTargetAssignment.members().keySet())
+ .withPreviousStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
+ .withCurrentStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
Review Comment:
Why are we passing "updatedMembersAndTargetAssignment.staticMembers" as previous static members and current static members? Maybe I am missing somethign
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23575:
URL: https://github.com/apache/kafka/pull/23575#discussion_r4133058737
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4627,7 +4625,14 @@ private UpdateTargetAssignmentResult<Map<String, TasksTuple>> maybeUpdateStreams
group.groupId(), groupEpoch, assignor, assignorTimeMs);
}
- records.addAll(assignmentResult.records());
+ new TargetAssignmentRecordsBuilder.StreamsTargetAssignmentRecordsBuilder(log, group.groupId())
+ .withTargetAssignmentMetadata(assignmentResult.targetAssignmentMetadata())
+ .withCurrentMemberIds(updatedMembersAndTargetAssignment.members().keySet())
+ .withPreviousStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
+ .withCurrentStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
Review Comment:
I opened #23622 for it. Your thoughts on the approach are appreciated!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #23575:
URL: https://github.com/apache/kafka/pull/23575#discussion_r4121727866
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4627,7 +4625,14 @@ private UpdateTargetAssignmentResult<Map<String, TasksTuple>> maybeUpdateStreams
group.groupId(), groupEpoch, assignor, assignorTimeMs);
}
- records.addAll(assignmentResult.records());
+ new TargetAssignmentRecordsBuilder.StreamsTargetAssignmentRecordsBuilder(log, group.groupId())
+ .withTargetAssignmentMetadata(assignmentResult.targetAssignmentMetadata())
+ .withCurrentMemberIds(updatedMembersAndTargetAssignment.members().keySet())
+ .withPreviousStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
+ .withCurrentStaticMembers(updatedMembersAndTargetAssignment.staticMembers())
Review Comment:
@squah-confluent It seems that we have an opportunity to skip the computation between the previous and the existing state members when offloading path is not used. Computing the diff is wasteful is this case. We can address it separately though.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23575:
URL: https://github.com/apache/kafka/pull/23575#discussion_r4107170551
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
Review Comment:
Thanks, that does need updating.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #23575:
URL: https://github.com/apache/kafka/pull/23575#discussion_r4104372927
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
Review Comment:
nit: We need to update the doc.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [8/N]: Split streams TargetAssignmentBuilder [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent merged PR #23575:
URL: https://github.com/apache/kafka/pull/23575
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org