dajac commented on code in PR #23883:
URL: https://github.com/apache/kafka/pull/23883#discussion_r4350776941
##########
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:
Hmm... Isn't two lookups per topic unecessary? We already allocate a BitSet
so I am not sure to understand why allocating a small record with the bitset
and an int is that problematic. It will only live during the execution of the
assignor, which is short.
--
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]
