squah-confluent commented on code in PR #23801:
URL: https://github.com/apache/kafka/pull/23801#discussion_r4264335722
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/uniform2/IndexedGroup.java:
##########
@@ -29,30 +29,35 @@
import java.util.Map;
/**
- * The members, the subscribed topics and the cohorts of a group, numbered by
small integers.
+ * The members, the subscribed topics and the cohorts of a group, numbered by
small integers. It is
+ * built by {@link Builder}.
*
* <p>Members are numbered in member id order and topics in topic id order, so
that the assignment
* is deterministic and does not depend on the iteration order of the
collections in the spec.
*
* <p>A cohort is a set of members having the same subscription. Cohorts are
numbered in the order
* of their first member. Grouping the members by subscription lets the
assignor store and process
- * every distinct subscription once, however many members share it.
+ * every distinct subscription once, however many members share it. A group
with a homogeneous
+ * subscription type is a single cohort; an empty group is a single cohort
without members.
*
* <p>Only topics subscribed by at least one member and which exist are
retained.
- *
- * <p>With a homogeneous subscription type, only the subscription of the first
member is read, as
- * the coordinator guarantees that every member has the same one, and the
group is a single cohort;
- * an empty group is a single cohort without members. Otherwise every
subscription is read once and
- * looked up by a hash of its topics that does not depend on their order, see
- * {@link CohortsBySubscription}. The cohorts with the same hash are compared
exactly, so members
- * are never assigned to the wrong cohort.
*/
final class IndexedGroup {
/**
- * Reads the subscriptions of the members, indexing their topics that
exist in the order they
- * are seen, and remembering their partition counts.
+ * Builds the {@link IndexedGroup} of a group spec.
+ *
+ * <p>With a homogeneous subscription type, only the subscription of the
first member is read,
+ * as the coordinator guarantees that every member has the same one.
Otherwise every
+ * subscription is read once and looked up by a hash of its topics that
does not depend on their
+ * order. The cohorts with the same hash are compared exactly, so members
are never assigned to
+ * the wrong cohort.
+ *
+ * <p>A subscription is only iterated, so it may be a {@code TopicIds}.
The topics that do not
+ * exist are dropped from it.
Review Comment:
nit: Can we spell out what's special about TopicIds?
```suggestion
* <p>A subscription is only iterated, so it may be a {@code TopicIds},
whose size() counts
* topic names that do not resolve. The topics that do not exist are
dropped from the iterator.
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/uniform2/IndexedGroup.java:
##########
@@ -64,109 +69,218 @@ private static final class SubscriptionReader {
private final SubscribedTopicDescriber subscribedTopicDescriber;
/**
- * Per topic seen, its index in the order seen, or -1 if it does not
exist, so that the
- * describer is asked once per topic.
+ * The mask applied to the hash of every subscription: all the bits in
the assignor, none in
+ * a test that makes every subscription collide.
*/
- private final Map<Uuid, Integer> topicIndices = new HashMap<>();
+ private final long hashMask;
/**
- * The topics seen that exist, in the order seen.
+ * The member ids, in ascending order.
*/
- private final List<Uuid> topicIds = new ArrayList<>();
+ private final String[] memberIds;
/**
- * The partition counts of the topics seen that exist, in the order
seen.
+ * Per member, its cohort.
*/
- private final IntList partitionCounts = new IntList(16);
+ private final int[] cohortOfMember;
- SubscriptionReader(GroupSpec groupSpec, SubscribedTopicDescriber
subscribedTopicDescriber) {
- this.groupSpec = groupSpec;
- this.subscribedTopicDescriber = subscribedTopicDescriber;
- }
+ /**
+ * Per topic id seen in a subscription, its index, or -1 if the topic
does not exist, so
+ * that the describer is asked once per topic. Indexed in the order
seen, then in topic id
+ * order by {@link #build()}.
+ */
+ private final Map<Uuid, Integer> topicIndices = new HashMap<>();
/**
- * Reads the subscription of the member.
- *
- * @param memberId The member id.
- * @param topics Receives the indices of its subscribed topics that
exist.
+ * The topics seen that exist, in the order seen.
*/
- void readSubscribedTopics(String memberId, IntList topics) {
- // Only the iteration is used: the subscription may be a TopicIds,
whose size() also
- // counts the names that do not resolve.
- for (Uuid topicId :
groupSpec.memberSubscription(memberId).subscribedTopicIds()) {
- var index = topicIndices.get(topicId);
- if (index == null) {
- // Seen for the first time: assigned the next index,
unless it does not exist.
- int partitionCount =
subscribedTopicDescriber.numPartitions(topicId);
- index = partitionCount < 0 ? -1 : topicIds.size();
- if (index >= 0) {
- topicIds.add(topicId);
- partitionCounts.add(partitionCount);
- }
- topicIndices.put(topicId, index);
- }
- if (index >= 0) {
- topics.add(index);
- }
- }
- }
- }
+ private final List<Uuid> seenTopicIds = new ArrayList<>();
- /**
- * A hash map from subscriptions to cohorts.
- *
- * <p>A subscription is summarized by a hash that does not depend on the
order of its topics,
- * and the cohorts with the same hash are listed together, usually one per
hash.
- *
- * <p>Subscription equality is tested by marking the topics of the
subscription being looked
- * up in an array, then checking that a cohort has as many topics, all of
them marked. Unlike a
- * {@code HashMap} keyed by the subscription, this needs no sorted copy of
every member's
- * subscription.
- */
- static final class CohortsBySubscription {
/**
- * The mask applied to every hash: all the bits in the assignor, none
in a test that makes
- * every subscription collide.
+ * The partition counts of the topics seen that exist, in the order
seen.
*/
- private final long hashMask;
+ private final IntList seenPartitionCounts = new IntList(16);
/**
- * Per cohort, its subscribed topics, indexed in the order they were
seen.
+ * Per cohort, its subscribed topics. Indexed in the order seen, then
in topic id order by
+ * {@link #build()}.
*/
private final List<int[]> subscriptions = new ArrayList<>();
/**
- * Per hash of a subscription, the cohorts whose subscription has that
hash.
+ * Per hash of a subscription, the cohorts whose subscription has that
hash, usually one.
Review Comment:
nit: The last part reads unnaturally. I can't put my finger on why.
```suggestion
* Per hash of a subscription, the cohorts whose subscription has
that hash. There is
* usually only one cohort per hash.
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
