squah-confluent commented on code in PR #23883:
URL: https://github.com/apache/kafka/pull/23883#discussion_r4348308890
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/GroupSpecImpl.java:
##########
@@ -93,14 +101,54 @@ public SubscriptionType subscriptionType() {
/**
* {@inheritDoc}
+ * <p>
+ * Partitions which do not exist in the metadata image are reported as
unassigned.
*/
@Override
public boolean isPartitionAssigned(Uuid topicId, int partitionId) {
- Map<Integer, String> partitionMap =
invertedMemberAssignment.get(topicId);
- if (partitionMap == null) {
- return false;
+ if (assignedTopicPartitions == null) {
+ assignedTopicPartitions = computeAssignedTopicPartitions();
+ }
+ BitSet assignedPartitions = assignedTopicPartitions.get(topicId);
+ return assignedPartitions != null &&
assignedPartitions.get(partitionId);
+ }
+
+ private Map<Uuid, BitSet> computeAssignedTopicPartitions() {
+ Map<Uuid, BitSet> assignedTopicPartitions = new HashMap<>();
+ for (MemberSubscriptionAndAssignmentImpl member : members.values()) {
+ for (Map.Entry<Uuid, Set<Integer>> entry :
member.partitions().entrySet()) {
+ BitSet assignedPartitions =
assignedTopicPartitions.computeIfAbsent(
+ entry.getKey(),
+ topicId -> new BitSet(
+ metadataImage.topicMetadata(topicId)
+
.map(CoordinatorMetadataImage.TopicMetadata::partitionCount)
+ .orElse(0)
+ )
+ );
+ for (int partition : entry.getValue()) {
+ if (partition >= assignedPartitions.size()) {
+ // Drop partitions beyond BitSet.size(), to avoid
over-allocating memory in
+ // the BitSet. Note that BitSet.size() rounds up to
the nearest multiple of
+ // 64, so non-existent partitions can still fall
through.
+ continue;
+ }
+
+ assignedPartitions.set(partition);
+ }
+ }
}
- return partitionMap.containsKey(partitionId);
+
+ // The filter for non-existent partitions checks against
BitSet.size(), which is rounded up
+ // to the nearest multiple of 64, so drop the remaining partitions
that do not exist.
+ for (Map.Entry<Uuid, BitSet> entry :
assignedTopicPartitions.entrySet()) {
+ int partitionCount = metadataImage.topicMetadata(entry.getKey())
+ .map(CoordinatorMetadataImage.TopicMetadata::partitionCount)
+ .orElse(0);
+ BitSet assignedPartitions = entry.getValue();
+ assignedPartitions.clear(partitionCount,
assignedPartitions.size());
+ }
Review Comment:
I considered it but was not a fan of it. Unless we make another pass through
the map, we would hold on to the records after the method returns when we no
longer need the extra data.
--
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]
