dajac commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2081913011
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -19,41 +19,58 @@
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignor;
import
org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.metadata.PartitionRegistration;
+import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
/**
* The subscribed topic metadata class is used by the {@link
PartitionAssignor} to obtain
* topic and partition metadata for the topics that the modern group is
subscribed to.
*/
public class SubscribedTopicDescriberImpl implements SubscribedTopicDescriber {
/**
- * The topic Ids mapped to their corresponding {@link TopicMetadata}
- * object, which contains topic and partition metadata.
+ * The topic Ids set.
*/
- private final Map<Uuid, TopicMetadata> topicMetadata;
+ private final Set<Uuid> subscriptionTopicIdSet;
private final Map<Uuid, Set<Integer>> topicPartitionAllowedMap;
- public SubscribedTopicDescriberImpl(Map<Uuid, TopicMetadata>
topicMetadata) {
- this(topicMetadata, null);
+ /**
+ * The metadata image that contains the latest metadata information.
+ */
+ private final MetadataImage metadataImage;
+
+ public SubscribedTopicDescriberImpl(Set<Uuid> subscriptionTopicIdSet,
MetadataImage metadataImage) {
+ this(subscriptionTopicIdSet, metadataImage, null);
}
- public SubscribedTopicDescriberImpl(Map<Uuid, TopicMetadata>
topicMetadata, Map<Uuid, Set<Integer>> topicPartitionAllowedMap) {
- this.topicMetadata = Objects.requireNonNull(topicMetadata);
+ public SubscribedTopicDescriberImpl(Set<Uuid> subscriptionTopicIdSet,
MetadataImage metadataImage, Map<Uuid, Set<Integer>> topicPartitionAllowedMap) {
+ this.subscriptionTopicIdSet =
Objects.requireNonNull(subscriptionTopicIdSet);
+ this.metadataImage = Objects.requireNonNull(metadataImage);
this.topicPartitionAllowedMap = topicPartitionAllowedMap;
}
/**
- * Map of topic Ids to topic metadata.
+ * Set of topic Ids.
+ *
+ * @return The set of topic Ids.
+ */
+ public Set<Uuid> subscriptionTopicIdSet() {
+ return this.subscriptionTopicIdSet;
Review Comment:
Given how this is used (mainly in tests), I think that we should rather try
to remove it. Would it be possible?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3652,11 +3652,11 @@ private Assignment updateTargetAssignment(
new
TargetAssignmentBuilder.ConsumerTargetAssignmentBuilder(group.groupId(),
groupEpoch, consumerGroupAssignors.get(preferredServerAssignor))
.withMembers(group.members())
.withStaticMembers(group.staticMembers())
- .withSubscriptionMetadata(subscriptionMetadata)
+
.withSubscriptionTopicIdSet(subscriptionMetadata.values().stream().map(TopicMetadata::id).collect(Collectors.toSet()))
Review Comment:
I think that our goal is to get rid of `subscriptionMetadata` so it is a bit
surprising to still rely on it here. Could we avoid it?
--
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
