You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2025/05/01 12:49:14 UTC
[PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl (wip) [kafka]
FrankYang0529 opened a new pull request, #19611:
URL: https://github.com/apache/kafka/pull/19611
Replace `TopicMetadata` with `subscriptionTopicIdSet` and `MetadataImage` in `SubscribedTopicDescriberImpl`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #19611:
URL: https://github.com/apache/kafka/pull/19611
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094976739
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -112,7 +112,7 @@ public Set<Integer> assignablePartitions(Uuid topicId) {
}
if (topicPartitionAllowedMap.isEmpty()) {
- return Set.copyOf(topic.partitions().keySet());
+ return topic.partitions().keySet();
Review Comment:
nit: I would use Collections.unmodifiableSet to ensure that it cannot be mutated by the caller.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19611:
URL: https://github.com/apache/kafka/pull/19611#issuecomment-2901178153
@AndrewJSchofield thanks for this report. we will check it ASAP
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094372479
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -19,41 +19,42 @@
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 map of topic Ids to the set of allowed partitions for each topic.
+ * If this is empty, all partitions are allowed.
*/
- private final Map<Uuid, TopicMetadata> topicMetadata;
- private final Map<Uuid, Set<Integer>> topicPartitionAllowedMap;
+ private final Optional<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(Map<Uuid, TopicMetadata> topicMetadata, Map<Uuid, Set<Integer>> topicPartitionAllowedMap) {
- this.topicMetadata = Objects.requireNonNull(topicMetadata);
- this.topicPartitionAllowedMap = topicPartitionAllowedMap;
+ public SubscribedTopicDescriberImpl(MetadataImage metadataImage) {
+ this(metadataImage, Optional.empty());
}
- /**
- * Map of topic Ids to topic metadata.
- *
- * @return The map of topic Ids to topic metadata.
- */
- public Map<Uuid, TopicMetadata> topicMetadata() {
- return this.topicMetadata;
+ public SubscribedTopicDescriberImpl(
+ MetadataImage metadataImage,
+ Optional<Map<Uuid, Set<Integer>>> topicPartitionAllowedMap
+ ) {
+ this.metadataImage = Objects.requireNonNull(metadataImage);
+ this.topicPartitionAllowedMap = topicPartitionAllowedMap;
Review Comment:
Yes, updated it. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094372605
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##########
@@ -61,15 +63,12 @@ public class OptimizedUniformAssignmentBuilderTest {
@Test
public void testOneMemberNoTopicSubscription() {
+ MetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(topic1Uuid, topic1Name, 3)
+ .addRacks()
Review Comment:
Currently, assignor doesn't use `racksForPartition`, so it's not necessary. Remove it from assignor test cases.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #19611:
URL: https://github.com/apache/kafka/pull/19611#issuecomment-2876171380
> Out of curiosity, have you tried to run the benchmarks before and after the change to see the impact?
Since previous `SubscribedTopicDescriberImpl` only supports `numOfPartitions`, I only test this function. The result is 10^(-5) vs 10^(-6).
### This PR
```
Benchmark (partitionsPerTopic) (topicCount) Mode Cnt Score Error Units
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 5 10 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 5 100 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 5 1000 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 10 10 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 10 100 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 10 1000 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 50 10 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 50 100 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 50 1000 avgt 5 ≈ 10⁻⁵ ms/op
JMH benchmarks done
```
<details>
<summary> SubscribedTopicDescriberImplBenchmark.java
</summary>
```java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.jmh.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.modern.SubscribedTopicDescriberImpl;
import org.apache.kafka.coordinator.group.modern.TopicIds;
import org.apache.kafka.image.MetadataImage;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Warmup;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@State(Scope.Benchmark)
@Fork(value = 1)
@Warmup(iterations = 3)
@Measurement(iterations = 5)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
public class SubscribedTopicDescriberImplBenchmark {
@Param({"5", "10", "50"})
private int partitionsPerTopic;
@Param({"10", "100", "1000"})
private int topicCount;
private List<String> allTopicNames = List.of();
private Set<Uuid> subscriptionTopicIdSet;
private MetadataImage metadataImage;
private TopicIds.TopicResolver topicResolver;
private SubscribedTopicDescriber subscribedTopicDescriber;
private String topicName;
private Uuid topicId;
private void setupTopics() {
allTopicNames = AssignorBenchmarkUtils.createTopicNames(topicCount);
topicName = allTopicNames.get(0);
metadataImage = AssignorBenchmarkUtils.createMetadataImage(allTopicNames, partitionsPerTopic);
topicResolver = new TopicIds.CachedTopicResolver(metadataImage.topics());
topicId = topicResolver.id(topicName);
subscriptionTopicIdSet = AssignorBenchmarkUtils.subscriptionTopicIdSet(metadataImage);
subscribedTopicDescriber = new SubscribedTopicDescriberImpl(subscriptionTopicIdSet, metadataImage);
}
@Setup(Level.Trial)
public void setup() {
setupTopics();
}
@Benchmark
public void testNumOfPartitions() {
subscribedTopicDescriber.numPartitions(topicId);
}
}
```
</details>
### trunk
```
Benchmark (partitionsPerTopic) (topicCount) Mode Cnt Score Error Units
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 5 10 avgt 5 ≈ 10⁻⁶ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 5 100 avgt 5 ≈ 10⁻⁶ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 5 1000 avgt 5 ≈ 10⁻⁶ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 10 10 avgt 5 ≈ 10⁻⁶ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 10 100 avgt 5 ≈ 10⁻⁶ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 10 1000 avgt 5 ≈ 10⁻⁶ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 50 10 avgt 5 ≈ 10⁻⁵ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 50 100 avgt 5 ≈ 10⁻⁶ ms/op
SubscribedTopicDescriberImplBenchmark.testNumOfPartitions 50 1000 avgt 5 ≈ 10⁻⁶ ms/op
JMH benchmarks done
```
<details>
<summary> SubscribedTopicDescriberImplBenchmark.java
</summary>
```java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.jmh.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.modern.SubscribedTopicDescriberImpl;
import org.apache.kafka.coordinator.group.modern.TopicIds;
import org.apache.kafka.coordinator.group.modern.TopicMetadata;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.image.TopicsImage;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Warmup;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@State(Scope.Benchmark)
@Fork(value = 1)
@Warmup(iterations = 3)
@Measurement(iterations = 5)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
public class SubscribedTopicDescriberImplBenchmark {
@Param({"5", "10", "50"})
private int partitionsPerTopic;
@Param({"10", "100", "1000"})
private int topicCount;
private List<String> allTopicNames = List.of();
private Map<String, TopicMetadata> subscriptionMetadata;
private TopicsImage topicsImage;
private TopicIds.TopicResolver topicResolver;
private SubscribedTopicDescriber subscribedTopicDescriber;
private String topicName;
private Uuid topicId;
private void setupTopics() {
allTopicNames = AssignorBenchmarkUtils.createTopicNames(topicCount);
topicName = allTopicNames.get(0);
subscriptionMetadata = AssignorBenchmarkUtils.createSubscriptionMetadata(allTopicNames, partitionsPerTopic);
topicsImage = AssignorBenchmarkUtils.createTopicsImage(subscriptionMetadata);
topicResolver = new TopicIds.CachedTopicResolver(topicsImage);
topicId = topicResolver.id(topicName);
Map<Uuid, TopicMetadata> topicMetadata = AssignorBenchmarkUtils.createTopicMetadata(subscriptionMetadata);
subscribedTopicDescriber = new SubscribedTopicDescriberImpl(topicMetadata);
}
@Setup(Level.Trial)
public void setup() {
setupTopics();
}
@Benchmark
public void testNumOfPartitions() {
subscribedTopicDescriber.numPartitions(topicId);
}
}
```
</details>
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19611:
URL: https://github.com/apache/kafka/pull/19611#issuecomment-2901854903
@AndrewJSchofield the error is caused by #19380 which changed the partition order in request. I will file a PR to fix 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2095270860
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +81,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
Review Comment:
In KIP-881, we don't check whether replicas are online or offline. We basically assume that they will eventually come back. I would follow the same approach here.
Side note, this needs to be aligned with the computation of the hash. There, we include all replicas regardless of whether they are online or not.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2095283685
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +81,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
Review Comment:
If the offline replicas could be existent in this path, we need to add null check for `broker`, right?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2095261511
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +81,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
Review Comment:
I was assuming that we should not. If I recall correctly, this is what we did in KIP-881. I need to double check to be sure.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094571756
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +80,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
+ }
+ return racks;
+ }
+ }
return Set.of();
}
/**
- * Returns a set of assignable partitions from the topic metadata.
- * If the allowed partition map is null, all the partitions in the corresponding
- * topic metadata are returned for the argument topic id. If allowed map is empty,
+ * Returns a set of assignable partitions from the metadata image.
+ * If the allowed partition map is Optional.empty(), all the partitions in the corresponding
+ * topic image are returned for the argument topic id. If allowed map is empty,
* empty set is returned.
*
* @param topicId The uuid of the topic
* @return Set of integers if assignable partitions available, empty otherwise.
*/
@Override
public Set<Integer> assignablePartitions(Uuid topicId) {
- TopicMetadata topic = this.topicMetadata.get(topicId);
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
if (topic == null) {
return Set.of();
}
- if (topicPartitionAllowedMap == null) {
- return IntStream.range(0, topic.numPartitions()).boxed().collect(Collectors.toUnmodifiableSet());
+ if (topicPartitionAllowedMap.isEmpty()) {
+ return Set.copyOf(topic.partitions().keySet());
Review Comment:
As @dajac mentioned, maybe we can make `TopicImage` create immutable collection as it is never modified once built.
```java
public TopicImage(String name,
Uuid id,
Map<Integer, PartitionRegistration> partitions) {
this.name = name;
this.id = id;
this.partitions = Collections.unmodifiableMap(partitions);
}
```
with this change, we can return `topic.partitions().keySet()` directly as it is immutable now.
BTW, I have opened https://issues.apache.org/jira/browse/KAFKA-19305 to fix other image classes too
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094991495
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -112,7 +112,7 @@ public Set<Integer> assignablePartitions(Uuid topicId) {
}
if (topicPartitionAllowedMap.isEmpty()) {
- return Set.copyOf(topic.partitions().keySet());
+ return topic.partitions().keySet();
Review Comment:
Updated it. Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2095258037
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -19,41 +19,42 @@
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 map of topic Ids to the set of allowed partitions for each topic.
+ * If this is empty, all partitions are allowed.
*/
- private final Map<Uuid, TopicMetadata> topicMetadata;
- private final Map<Uuid, Set<Integer>> topicPartitionAllowedMap;
+ private final Optional<Map<Uuid, Set<Integer>>> topicPartitionAllowedMap;
Review Comment:
Using an Optional is better in my opinion.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2093435595
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -19,41 +19,42 @@
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 map of topic Ids to the set of allowed partitions for each topic.
+ * If this is empty, all partitions are allowed.
*/
- private final Map<Uuid, TopicMetadata> topicMetadata;
- private final Map<Uuid, Set<Integer>> topicPartitionAllowedMap;
+ private final Optional<Map<Uuid, Set<Integer>>> topicPartitionAllowedMap;
Review Comment:
There is a related PR in this area https://github.com/apache/kafka/pull/19739. It's correcting a mistake.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094404832
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +80,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
+ }
+ return racks;
+ }
+ }
return Set.of();
}
/**
- * Returns a set of assignable partitions from the topic metadata.
- * If the allowed partition map is null, all the partitions in the corresponding
- * topic metadata are returned for the argument topic id. If allowed map is empty,
+ * Returns a set of assignable partitions from the metadata image.
+ * If the allowed partition map is Optional.empty(), all the partitions in the corresponding
+ * topic image are returned for the argument topic id. If allowed map is empty,
* empty set is returned.
*
* @param topicId The uuid of the topic
* @return Set of integers if assignable partitions available, empty otherwise.
*/
@Override
public Set<Integer> assignablePartitions(Uuid topicId) {
- TopicMetadata topic = this.topicMetadata.get(topicId);
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
if (topic == null) {
return Set.of();
}
- if (topicPartitionAllowedMap == null) {
- return IntStream.range(0, topic.numPartitions()).boxed().collect(Collectors.toUnmodifiableSet());
+ if (topicPartitionAllowedMap.isEmpty()) {
+ return Set.copyOf(topic.partitions().keySet());
Review Comment:
Yeah, this is correct. However, the TopicImage is never mutated once built. The metadata image is immutable.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2095033588
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -112,7 +112,7 @@ public Set<Integer> assignablePartitions(Uuid topicId) {
}
if (topicPartitionAllowedMap.isEmpty()) {
- return Set.copyOf(topic.partitions().keySet());
+ return topic.partitions().keySet();
Review Comment:
@dajac is right, we can remove the wrap in addressing KAFKA-19305
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094377147
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +80,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
+ }
+ return racks;
+ }
+ }
return Set.of();
}
/**
- * Returns a set of assignable partitions from the topic metadata.
- * If the allowed partition map is null, all the partitions in the corresponding
- * topic metadata are returned for the argument topic id. If allowed map is empty,
+ * Returns a set of assignable partitions from the metadata image.
+ * If the allowed partition map is Optional.empty(), all the partitions in the corresponding
+ * topic image are returned for the argument topic id. If allowed map is empty,
* empty set is returned.
*
* @param topicId The uuid of the topic
* @return Set of integers if assignable partitions available, empty otherwise.
*/
@Override
public Set<Integer> assignablePartitions(Uuid topicId) {
- TopicMetadata topic = this.topicMetadata.get(topicId);
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
if (topic == null) {
return Set.of();
}
- if (topicPartitionAllowedMap == null) {
- return IntStream.range(0, topic.numPartitions()).boxed().collect(Collectors.toUnmodifiableSet());
+ if (topicPartitionAllowedMap.isEmpty()) {
+ return Set.copyOf(topic.partitions().keySet());
Review Comment:
IIUC, the production code uses `HashMap` to build partitions in `TopicImage`. From [Map#keySet Java doc](https://devdocs.io/openjdk~17/java.base/java/util/map#keySet()), the change to the set can be reflected in the map. Is it safer to use `Set.copyOf` here?
https://github.com/apache/kafka/blob/f26974b16db7903bd350f06b5796acac52604d4c/metadata/src/main/java/org/apache/kafka/image/TopicDelta.java#L142-L157
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094374099
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##########
@@ -245,22 +237,21 @@ public void testFirstAssignmentNumMembersGreaterThanTotalNumPartitions() {
@Test
public void testValidityAndBalanceForLargeSampleSet() {
- Map<Uuid, TopicMetadata> topicMetadata = new HashMap<>();
- for (int i = 1; i < 100; i++) {
+ MetadataImageBuilder metadataImageBuilder = new MetadataImageBuilder();
+ Set<Uuid> topicIds = new HashSet<>();
+ IntStream.range(1, 101).forEach(i -> {
Review Comment:
Updated it.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicMetadataTest.java:
##########
@@ -17,94 +17,111 @@
package org.apache.kafka.coordinator.group.modern;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.MetadataImageBuilder;
+import org.apache.kafka.image.MetadataImage;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
+import java.util.stream.IntStream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class SubscribedTopicMetadataTest {
- private Map<Uuid, TopicMetadata> topicMetadataMap;
private SubscribedTopicDescriberImpl subscribedTopicMetadata;
+ private MetadataImage metadataImage;
+ private final int numPartitions = 5;
@BeforeEach
public void setUp() {
- topicMetadataMap = new HashMap<>();
- for (int i = 0; i < 5; i++) {
+ MetadataImageBuilder metadataImageBuilder = new MetadataImageBuilder();
+ IntStream.range(0, 5).forEach(i -> {
Review Comment:
Updated 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2095148638
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +81,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
Review Comment:
Should we check the offline replica?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +81,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
+ }
+ return racks;
Review Comment:
Should we return immutable set?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -19,41 +19,42 @@
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 map of topic Ids to the set of allowed partitions for each topic.
+ * If this is empty, all partitions are allowed.
*/
- private final Map<Uuid, TopicMetadata> topicMetadata;
- private final Map<Uuid, Set<Integer>> topicPartitionAllowedMap;
+ private final Optional<Map<Uuid, Set<Integer>>> topicPartitionAllowedMap;
Review Comment:
#19739 uses null, in contrast, this PR uses `Optional`. @AndrewJSchofield WDYT?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094785891
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +80,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
+ }
+ return racks;
+ }
+ }
return Set.of();
}
/**
- * Returns a set of assignable partitions from the topic metadata.
- * If the allowed partition map is null, all the partitions in the corresponding
- * topic metadata are returned for the argument topic id. If allowed map is empty,
+ * Returns a set of assignable partitions from the metadata image.
+ * If the allowed partition map is Optional.empty(), all the partitions in the corresponding
+ * topic image are returned for the argument topic id. If allowed map is empty,
* empty set is returned.
*
* @param topicId The uuid of the topic
* @return Set of integers if assignable partitions available, empty otherwise.
*/
@Override
public Set<Integer> assignablePartitions(Uuid topicId) {
- TopicMetadata topic = this.topicMetadata.get(topicId);
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
if (topic == null) {
return Set.of();
}
- if (topicPartitionAllowedMap == null) {
- return IntStream.range(0, topic.numPartitions()).boxed().collect(Collectors.toUnmodifiableSet());
+ if (topicPartitionAllowedMap.isEmpty()) {
+ return Set.copyOf(topic.partitions().keySet());
Review Comment:
Thanks for the review and suggestion. Remove `Set.copyOf` here.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2095262127
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +81,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
+ }
+ return racks;
Review Comment:
Good call.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #19611:
URL: https://github.com/apache/kafka/pull/19611#issuecomment-2876345298
@FrankYang0529 Thanks for the results. Have you also tried to run `ServerSideAssignorBenchmark`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2095290200
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +81,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
Review Comment:
We do null check in hash function. It's better to add the check here.
https://github.com/apache/kafka/blob/ce4940f9891a96819e54f8db097ce3824876e8e5/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java#L408-L413
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19611:
URL: https://github.com/apache/kafka/pull/19611#issuecomment-2901102431
@FrankYang0529 I wonder whether this PR could be the cause of the recent flakiness in `PlaintextAdminIntegrationTest.testDeleteConsumerGroupOffsets`. The test checks that the correct error code is returned if an attempt is made to delete a consumer group offset when there is a subscription to the topic whose offsets are being deleted.
https://develocity.apache.org/scans/tests?search.relativeStartTime=P28D&search.rootProjectNames=kafka&search.timeZoneId=Europe%2FLondon&tests.container=kafka.api.PlaintextAdminIntegrationTest&tests.test=testDeleteConsumerGroupOffsets(String)%5B1%5D
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [3/N] Get rid of TopicMetadata in SubscribedTopicDescriberImpl [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19611:
URL: https://github.com/apache/kafka/pull/19611#discussion_r2094164148
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##########
@@ -61,15 +63,12 @@ public class OptimizedUniformAssignmentBuilderTest {
@Test
public void testOneMemberNoTopicSubscription() {
+ MetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(topic1Uuid, topic1Name, 3)
+ .addRacks()
Review Comment:
nit: Is `addRacks` necessary here and in all the other places?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -79,49 +80,65 @@ public int numPartitions(Uuid topicId) {
*/
@Override
public Set<String> racksForPartition(Uuid topicId, int partition) {
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
+ if (topic != null) {
+ PartitionRegistration partitionRegistration = topic.partitions().get(partition);
+ if (partitionRegistration != null) {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ // Only add the rack if it is available for the broker/replica.
+ metadataImage.cluster().broker(replica).rack().ifPresent(racks::add);
+ }
+ return racks;
+ }
+ }
return Set.of();
}
/**
- * Returns a set of assignable partitions from the topic metadata.
- * If the allowed partition map is null, all the partitions in the corresponding
- * topic metadata are returned for the argument topic id. If allowed map is empty,
+ * Returns a set of assignable partitions from the metadata image.
+ * If the allowed partition map is Optional.empty(), all the partitions in the corresponding
+ * topic image are returned for the argument topic id. If allowed map is empty,
* empty set is returned.
*
* @param topicId The uuid of the topic
* @return Set of integers if assignable partitions available, empty otherwise.
*/
@Override
public Set<Integer> assignablePartitions(Uuid topicId) {
- TopicMetadata topic = this.topicMetadata.get(topicId);
+ TopicImage topic = metadataImage.topics().getTopic(topicId);
if (topic == null) {
return Set.of();
}
- if (topicPartitionAllowedMap == null) {
- return IntStream.range(0, topic.numPartitions()).boxed().collect(Collectors.toUnmodifiableSet());
+ if (topicPartitionAllowedMap.isEmpty()) {
+ return Set.copyOf(topic.partitions().keySet());
Review Comment:
Do we really need to make a copy of the Set? The image is immutable so we could just return a non modifiable version. What do you think?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##########
@@ -245,22 +237,21 @@ public void testFirstAssignmentNumMembersGreaterThanTotalNumPartitions() {
@Test
public void testValidityAndBalanceForLargeSampleSet() {
- Map<Uuid, TopicMetadata> topicMetadata = new HashMap<>();
- for (int i = 1; i < 100; i++) {
+ MetadataImageBuilder metadataImageBuilder = new MetadataImageBuilder();
+ Set<Uuid> topicIds = new HashSet<>();
+ IntStream.range(1, 101).forEach(i -> {
Review Comment:
Is 1 to 101 correct? It may be easier to keep the for loop.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java:
##########
@@ -19,41 +19,42 @@
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 map of topic Ids to the set of allowed partitions for each topic.
+ * If this is empty, all partitions are allowed.
*/
- private final Map<Uuid, TopicMetadata> topicMetadata;
- private final Map<Uuid, Set<Integer>> topicPartitionAllowedMap;
+ private final Optional<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(Map<Uuid, TopicMetadata> topicMetadata, Map<Uuid, Set<Integer>> topicPartitionAllowedMap) {
- this.topicMetadata = Objects.requireNonNull(topicMetadata);
- this.topicPartitionAllowedMap = topicPartitionAllowedMap;
+ public SubscribedTopicDescriberImpl(MetadataImage metadataImage) {
+ this(metadataImage, Optional.empty());
}
- /**
- * Map of topic Ids to topic metadata.
- *
- * @return The map of topic Ids to topic metadata.
- */
- public Map<Uuid, TopicMetadata> topicMetadata() {
- return this.topicMetadata;
+ public SubscribedTopicDescriberImpl(
+ MetadataImage metadataImage,
+ Optional<Map<Uuid, Set<Integer>>> topicPartitionAllowedMap
+ ) {
+ this.metadataImage = Objects.requireNonNull(metadataImage);
+ this.topicPartitionAllowedMap = topicPartitionAllowedMap;
Review Comment:
nit: Should we requireNonNull too?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicMetadataTest.java:
##########
@@ -17,94 +17,111 @@
package org.apache.kafka.coordinator.group.modern;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.MetadataImageBuilder;
+import org.apache.kafka.image.MetadataImage;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
+import java.util.stream.IntStream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class SubscribedTopicMetadataTest {
- private Map<Uuid, TopicMetadata> topicMetadataMap;
private SubscribedTopicDescriberImpl subscribedTopicMetadata;
+ private MetadataImage metadataImage;
+ private final int numPartitions = 5;
@BeforeEach
public void setUp() {
- topicMetadataMap = new HashMap<>();
- for (int i = 0; i < 5; i++) {
+ MetadataImageBuilder metadataImageBuilder = new MetadataImageBuilder();
+ IntStream.range(0, 5).forEach(i -> {
Review Comment:
nit: Let's keep the for loop here as well.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org