dajac commented on code in PR #16068:
URL: https://github.com/apache/kafka/pull/16068#discussion_r1613460654
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilder.java:
##########
@@ -277,13 +277,22 @@ public TargetAssignmentBuilder removeMember(
*/
public TargetAssignmentResult build() throws PartitionAssignorException {
Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
+ Map<String, Map<Uuid, Set<Integer>>> assignedPartitions = new
HashMap<>(members.size());
Review Comment:
Hum... Do we really need this? Would it be possible to pass
`targetAssignment` to GroupSpecImpl like we did with
`invertedTargetAssignment`? This will save us from having to create a new
HashMap with the same data.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberSubscriptionSpecImpl.java:
##########
@@ -0,0 +1,69 @@
+package org.apache.kafka.coordinator.group.assignor;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Implementation of the {@link MemberSubscriptionSpec} interface.
+ */
+public class MemberSubscriptionSpecImpl implements MemberSubscriptionSpec {
+ private final String memberId;
+ private final Optional<String> rackId;
+ private final Set<Uuid> subscribedTopicIds;
+
+ /**
+ * Constructs a new {@code MemberSubscriptionSpecImpl}.
+ *
+ * @param memberId The member Id.
+ * @param rackId The rack Id.
+ * @param subscribedTopicIds The set of subscribed topic Ids.
+ */
+ public MemberSubscriptionSpecImpl(String memberId, Optional<String>
rackId, Set<Uuid> subscribedTopicIds) {
+ this.memberId = memberId;
+ this.rackId = rackId;
+ this.subscribedTopicIds = subscribedTopicIds;
Review Comment:
nit: Let's ensure that they are non-null with `Objects.requireNonNull`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImpl.java:
##########
@@ -35,6 +37,11 @@ public class GroupSpecImpl implements GroupSpec {
*/
private final SubscriptionType subscriptionType;
+ /**
+ * Partitions assigned keyed by topicId.
+ */
+ private final Map<String, Map<Uuid, Set<Integer>>> assignedPartitions;
Review Comment:
Is this the targetAssignment? If so, should we call it targetAssignment to
be aligned with invertedTargetAssignment?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupSpec.java:
##########
@@ -26,9 +26,9 @@
*/
public interface GroupSpec {
/**
- * @return Member metadata keyed by member Id.
+ * @return Member subscription metadata keyed by member Id.
*/
- Map<String, MemberSubscriptionSpec> members();
+ Map<String, MemberSubscriptionSpec> memberSubscriptions();
Review Comment:
I think that we also wanted to return a `Collection` here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupSpec.java:
##########
@@ -39,4 +40,12 @@ public interface GroupSpec {
* False, otherwise.
*/
boolean isPartitionAssigned(Uuid topicId, int partitionId);
+
+ /**
+ * Gets the current assignment for a member.
+ *
+ * @param memberId The member Id.
+ * @return A map of topic Ids to sets of partition numbers.
+ */
+ Map<Uuid, Set<Integer>> currentMemberAssignment(String memberId);
Review Comment:
nit: `assignment` or `memberAssignment`?
--
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
