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 11:25:50 UTC
[PR] KAFKA-20292 [7/N]: Split consumer and share TargetAssignmentBuilder [kafka]
squah-confluent opened a new pull request, #23574:
URL: https://github.com/apache/kafka/pull/23574
...and handle membership changes during offloaded consumer and share
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 [7/N]: Split consumer and share TargetAssignmentBuilder [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent merged PR #23574:
URL: https://github.com/apache/kafka/pull/23574
--
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 [7/N]: Split consumer and share TargetAssignmentBuilder [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #23574:
URL: https://github.com/apache/kafka/pull/23574#discussion_r4104298941
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
Review Comment:
nit: We need to update the doc here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
@@ -31,66 +30,31 @@
import org.apache.kafka.coordinator.group.modern.share.ShareGroupMember;
import org.apache.kafka.coordinator.group.util.UnionSet;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
/**
- * Build a new Target Assignment based on the provided parameters. As a result,
- * it yields the records that must be persisted to the log and the new member
- * assignments as a map.
- *
- * Records are only created for members which have a new target assignment. If
- * their assignment did not change, no new record is needed.
- *
- * When a member is deleted, it is assumed that its target assignment record
- * is deleted as part of the member deletion process. In other words, this class
- * does not yield a tombstone for removed members.
+ * Build a new Target Assignment based on the provided parameters.
*/
public abstract class TargetAssignmentBuilder<T extends ModernGroupMember, U extends TargetAssignmentBuilder<T, U>> {
/**
- * The assignment result returned by {{@link TargetAssignmentBuilder#build()}}.
+ * The assignment result returned by {@link TargetAssignmentBuilder#build()}.
+ *
+ * @param targetAssignment The new target assignment for the group.
+ * @param targetAssignmentMetadata The new target assignment metadata.
*/
- public static class TargetAssignmentResult {
- /**
- * The records that must be applied to the __consumer_offsets
- * topics to persist the new target assignment.
- */
- private final List<CoordinatorRecord> records;
-
- /**
- * The new target assignment for the group.
- */
- private final Map<String, MemberAssignment> targetAssignment;
-
- TargetAssignmentResult(
- List<CoordinatorRecord> records,
- Map<String, MemberAssignment> targetAssignment
- ) {
- Objects.requireNonNull(records);
+ public record TargetAssignmentResult(
+ Map<String, Assignment> targetAssignment,
Review Comment:
Why did you chose to use `Assignment` instead of keeping `MemberAssignment`? As `Assignment` implements `MemberAssignment`, we may also have an opportunity here as we may be able to avoid allocating a new `Assignment` for each member.
--
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 [7/N]: Split consumer and share TargetAssignmentBuilder [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23574:
URL: https://github.com/apache/kafka/pull/23574#discussion_r4107167514
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
Review Comment:
Thanks, that does need updating.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
@@ -31,66 +30,31 @@
import org.apache.kafka.coordinator.group.modern.share.ShareGroupMember;
import org.apache.kafka.coordinator.group.util.UnionSet;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
/**
- * Build a new Target Assignment based on the provided parameters. As a result,
- * it yields the records that must be persisted to the log and the new member
- * assignments as a map.
- *
- * Records are only created for members which have a new target assignment. If
- * their assignment did not change, no new record is needed.
- *
- * When a member is deleted, it is assumed that its target assignment record
- * is deleted as part of the member deletion process. In other words, this class
- * does not yield a tombstone for removed members.
+ * Build a new Target Assignment based on the provided parameters.
*/
public abstract class TargetAssignmentBuilder<T extends ModernGroupMember, U extends TargetAssignmentBuilder<T, U>> {
/**
- * The assignment result returned by {{@link TargetAssignmentBuilder#build()}}.
+ * The assignment result returned by {@link TargetAssignmentBuilder#build()}.
+ *
+ * @param targetAssignment The new target assignment for the group.
+ * @param targetAssignmentMetadata The new target assignment metadata.
*/
- public static class TargetAssignmentResult {
- /**
- * The records that must be applied to the __consumer_offsets
- * topics to persist the new target assignment.
- */
- private final List<CoordinatorRecord> records;
-
- /**
- * The new target assignment for the group.
- */
- private final Map<String, MemberAssignment> targetAssignment;
-
- TargetAssignmentResult(
- List<CoordinatorRecord> records,
- Map<String, MemberAssignment> targetAssignment
- ) {
- Objects.requireNonNull(records);
+ public record TargetAssignmentResult(
+ Map<String, Assignment> targetAssignment,
Review Comment:
I can't remember exactly since the code was written a while back. I think it was because it was less friction since `GroupMetadataManager` works with `Assignment`s and it eliminates the outer conversion at the `GroupMetadataManager` level.
I think it's best left to a future refactor since we would want to update `GroupMetadataManager` and `TargetAssignmentRecordsBuilder` too?
--
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 [7/N]: Split consumer and share TargetAssignmentBuilder [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23574:
URL: https://github.com/apache/kafka/pull/23574#discussion_r4126029298
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
@@ -31,66 +30,31 @@
import org.apache.kafka.coordinator.group.modern.share.ShareGroupMember;
import org.apache.kafka.coordinator.group.util.UnionSet;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
/**
- * Build a new Target Assignment based on the provided parameters. As a result,
- * it yields the records that must be persisted to the log and the new member
- * assignments as a map.
- *
- * Records are only created for members which have a new target assignment. If
- * their assignment did not change, no new record is needed.
- *
- * When a member is deleted, it is assumed that its target assignment record
- * is deleted as part of the member deletion process. In other words, this class
- * does not yield a tombstone for removed members.
+ * Build a new Target Assignment based on the provided parameters.
*/
public abstract class TargetAssignmentBuilder<T extends ModernGroupMember, U extends TargetAssignmentBuilder<T, U>> {
/**
- * The assignment result returned by {{@link TargetAssignmentBuilder#build()}}.
+ * The assignment result returned by {@link TargetAssignmentBuilder#build()}.
+ *
+ * @param targetAssignment The new target assignment for the group.
+ * @param targetAssignmentMetadata The new target assignment metadata.
*/
- public static class TargetAssignmentResult {
- /**
- * The records that must be applied to the __consumer_offsets
- * topics to persist the new target assignment.
- */
- private final List<CoordinatorRecord> records;
-
- /**
- * The new target assignment for the group.
- */
- private final Map<String, MemberAssignment> targetAssignment;
-
- TargetAssignmentResult(
- List<CoordinatorRecord> records,
- Map<String, MemberAssignment> targetAssignment
- ) {
- Objects.requireNonNull(records);
+ public record TargetAssignmentResult(
+ Map<String, Assignment> targetAssignment,
Review Comment:
Filed https://issues.apache.org/jira/browse/KAFKA-21177 for the refactor
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
@@ -31,66 +30,31 @@
import org.apache.kafka.coordinator.group.modern.share.ShareGroupMember;
import org.apache.kafka.coordinator.group.util.UnionSet;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
/**
- * Build a new Target Assignment based on the provided parameters. As a result,
- * it yields the records that must be persisted to the log and the new member
- * assignments as a map.
- *
- * Records are only created for members which have a new target assignment. If
- * their assignment did not change, no new record is needed.
- *
- * When a member is deleted, it is assumed that its target assignment record
- * is deleted as part of the member deletion process. In other words, this class
- * does not yield a tombstone for removed members.
+ * Build a new Target Assignment based on the provided parameters.
*/
public abstract class TargetAssignmentBuilder<T extends ModernGroupMember, U extends TargetAssignmentBuilder<T, U>> {
/**
- * The assignment result returned by {{@link TargetAssignmentBuilder#build()}}.
+ * The assignment result returned by {@link TargetAssignmentBuilder#build()}.
+ *
+ * @param targetAssignment The new target assignment for the group.
+ * @param targetAssignmentMetadata The new target assignment metadata.
*/
- public static class TargetAssignmentResult {
- /**
- * The records that must be applied to the __consumer_offsets
- * topics to persist the new target assignment.
- */
- private final List<CoordinatorRecord> records;
-
- /**
- * The new target assignment for the group.
- */
- private final Map<String, MemberAssignment> targetAssignment;
-
- TargetAssignmentResult(
- List<CoordinatorRecord> records,
- Map<String, MemberAssignment> targetAssignment
- ) {
- Objects.requireNonNull(records);
+ public record TargetAssignmentResult(
+ Map<String, Assignment> targetAssignment,
Review Comment:
Filed https://issues.apache.org/jira/browse/KAFKA-21177 for the refactor so that I don't forget
--
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 [7/N]: Split consumer and share TargetAssignmentBuilder [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23574:
URL: https://github.com/apache/kafka/pull/23574#discussion_r4106028259
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
@@ -31,66 +30,31 @@
import org.apache.kafka.coordinator.group.modern.share.ShareGroupMember;
import org.apache.kafka.coordinator.group.util.UnionSet;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
/**
- * Build a new Target Assignment based on the provided parameters. As a result,
- * it yields the records that must be persisted to the log and the new member
- * assignments as a map.
- *
- * Records are only created for members which have a new target assignment. If
- * their assignment did not change, no new record is needed.
- *
- * When a member is deleted, it is assumed that its target assignment record
- * is deleted as part of the member deletion process. In other words, this class
- * does not yield a tombstone for removed members.
+ * Build a new Target Assignment based on the provided parameters.
*/
public abstract class TargetAssignmentBuilder<T extends ModernGroupMember, U extends TargetAssignmentBuilder<T, U>> {
/**
- * The assignment result returned by {{@link TargetAssignmentBuilder#build()}}.
+ * The assignment result returned by {@link TargetAssignmentBuilder#build()}.
+ *
+ * @param targetAssignment The new target assignment for the group.
+ * @param targetAssignmentMetadata The new target assignment metadata.
*/
- public static class TargetAssignmentResult {
- /**
- * The records that must be applied to the __consumer_offsets
- * topics to persist the new target assignment.
- */
- private final List<CoordinatorRecord> records;
-
- /**
- * The new target assignment for the group.
- */
- private final Map<String, MemberAssignment> targetAssignment;
-
- TargetAssignmentResult(
- List<CoordinatorRecord> records,
- Map<String, MemberAssignment> targetAssignment
- ) {
- Objects.requireNonNull(records);
+ public record TargetAssignmentResult(
+ Map<String, Assignment> targetAssignment,
Review Comment:
I can't remember exactly since the code was written a while back. I think it was because it was less friction since `GroupMetadataManager` works with `Assignment`s and it eliminates the outer conversion at the `GroupMetadataManager` level.
I think it's best left to a future refactor since we would want to update ``GroupMetadataManager` and `TargetAssignmentRecordsBuilder` too?
--
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