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/02/04 15:14:30 UTC
[PR] KAFKA-17565: Move MetadataCache interface to metadata module (wip) [kafka]
FrankYang0529 opened a new pull request, #18801:
URL: https://github.com/apache/kafka/pull/18801
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-17565: Move MetadataCache interface to metadata module (wip) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2634364064
I realize your PR is still wip, so feel free to ignore the two comments if you were still working through that code. :)
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1997636633
##########
metadata/src/main/java/org/apache/kafka/metadata/MetadataCache.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.admin.BrokerMetadata;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
+import org.apache.kafka.common.message.DescribeClientQuotasResponseData;
+import org.apache.kafka.common.message.DescribeTopicPartitionsResponseData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.server.common.FinalizedFeatures;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public interface MetadataCache extends ConfigRepository {
+
+ /**
+ * Return topic metadata for a given set of topics and listener. See KafkaApis#handleTopicMetadataRequest for details
+ * on the use of the two boolean flags.
+ *
+ * @param topics The set of topics.
+ * @param listenerName The listener name.
+ * @param errorUnavailableEndpoints If true, we return an error on unavailable brokers. This is used to support
+ * MetadataResponse version 0.
+ * @param errorUnavailableListeners If true, return LEADER_NOT_AVAILABLE if the listener is not found on the leader.
+ * This is used for MetadataResponse versions 0-5.
+ * @return A collection of topic metadata.
+ */
+ List<MetadataResponseData.MetadataResponseTopic> getTopicMetadata(
+ Set<String> topics,
+ ListenerName listenerName,
+ boolean errorUnavailableEndpoints,
+ boolean errorUnavailableListeners);
+
+ Set<String> getAllTopics();
+
+ boolean hasAliveBroker(int brokerId);
+
+ Optional<Long> getAliveBrokerEpoch(int brokerId);
+
+ boolean isBrokerFenced(int brokerId);
+
+ boolean isBrokerShuttingDown(int brokerId);
+
+ Uuid getTopicId(String topicName);
+
+ Optional<String> getTopicName(Uuid topicId);
+
+ Optional<Node> getAliveBrokerNode(int brokerId, ListenerName listenerName);
+
+ List<Node> getAliveBrokerNodes(ListenerName listenerName);
+
+ List<Node> getBrokerNodes(ListenerName listenerName);
+
+ Optional<LeaderAndIsr> getLeaderAndIsr(String topic, int partitionId);
+
+ /**
+ * Return the number of partitions in the given topic, or None if the given topic does not exist.
+ */
+ Optional<Integer> numPartitions(String topic);
+
+ Map<Uuid, String> topicIdsToNames();
+
+ /**
+ * Get a partition leader's endpoint
+ *
+ * @return If the leader is known, and the listener name is available, return Some(node). If the leader is known,
+ * but the listener is unavailable, return Some(Node.NO_NODE). Otherwise, if the leader is not known,
+ * return None
+ */
+ Optional<Node> getPartitionLeaderEndpoint(String topic, int partitionId, ListenerName listenerName);
+
+ Map<Integer, Node> getPartitionReplicaEndpoints(TopicPartition tp, ListenerName listenerName);
+
+ boolean contains(String topic);
+
+ boolean contains(TopicPartition tp);
+
+ MetadataVersion metadataVersion();
+
+ Optional<Integer> getRandomAliveBrokerId();
+
+ FinalizedFeatures features();
+
+ DescribeClientQuotasResponseData describeClientQuotas(DescribeClientQuotasRequestData request);
+
+ DescribeUserScramCredentialsResponseData describeScramCredentials(DescribeUserScramCredentialsRequestData request);
+
+ /**
+ * Get the topic metadata for the given topics.
+ *
+ * The quota is used to limit the number of partitions to return. The NextTopicPartition field points to the first
+ * partition can't be returned due the limit.
+ * If a topic can't return any partition due to quota limit reached, this topic will not be included in the response.
+ *
+ * Note, the topics should be sorted in alphabetical order. The topics in the DescribeTopicPartitionsResponseData
+ * will also be sorted in alphabetical order.
+ *
+ * @param topics The iterator of topics and their corresponding first partition id to fetch.
+ * @param listenerName The listener name.
+ * @param topicPartitionStartIndex The start partition index for the first topic
+ * @param maximumNumberOfPartitions The max number of partitions to return.
+ * @param ignoreTopicsWithExceptions Whether ignore the topics with exception.
+ */
+ DescribeTopicPartitionsResponseData describeTopicResponse(
+ Iterator<String> topics,
+ ListenerName listenerName,
+ Function<String, Integer> topicPartitionStartIndex,
+ int maximumNumberOfPartitions,
+ boolean ignoreTopicsWithExceptions);
+
+ static Cluster toCluster(String clusterId, MetadataImage image) {
+ Map<Integer, List<Node>> brokerToNodes = new HashMap<>();
+ image.cluster().brokers().values().stream()
+ .filter(broker -> !broker.fenced())
+ .forEach(broker -> brokerToNodes.put(broker.id(), broker.nodes()));
+
+ List<PartitionInfo> partitionInfos = new ArrayList<>();
+ Set<String> internalTopics = new HashSet<>();
+
+ image.topics().topicsByName().values().forEach(topic -> {
+ topic.partitions().forEach((partitionId, partition) -> {
+ List<Node> nodes = brokerToNodes.get(partition.leader);
+ if (nodes != null) {
+ nodes.forEach(node -> {
+ partitionInfos.add(new PartitionInfo(
+ topic.name(),
+ partitionId,
+ node,
+ toArray(partition.replicas, brokerToNodes),
+ toArray(partition.isr, brokerToNodes),
+ getOfflineReplicas(image, partition).stream()
+ .map(brokerToNodes::get)
+ .flatMap(Collection::stream)
+ .toArray(Node[]::new)
+ ));
+ });
+ if (Topic.isInternal(topic.name())) {
+ internalTopics.add(topic.name());
+ }
+ }
+ });
+ });
+
+ Node controllerNode = Optional.ofNullable(brokerToNodes.get(getRandomAliveBroker(image).orElse(-1)))
+ .map(nodes -> nodes.get(0))
+ .orElse(Node.noNode());
+
+ return new Cluster(
+ clusterId,
+ brokerToNodes.values().stream().flatMap(Collection::stream).collect(Collectors.toList()),
+ partitionInfos,
+ Collections.emptySet(),
+ internalTopics,
+ controllerNode
+ );
+ }
+
+ private static Node[] toArray(int[] replicas, Map<Integer, List<Node>> brokerToNodes) {
+ return Arrays.stream(replicas)
+ .mapToObj(brokerToNodes::get)
+ .flatMap(Collection::stream)
+ .toArray(Node[]::new);
+ }
+
+ private static List<Integer> getOfflineReplicas(MetadataImage image, PartitionRegistration partition) {
+ List<Integer> offlineReplicas = new ArrayList<>();
+ for (int brokerId : partition.replicas) {
+ BrokerRegistration broker = image.cluster().broker(brokerId);
+ if (broker == null || isReplicaOffline(partition, broker)) {
+ offlineReplicas.add(brokerId);
+ }
+ }
+ return offlineReplicas;
+ }
+
+ private static boolean isReplicaOffline(PartitionRegistration partition, BrokerRegistration broker) {
+ return broker.fenced() || !broker.hasOnlineDir(partition.directory(broker.id()));
+ }
+
+ private static Optional<Integer> getRandomAliveBroker(MetadataImage image) {
+ List<BrokerMetadata> aliveBrokers = getAliveBrokers(image);
+ if (aliveBrokers.isEmpty()) {
+ return Optional.empty();
+ } else {
+ return Optional.of(aliveBrokers.get(ThreadLocalRandom.current().nextInt(aliveBrokers.size())).id);
+ }
+ }
+
+ private static List<BrokerMetadata> getAliveBrokers(MetadataImage image) {
Review Comment:
Do we really need to use `BrokerMetadata`? It seems we can remove `getAliveBrokers` and then simplify `getRandomAliveBroker`.
```java
private static Optional<Integer> getRandomAliveBroker(MetadataImage image) {
var aliveBrokers = image.cluster().brokers().values().stream()
.filter(broker -> !broker.fenced())
.map(BrokerRegistration::id)
.toList();
if (aliveBrokers.isEmpty()) return Optional.empty();
return Optional.of(aliveBrokers.get(ThreadLocalRandom.current().nextInt(aliveBrokers.size())));
}
```
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2716624266
Hi @chia7712, I addressed all comments. Could you review this when you have time? 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-17565: Move MetadataCache interface to metadata module (wip) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1941434808
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -345,68 +344,69 @@ class KRaftMetadataCache(
Option(_currentImage.cluster.broker(brokerId)).count(_.inControlledShutdown) == 1
}
- override def getAliveBrokers(): Iterable[BrokerMetadata] = getAliveBrokers(_currentImage)
+ override def getAliveBrokers(): util.List[BrokerMetadata] = getAliveBrokers(_currentImage)
- private def getAliveBrokers(image: MetadataImage): Iterable[BrokerMetadata] = {
+ private def getAliveBrokers(image: MetadataImage): util.List[BrokerMetadata] = {
image.cluster().brokers().values().asScala.filterNot(_.fenced()).
map(b => new BrokerMetadata(b.id, b.rack))
- }
+ }.toList.asJava
- override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): Option[Node] = {
- Option(_currentImage.cluster().broker(brokerId)).filterNot(_.fenced()).
- flatMap(_.node(listenerName.value()).toScala)
+ override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): util.Optional[Node] = {
+ util.Optional.ofNullable(_currentImage.cluster().broker(brokerId))
+ .filter(broker => !broker.fenced())
+ .flatMap(broker => broker.node(listenerName.value()))
}
- override def getAliveBrokerNodes(listenerName: ListenerName): Seq[Node] = {
+ override def getAliveBrokerNodes(listenerName: ListenerName): util.List[Node] = {
_currentImage.cluster().brokers().values().asScala.filterNot(_.fenced()).
Review Comment:
Thanks for the suggestion. I will update it tomorrow.
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2680667964
Hi @mumrah, thanks for the review. I addressed all comments. For benchmark, I think we can do it in https://issues.apache.org/jira/browse/KAFKA-17573 when we migrate `KRaftMetadataCache`. The `MetadataCache` in this PR is more about interface, not implementation.
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2685571786
@mumrah I update benchmark to PR description. The result is a little different from yours. The difference between trunk and branch results on my laptop is close, so I think the result is acceptable.
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2727481739
@chia7712 I update PR description. 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-17565: Move MetadataCache interface to metadata module (wip) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1941423335
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -345,68 +344,69 @@ class KRaftMetadataCache(
Option(_currentImage.cluster.broker(brokerId)).count(_.inControlledShutdown) == 1
}
- override def getAliveBrokers(): Iterable[BrokerMetadata] = getAliveBrokers(_currentImage)
+ override def getAliveBrokers(): util.List[BrokerMetadata] = getAliveBrokers(_currentImage)
- private def getAliveBrokers(image: MetadataImage): Iterable[BrokerMetadata] = {
+ private def getAliveBrokers(image: MetadataImage): util.List[BrokerMetadata] = {
image.cluster().brokers().values().asScala.filterNot(_.fenced()).
map(b => new BrokerMetadata(b.id, b.rack))
- }
+ }.toList.asJava
- override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): Option[Node] = {
- Option(_currentImage.cluster().broker(brokerId)).filterNot(_.fenced()).
- flatMap(_.node(listenerName.value()).toScala)
+ override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): util.Optional[Node] = {
+ util.Optional.ofNullable(_currentImage.cluster().broker(brokerId))
+ .filter(broker => !broker.fenced())
+ .flatMap(broker => broker.node(listenerName.value()))
}
- override def getAliveBrokerNodes(listenerName: ListenerName): Seq[Node] = {
+ override def getAliveBrokerNodes(listenerName: ListenerName): util.List[Node] = {
_currentImage.cluster().brokers().values().asScala.filterNot(_.fenced()).
- flatMap(_.node(listenerName.value()).toScala).toSeq
+ flatMap(_.node(listenerName.value()).toScala).toList.asJava
Review Comment:
Note that `toList` converts to an immutable linked list - you do _not_ want to do that if you are then calling `asJava`. `toBuffer` is probably a better option.
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1997644083
##########
metadata/src/main/java/org/apache/kafka/metadata/MetadataCache.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.admin.BrokerMetadata;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
+import org.apache.kafka.common.message.DescribeClientQuotasResponseData;
+import org.apache.kafka.common.message.DescribeTopicPartitionsResponseData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.server.common.FinalizedFeatures;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public interface MetadataCache extends ConfigRepository {
+
+ /**
+ * Return topic metadata for a given set of topics and listener. See KafkaApis#handleTopicMetadataRequest for details
+ * on the use of the two boolean flags.
+ *
+ * @param topics The set of topics.
+ * @param listenerName The listener name.
+ * @param errorUnavailableEndpoints If true, we return an error on unavailable brokers. This is used to support
+ * MetadataResponse version 0.
+ * @param errorUnavailableListeners If true, return LEADER_NOT_AVAILABLE if the listener is not found on the leader.
+ * This is used for MetadataResponse versions 0-5.
+ * @return A collection of topic metadata.
+ */
+ List<MetadataResponseData.MetadataResponseTopic> getTopicMetadata(
+ Set<String> topics,
+ ListenerName listenerName,
+ boolean errorUnavailableEndpoints,
+ boolean errorUnavailableListeners);
+
+ Set<String> getAllTopics();
+
+ boolean hasAliveBroker(int brokerId);
+
+ Optional<Long> getAliveBrokerEpoch(int brokerId);
+
+ boolean isBrokerFenced(int brokerId);
+
+ boolean isBrokerShuttingDown(int brokerId);
+
+ Uuid getTopicId(String topicName);
+
+ Optional<String> getTopicName(Uuid topicId);
+
+ Optional<Node> getAliveBrokerNode(int brokerId, ListenerName listenerName);
+
+ List<Node> getAliveBrokerNodes(ListenerName listenerName);
+
+ List<Node> getBrokerNodes(ListenerName listenerName);
+
+ Optional<LeaderAndIsr> getLeaderAndIsr(String topic, int partitionId);
+
+ /**
+ * Return the number of partitions in the given topic, or None if the given topic does not exist.
+ */
+ Optional<Integer> numPartitions(String topic);
+
+ Map<Uuid, String> topicIdsToNames();
+
+ /**
+ * Get a partition leader's endpoint
+ *
+ * @return If the leader is known, and the listener name is available, return Some(node). If the leader is known,
+ * but the listener is unavailable, return Some(Node.NO_NODE). Otherwise, if the leader is not known,
+ * return None
+ */
+ Optional<Node> getPartitionLeaderEndpoint(String topic, int partitionId, ListenerName listenerName);
+
+ Map<Integer, Node> getPartitionReplicaEndpoints(TopicPartition tp, ListenerName listenerName);
+
+ boolean contains(String topic);
+
+ boolean contains(TopicPartition tp);
+
+ MetadataVersion metadataVersion();
+
+ Optional<Integer> getRandomAliveBrokerId();
+
+ FinalizedFeatures features();
+
+ DescribeClientQuotasResponseData describeClientQuotas(DescribeClientQuotasRequestData request);
+
+ DescribeUserScramCredentialsResponseData describeScramCredentials(DescribeUserScramCredentialsRequestData request);
+
+ /**
+ * Get the topic metadata for the given topics.
+ *
+ * The quota is used to limit the number of partitions to return. The NextTopicPartition field points to the first
+ * partition can't be returned due the limit.
+ * If a topic can't return any partition due to quota limit reached, this topic will not be included in the response.
+ *
+ * Note, the topics should be sorted in alphabetical order. The topics in the DescribeTopicPartitionsResponseData
+ * will also be sorted in alphabetical order.
+ *
+ * @param topics The iterator of topics and their corresponding first partition id to fetch.
+ * @param listenerName The listener name.
+ * @param topicPartitionStartIndex The start partition index for the first topic
+ * @param maximumNumberOfPartitions The max number of partitions to return.
+ * @param ignoreTopicsWithExceptions Whether ignore the topics with exception.
+ */
+ DescribeTopicPartitionsResponseData describeTopicResponse(
+ Iterator<String> topics,
+ ListenerName listenerName,
+ Function<String, Integer> topicPartitionStartIndex,
+ int maximumNumberOfPartitions,
+ boolean ignoreTopicsWithExceptions);
+
+ static Cluster toCluster(String clusterId, MetadataImage image) {
+ Map<Integer, List<Node>> brokerToNodes = new HashMap<>();
+ image.cluster().brokers().values().stream()
+ .filter(broker -> !broker.fenced())
+ .forEach(broker -> brokerToNodes.put(broker.id(), broker.nodes()));
+
+ List<PartitionInfo> partitionInfos = new ArrayList<>();
+ Set<String> internalTopics = new HashSet<>();
+
+ image.topics().topicsByName().values().forEach(topic -> {
+ topic.partitions().forEach((partitionId, partition) -> {
+ List<Node> nodes = brokerToNodes.get(partition.leader);
+ if (nodes != null) {
+ nodes.forEach(node -> {
+ partitionInfos.add(new PartitionInfo(
+ topic.name(),
+ partitionId,
+ node,
+ toArray(partition.replicas, brokerToNodes),
+ toArray(partition.isr, brokerToNodes),
+ getOfflineReplicas(image, partition).stream()
+ .map(brokerToNodes::get)
+ .flatMap(Collection::stream)
+ .toArray(Node[]::new)
+ ));
+ });
+ if (Topic.isInternal(topic.name())) {
+ internalTopics.add(topic.name());
+ }
+ }
+ });
+ });
+
+ Node controllerNode = Optional.ofNullable(brokerToNodes.get(getRandomAliveBroker(image).orElse(-1)))
+ .map(nodes -> nodes.get(0))
+ .orElse(Node.noNode());
+
+ return new Cluster(
+ clusterId,
+ brokerToNodes.values().stream().flatMap(Collection::stream).collect(Collectors.toList()),
+ partitionInfos,
+ Collections.emptySet(),
+ internalTopics,
+ controllerNode
+ );
+ }
+
+ private static Node[] toArray(int[] replicas, Map<Integer, List<Node>> brokerToNodes) {
+ return Arrays.stream(replicas)
+ .mapToObj(brokerToNodes::get)
+ .flatMap(Collection::stream)
+ .toArray(Node[]::new);
+ }
+
+ private static List<Integer> getOfflineReplicas(MetadataImage image, PartitionRegistration partition) {
+ List<Integer> offlineReplicas = new ArrayList<>();
+ for (int brokerId : partition.replicas) {
+ BrokerRegistration broker = image.cluster().broker(brokerId);
+ if (broker == null || isReplicaOffline(partition, broker)) {
+ offlineReplicas.add(brokerId);
+ }
+ }
+ return offlineReplicas;
+ }
+
+ private static boolean isReplicaOffline(PartitionRegistration partition, BrokerRegistration broker) {
+ return broker.fenced() || !broker.hasOnlineDir(partition.directory(broker.id()));
+ }
+
+ private static Optional<Integer> getRandomAliveBroker(MetadataImage image) {
+ List<BrokerMetadata> aliveBrokers = getAliveBrokers(image);
+ if (aliveBrokers.isEmpty()) {
+ return Optional.empty();
+ } else {
+ return Optional.of(aliveBrokers.get(ThreadLocalRandom.current().nextInt(aliveBrokers.size())).id);
+ }
+ }
+
+ private static List<BrokerMetadata> getAliveBrokers(MetadataImage image) {
Review Comment:
Update 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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1965712115
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -821,17 +821,19 @@ object TestUtils extends Logging {
waitUntilTrue(
() => brokers.forall { broker =>
if (expectedNumPartitions == 0) {
- broker.metadataCache.numPartitions(topic) == None
+ broker.metadataCache.numPartitions(topic).isEmpty
} else {
- broker.metadataCache.numPartitions(topic) == Some(expectedNumPartitions)
+ broker.metadataCache.numPartitions(topic).isPresent && broker.metadataCache.numPartitions(topic).get == expectedNumPartitions
Review Comment:
I think this can be simplified to use `filter` or maybe `orElse(null) = `
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -2383,7 +2385,7 @@ class KafkaApis(val requestChannel: RequestChannel,
() => {
val brokers = new DescribeClusterResponseData.DescribeClusterBrokerCollection()
val describeClusterRequest = request.body[DescribeClusterRequest]
- metadataCache.getBrokerNodes(request.context.listenerName).foreach { node =>
+ metadataCache.getBrokerNodes(request.context.listenerName).asScala.foreach { node =>
Review Comment:
Do we need to convert to Scala for the `foreach` here? Could we use the Java stream and its `forEach` instead?
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -320,18 +320,23 @@ class KRaftMetadataCache(
result
}
- override def getAllTopics(): Set[String] = _currentImage.topics().topicsByName().keySet().asScala
+ override def getAllTopics(): util.Set[String] = _currentImage.topics().topicsByName().keySet()
- override def getTopicPartitions(topicName: String): Set[TopicPartition] = {
- Option(_currentImage.topics().getTopic(topicName)) match {
- case None => Set.empty
- case Some(topic) => topic.partitions().keySet().asScala.map(new TopicPartition(topicName, _))
+ override def getTopicPartitions(topicName: String): util.Set[TopicPartition] = {
+ val topic = _currentImage.topics().getTopic(topicName)
+ if (topic == null) {
+ return util.Set.of;
}
+ topic.partitions.keySet.stream
+ .map(partitionId => new TopicPartition(topicName, partitionId))
+ .collect(Collectors.toSet());
}
- override def getTopicId(topicName: String): Uuid = _currentImage.topics().topicsByName().asScala.get(topicName).map(_.id()).getOrElse(Uuid.ZERO_UUID)
+ override def getTopicId(topicName: String): Uuid = util.Optional.ofNullable(_currentImage.topics.topicsByName.get(topicName))
+ .map(topic => topic.id)
Review Comment:
We can use the method reference style here: `.map(_.id)`
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1193,8 +1195,8 @@ class KafkaApis(val requestChannel: RequestChannel,
val coordinatorEndpoint = topicMetadata.head.partitions.asScala
.find(_.partitionIndex == partition)
.filter(_.leaderId != MetadataResponse.NO_LEADER_ID)
- .flatMap(metadata => metadataCache.
- getAliveBrokerNode(metadata.leaderId, request.context.listenerName))
+ .flatMap(metadata => OptionConverters.toScala(metadataCache.
Review Comment:
Can we let `topicMetadata` stay as as a Java collection and convert this whole stream to a Java stream?
##########
core/src/main/scala/kafka/server/DelayedElectLeader.scala:
##########
@@ -74,7 +75,7 @@ class DelayedElectLeader(
private def updateWaiting(): Unit = {
val metadataCache = replicaManager.metadataCache
val completedPartitions = waitingPartitions.collect {
- case (tp, leader) if metadataCache.getLeaderAndIsr(tp.topic, tp.partition).exists(_.leader == leader) => tp
+ case (tp, leader) if OptionConverters.toScala(metadataCache.getLeaderAndIsr(tp.topic, tp.partition)).exists(_.leader == leader) => tp
Review Comment:
The `exists` can be done in Java with `filter` + `isPresent`
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -345,68 +350,75 @@ class KRaftMetadataCache(
Option(_currentImage.cluster.broker(brokerId)).count(_.inControlledShutdown) == 1
}
- override def getAliveBrokers(): Iterable[BrokerMetadata] = getAliveBrokers(_currentImage)
+ override def getAliveBrokers(): util.List[BrokerMetadata] = getAliveBrokers(_currentImage)
- private def getAliveBrokers(image: MetadataImage): Iterable[BrokerMetadata] = {
- image.cluster().brokers().values().asScala.filterNot(_.fenced()).
- map(b => new BrokerMetadata(b.id, b.rack))
+ private def getAliveBrokers(image: MetadataImage): util.List[BrokerMetadata] = {
+ _currentImage.cluster.brokers.values.stream
+ .filter(broker => !broker.fenced)
+ .map(broker => new BrokerMetadata(broker.id, broker.rack))
+ .collect(Collectors.toList())
}
- override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): Option[Node] = {
- Option(_currentImage.cluster().broker(brokerId)).filterNot(_.fenced()).
- flatMap(_.node(listenerName.value()).toScala)
+ override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): util.Optional[Node] = {
+ util.Optional.ofNullable(_currentImage.cluster.broker(brokerId))
+ .filter(broker => !broker.fenced)
Review Comment:
With Java 11 we can now use `Predicate.not` for negating filters with lambda. This can be
```
.filter(Predicate.not(_.fenced))
```
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -239,12 +239,12 @@ class KRaftMetadataCache(
}
// errorUnavailableEndpoints exists to support v0 MetadataResponses
- override def getTopicMetadata(topics: Set[String],
+ override def getTopicMetadata(topics: util.Set[String],
listenerName: ListenerName,
errorUnavailableEndpoints: Boolean = false,
- errorUnavailableListeners: Boolean = false): Seq[MetadataResponseTopic] = {
+ errorUnavailableListeners: Boolean = false): util.List[MetadataResponseTopic] = {
val image = _currentImage
- topics.toSeq.flatMap { topic =>
+ topics.asScala.toSeq.flatMap { topic =>
Review Comment:
Can we avoid this conversion to Scala and back to Java collection? Since `topics` is now a Java set, it seems like we can use a Java stream here.
##########
core/src/main/scala/kafka/cluster/Replica.scala:
##########
@@ -113,7 +113,7 @@ class Replica(val brokerId: Int, val topicPartition: TopicPartition, val metadat
replicaState.updateAndGet { currentReplicaState =>
val cachedBrokerEpoch = metadataCache.getAliveBrokerEpoch(brokerId)
// Fence the update if it provides a stale broker epoch.
- if (brokerEpoch != -1 && cachedBrokerEpoch.exists(_ > brokerEpoch)) {
+ if (brokerEpoch != -1 && cachedBrokerEpoch.isPresent && cachedBrokerEpoch.get > brokerEpoch) {
Review Comment:
Simplify to `filter` + `isPresent`
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -186,9 +188,9 @@ class AddPartitionsToTxnManager(
}
private def getTransactionCoordinator(partition: Int): Option[Node] = {
- metadataCache.getLeaderAndIsr(Topic.TRANSACTION_STATE_TOPIC_NAME, partition)
+ OptionConverters.toScala(metadataCache.getLeaderAndIsr(Topic.TRANSACTION_STATE_TOPIC_NAME, partition))
.filter(_.leader != MetadataResponse.NO_LEADER_ID)
- .flatMap(metadata => metadataCache.getAliveBrokerNode(metadata.leader, interBrokerListenerName))
+ .flatMap(metadata => OptionConverters.toScala(metadataCache.getAliveBrokerNode(metadata.leader, interBrokerListenerName)))
Review Comment:
I think this would be better if we keep the Java Optional and convert it at the end.
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2682316716
I ran one of the benchmarks and there actually seems to be a slight improvement! cc @ijuma
benchmark:
```
./jmh.sh -f 2 -i 2 -wi 2 org.apache.kafka.jmh.server.CheckpointBench.measureCheckpointHighWatermarks
```
trunk:
```
Benchmark (numPartitions) (numTopics) Mode Cnt Score Error Units
CheckpointBench.measureCheckpointHighWatermarks 3 100 thrpt 4 0.154 ± 0.010 ops/ms
CheckpointBench.measureCheckpointHighWatermarks 3 1000 thrpt 4 0.101 ± 0.032 ops/ms
CheckpointBench.measureCheckpointHighWatermarks 3 2000 thrpt 4 0.091 ± 0.019 ops/ms
JMH benchmarks done
```
this PR:
```
Benchmark (numPartitions) (numTopics) Mode Cnt Score Error Units
CheckpointBench.measureCheckpointHighWatermarks 3 100 thrpt 4 0.153 ± 0.109 ops/ms
CheckpointBench.measureCheckpointHighWatermarks 3 1000 thrpt 4 0.141 ± 0.031 ops/ms
CheckpointBench.measureCheckpointHighWatermarks 3 2000 thrpt 4 0.129 ± 0.033 ops/ms
```
@FrankYang0529 now I definitely think you should run the benchmarks for this PR so you can take credit for the performance improvement 😁
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1969879516
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1192,17 +1192,18 @@ class KafkaApis(val requestChannel: RequestChannel,
if (topicMetadata.head.errorCode != Errors.NONE.code) {
(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode)
} else {
- val coordinatorEndpoint = topicMetadata.head.partitions.asScala
- .find(_.partitionIndex == partition)
+ val coordinatorEndpoint = topicMetadata.head.partitions.stream()
+ .filter(_.partitionIndex() == partition)
.filter(_.leaderId != MetadataResponse.NO_LEADER_ID)
- .flatMap(metadata => OptionConverters.toScala(metadataCache.
- getAliveBrokerNode(metadata.leaderId, request.context.listenerName)))
+ .map(metadata => metadataCache.getAliveBrokerNode(metadata.leaderId, request.context.listenerName))
+ .filter(_.isPresent)
+ .map(_.get)
Review Comment:
I think we could reduce this to a flatMap with Optional#stream
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2727474472
@FrankYang0529 Could you please briefly summarize the changes of this PR? Also, it would be better to list the deleted (unused) APIs
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1978888530
##########
metadata/src/main/java/org/apache/kafka/metadata/MetadataCache.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.admin.BrokerMetadata;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
+import org.apache.kafka.common.message.DescribeClientQuotasResponseData;
+import org.apache.kafka.common.message.DescribeTopicPartitionsResponseData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.server.common.FinalizedFeatures;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public interface MetadataCache extends ConfigRepository {
+
+ /**
+ * Return topic metadata for a given set of topics and listener. See KafkaApis#handleTopicMetadataRequest for details
+ * on the use of the two boolean flags.
+ *
+ * @param topics The set of topics.
+ * @param listenerName The listener name.
+ * @param errorUnavailableEndpoints If true, we return an error on unavailable brokers. This is used to support
+ * MetadataResponse version 0.
+ * @param errorUnavailableListeners If true, return LEADER_NOT_AVAILABLE if the listener is not found on the leader.
+ * This is used for MetadataResponse versions 0-5.
+ * @return A collection of topic metadata.
+ */
+ List<MetadataResponseData.MetadataResponseTopic> getTopicMetadata(
+ Set<String> topics,
+ ListenerName listenerName,
+ boolean errorUnavailableEndpoints,
+ boolean errorUnavailableListeners);
+
+ Set<String> getAllTopics();
+
+ Set<TopicPartition> getTopicPartitions(String topicName);
Review Comment:
it seems the usage of this method is to get the number of partitions. could you please rewrite the tests to use `numPartitions` instead?
##########
metadata/src/main/java/org/apache/kafka/metadata/MetadataCache.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.admin.BrokerMetadata;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
+import org.apache.kafka.common.message.DescribeClientQuotasResponseData;
+import org.apache.kafka.common.message.DescribeTopicPartitionsResponseData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.server.common.FinalizedFeatures;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public interface MetadataCache extends ConfigRepository {
+
+ /**
+ * Return topic metadata for a given set of topics and listener. See KafkaApis#handleTopicMetadataRequest for details
+ * on the use of the two boolean flags.
+ *
+ * @param topics The set of topics.
+ * @param listenerName The listener name.
+ * @param errorUnavailableEndpoints If true, we return an error on unavailable brokers. This is used to support
+ * MetadataResponse version 0.
+ * @param errorUnavailableListeners If true, return LEADER_NOT_AVAILABLE if the listener is not found on the leader.
+ * This is used for MetadataResponse versions 0-5.
+ * @return A collection of topic metadata.
+ */
+ List<MetadataResponseData.MetadataResponseTopic> getTopicMetadata(
+ Set<String> topics,
+ ListenerName listenerName,
+ boolean errorUnavailableEndpoints,
+ boolean errorUnavailableListeners);
+
+ Set<String> getAllTopics();
+
+ Set<TopicPartition> getTopicPartitions(String topicName);
+
+ boolean hasAliveBroker(int brokerId);
+
+ List<BrokerMetadata> getAliveBrokers();
Review Comment:
this is used by test infra only, and we can rewrite the check by `hasAliveBroker`.
*before*
```
TestUtils.waitForCondition(() ->
brokers().values().stream().allMatch(brokerServer -> brokerServer.metadataCache().getAliveBrokers().size() == brokers.size()),
"Failed to wait for publisher to publish the metadata update to each broker.");
```
*after*
```
TestUtils.waitForCondition(() ->
brokers.values().stream().map(BrokerServer::metadataCache)
.allMatch(cache -> brokers.values().stream().map(b -> b.config().brokerId()).allMatch(cache::hasAliveBroker)),
"Failed to wait for publisher to publish the metadata update to each broker.");
```
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2682397445
@mumrah, thanks for the suggestion and review. I will do benchmark and fix the conflict tomorrow. Appreciate 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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #18801:
URL: https://github.com/apache/kafka/pull/18801#issuecomment-2657861038
@FrankYang0529 please fix the conflicts, 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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #18801:
URL: https://github.com/apache/kafka/pull/18801
--
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-17565: Move MetadataCache interface to metadata module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1973109296
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -345,68 +352,75 @@ class KRaftMetadataCache(
Option(_currentImage.cluster.broker(brokerId)).count(_.inControlledShutdown) == 1
}
- override def getAliveBrokers(): Iterable[BrokerMetadata] = getAliveBrokers(_currentImage)
+ override def getAliveBrokers(): util.List[BrokerMetadata] = getAliveBrokers(_currentImage)
- private def getAliveBrokers(image: MetadataImage): Iterable[BrokerMetadata] = {
- image.cluster().brokers().values().asScala.filterNot(_.fenced()).
- map(b => new BrokerMetadata(b.id, b.rack))
+ private def getAliveBrokers(image: MetadataImage): util.List[BrokerMetadata] = {
+ _currentImage.cluster().brokers().values().stream()
Review Comment:
`_currentImage` -> `image`
##########
metadata/src/main/java/org/apache/kafka/metadata/MetadataCache.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.admin.BrokerMetadata;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
+import org.apache.kafka.common.message.DescribeClientQuotasResponseData;
+import org.apache.kafka.common.message.DescribeTopicPartitionsResponseData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.server.common.FinalizedFeatures;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public interface MetadataCache extends ConfigRepository {
+
+ /**
+ * Return topic metadata for a given set of topics and listener. See KafkaApis#handleTopicMetadataRequest for details
+ * on the use of the two boolean flags.
+ *
+ * @param topics The set of topics.
+ * @param listenerName The listener name.
+ * @param errorUnavailableEndpoints If true, we return an error on unavailable brokers. This is used to support
+ * MetadataResponse version 0.
+ * @param errorUnavailableListeners If true, return LEADER_NOT_AVAILABLE if the listener is not found on the leader.
+ * This is used for MetadataResponse versions 0-5.
+ * @return A collection of topic metadata.
+ */
+ List<MetadataResponseData.MetadataResponseTopic> getTopicMetadata(
+ Set<String> topics,
+ ListenerName listenerName,
+ boolean errorUnavailableEndpoints,
+ boolean errorUnavailableListeners);
+
+ Set<String> getAllTopics();
+
+ Set<TopicPartition> getTopicPartitions(String topicName);
+
+ boolean hasAliveBroker(int brokerId);
+
+ List<BrokerMetadata> getAliveBrokers();
+
+ Optional<Long> getAliveBrokerEpoch(int brokerId);
+
+ boolean isBrokerFenced(int brokerId);
+
+ boolean isBrokerShuttingDown(int brokerId);
+
+ Uuid getTopicId(String topicName);
+
+ Optional<String> getTopicName(Uuid topicId);
+
+ Optional<Node> getAliveBrokerNode(int brokerId, ListenerName listenerName);
+
+ List<Node> getAliveBrokerNodes(ListenerName listenerName);
+
+ List<Node> getBrokerNodes(ListenerName listenerName);
+
+ Optional<LeaderAndIsr> getLeaderAndIsr(String topic, int partitionId);
+
+ /**
+ * Return the number of partitions in the given topic, or None if the given topic does not exist.
+ */
+ Optional<Integer> numPartitions(String topic);
+
+ Map<String, Uuid> topicNamesToIds();
Review Comment:
it is useless - maybe we can remove it to simplify the interface.
##########
metadata/src/main/java/org/apache/kafka/metadata/MetadataCache.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.admin.BrokerMetadata;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
+import org.apache.kafka.common.message.DescribeClientQuotasResponseData;
+import org.apache.kafka.common.message.DescribeTopicPartitionsResponseData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.server.common.FinalizedFeatures;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public interface MetadataCache extends ConfigRepository {
+
+ /**
+ * Return topic metadata for a given set of topics and listener. See KafkaApis#handleTopicMetadataRequest for details
+ * on the use of the two boolean flags.
+ *
+ * @param topics The set of topics.
+ * @param listenerName The listener name.
+ * @param errorUnavailableEndpoints If true, we return an error on unavailable brokers. This is used to support
+ * MetadataResponse version 0.
+ * @param errorUnavailableListeners If true, return LEADER_NOT_AVAILABLE if the listener is not found on the leader.
+ * This is used for MetadataResponse versions 0-5.
+ * @return A collection of topic metadata.
+ */
+ List<MetadataResponseData.MetadataResponseTopic> getTopicMetadata(
+ Set<String> topics,
+ ListenerName listenerName,
+ boolean errorUnavailableEndpoints,
+ boolean errorUnavailableListeners);
+
+ Set<String> getAllTopics();
+
+ Set<TopicPartition> getTopicPartitions(String topicName);
+
+ boolean hasAliveBroker(int brokerId);
+
+ List<BrokerMetadata> getAliveBrokers();
+
+ Optional<Long> getAliveBrokerEpoch(int brokerId);
+
+ boolean isBrokerFenced(int brokerId);
+
+ boolean isBrokerShuttingDown(int brokerId);
+
+ Uuid getTopicId(String topicName);
+
+ Optional<String> getTopicName(Uuid topicId);
+
+ Optional<Node> getAliveBrokerNode(int brokerId, ListenerName listenerName);
+
+ List<Node> getAliveBrokerNodes(ListenerName listenerName);
+
+ List<Node> getBrokerNodes(ListenerName listenerName);
+
+ Optional<LeaderAndIsr> getLeaderAndIsr(String topic, int partitionId);
+
+ /**
+ * Return the number of partitions in the given topic, or None if the given topic does not exist.
+ */
+ Optional<Integer> numPartitions(String topic);
+
+ Map<String, Uuid> topicNamesToIds();
+
+ Map<Uuid, String> topicIdsToNames();
+
+ Map.Entry<Map<String, Uuid>, Map<Uuid, String>> topicIdInfo();
Review Comment:
ditto
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -239,35 +239,37 @@ class KRaftMetadataCache(
}
// errorUnavailableEndpoints exists to support v0 MetadataResponses
- override def getTopicMetadata(topics: Set[String],
+ override def getTopicMetadata(topics: util.Set[String],
listenerName: ListenerName,
errorUnavailableEndpoints: Boolean = false,
- errorUnavailableListeners: Boolean = false): Seq[MetadataResponseTopic] = {
+ errorUnavailableListeners: Boolean = false): util.List[MetadataResponseTopic] = {
val image = _currentImage
- topics.toSeq.flatMap { topic =>
- getPartitionMetadata(image, topic, listenerName, errorUnavailableEndpoints, errorUnavailableListeners).map { partitionMetadata =>
- new MetadataResponseTopic()
- .setErrorCode(Errors.NONE.code)
- .setName(topic)
- .setTopicId(Option(image.topics().getTopic(topic).id()).getOrElse(Uuid.ZERO_UUID))
- .setIsInternal(Topic.isInternal(topic))
- .setPartitions(partitionMetadata.toBuffer.asJava)
+ topics.stream().flatMap(topic =>
+ getPartitionMetadata(image, topic, listenerName, errorUnavailableEndpoints, errorUnavailableListeners) match {
+ case Some(partitionMetadata) =>
+ util.stream.Stream.of(new MetadataResponseTopic()
+ .setErrorCode(Errors.NONE.code)
+ .setName(topic)
+ .setTopicId(Option(image.topics().getTopic(topic).id()).getOrElse(Uuid.ZERO_UUID))
+ .setIsInternal(Topic.isInternal(topic))
+ .setPartitions(partitionMetadata.toBuffer.asJava))
+ case None => util.stream.Stream.empty()
Review Comment:
please remove the redundant space
##########
test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java:
##########
@@ -290,7 +290,7 @@ default void waitForTopic(String topic, int partitions) throws InterruptedExcept
TestUtils.waitForCondition(
() -> brokers.stream().allMatch(broker -> partitions == 0 ?
broker.metadataCache().numPartitions(topic).isEmpty() :
- broker.metadataCache().numPartitions(topic).contains(partitions)
+ broker.metadataCache().numPartitions(topic).get() == partitions
Review Comment:
`broker.metadataCache().numPartitions(topic).filter(p -> p == partitions).isPresent()`
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -239,35 +239,37 @@ class KRaftMetadataCache(
}
// errorUnavailableEndpoints exists to support v0 MetadataResponses
- override def getTopicMetadata(topics: Set[String],
+ override def getTopicMetadata(topics: util.Set[String],
listenerName: ListenerName,
errorUnavailableEndpoints: Boolean = false,
- errorUnavailableListeners: Boolean = false): Seq[MetadataResponseTopic] = {
+ errorUnavailableListeners: Boolean = false): util.List[MetadataResponseTopic] = {
val image = _currentImage
- topics.toSeq.flatMap { topic =>
- getPartitionMetadata(image, topic, listenerName, errorUnavailableEndpoints, errorUnavailableListeners).map { partitionMetadata =>
- new MetadataResponseTopic()
- .setErrorCode(Errors.NONE.code)
- .setName(topic)
- .setTopicId(Option(image.topics().getTopic(topic).id()).getOrElse(Uuid.ZERO_UUID))
- .setIsInternal(Topic.isInternal(topic))
- .setPartitions(partitionMetadata.toBuffer.asJava)
+ topics.stream().flatMap(topic =>
+ getPartitionMetadata(image, topic, listenerName, errorUnavailableEndpoints, errorUnavailableListeners) match {
+ case Some(partitionMetadata) =>
+ util.stream.Stream.of(new MetadataResponseTopic()
+ .setErrorCode(Errors.NONE.code)
+ .setName(topic)
+ .setTopicId(Option(image.topics().getTopic(topic).id()).getOrElse(Uuid.ZERO_UUID))
+ .setIsInternal(Topic.isInternal(topic))
+ .setPartitions(partitionMetadata.toBuffer.asJava))
+ case None => util.stream.Stream.empty()
}
- }
+ ).collect(Collectors.toList())
}
override def describeTopicResponse(
- topics: Iterator[String],
+ topics: util.Iterator[String],
listenerName: ListenerName,
- topicPartitionStartIndex: String => Int,
+ topicPartitionStartIndex: util.function.Function[String, Integer],
maximumNumberOfPartitions: Int,
ignoreTopicsWithExceptions: Boolean
): DescribeTopicPartitionsResponseData = {
val image = _currentImage
var remaining = maximumNumberOfPartitions
val result = new DescribeTopicPartitionsResponseData()
breakable {
- topics.foreach { topicName =>
+ topics.asScala.foreach { topicName =>
Review Comment:
`forEachRemaining`
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -345,68 +352,75 @@ class KRaftMetadataCache(
Option(_currentImage.cluster.broker(brokerId)).count(_.inControlledShutdown) == 1
}
- override def getAliveBrokers(): Iterable[BrokerMetadata] = getAliveBrokers(_currentImage)
+ override def getAliveBrokers(): util.List[BrokerMetadata] = getAliveBrokers(_currentImage)
- private def getAliveBrokers(image: MetadataImage): Iterable[BrokerMetadata] = {
- image.cluster().brokers().values().asScala.filterNot(_.fenced()).
- map(b => new BrokerMetadata(b.id, b.rack))
+ private def getAliveBrokers(image: MetadataImage): util.List[BrokerMetadata] = {
+ _currentImage.cluster().brokers().values().stream()
+ .filter(Predicate.not(_.fenced))
+ .map(broker => new BrokerMetadata(broker.id, broker.rack))
+ .collect(Collectors.toList())
}
- override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): Option[Node] = {
- Option(_currentImage.cluster().broker(brokerId)).filterNot(_.fenced()).
- flatMap(_.node(listenerName.value()).toScala)
+ override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): util.Optional[Node] = {
+ util.Optional.ofNullable(_currentImage.cluster().broker(brokerId))
+ .filter(Predicate.not(_.fenced))
+ .flatMap(broker => broker.node(listenerName.value))
}
- override def getAliveBrokerNodes(listenerName: ListenerName): Seq[Node] = {
- _currentImage.cluster().brokers().values().asScala.filterNot(_.fenced()).
- flatMap(_.node(listenerName.value()).toScala).toSeq
+ override def getAliveBrokerNodes(listenerName: ListenerName): util.List[Node] = {
+ _currentImage.cluster.brokers.values.stream
+ .filter(Predicate.not(_.fenced))
+ .flatMap(broker => broker.node(listenerName.value).stream)
+ .collect(Collectors.toList())
}
- override def getBrokerNodes(listenerName: ListenerName): Seq[Node] = {
- _currentImage.cluster().brokers().values().asScala.flatMap(_.node(listenerName.value()).toScala).toSeq
+ override def getBrokerNodes(listenerName: ListenerName): util.List[Node] = {
+ _currentImage.cluster.brokers.values.stream
+ .flatMap(broker => broker.node(listenerName.value).stream)
+ .collect(Collectors.toList())
}
- override def getLeaderAndIsr(topicName: String, partitionId: Int): Option[LeaderAndIsr] = {
- Option(_currentImage.topics().getTopic(topicName)).
- flatMap(topic => Option(topic.partitions().get(partitionId))).
- flatMap(partition => Some(new LeaderAndIsr(partition.leader, partition.leaderEpoch,
+ override def getLeaderAndIsr(topicName: String, partitionId: Int): util.Optional[LeaderAndIsr] = {
+ util.Optional.ofNullable(_currentImage.topics().getTopic(topicName)).
+ flatMap(topic => util.Optional.ofNullable(topic.partitions().get(partitionId))).
+ flatMap(partition => util.Optional.ofNullable(new LeaderAndIsr(partition.leader, partition.leaderEpoch,
util.Arrays.asList(partition.isr.map(i => i: java.lang.Integer): _*), partition.leaderRecoveryState, partition.partitionEpoch)))
}
- override def numPartitions(topicName: String): Option[Int] = {
- Option(_currentImage.topics().getTopic(topicName)).
+ override def numPartitions(topicName: String): util.Optional[Integer] = {
+ util.Optional.ofNullable(_currentImage.topics().getTopic(topicName)).
map(topic => topic.partitions().size())
}
override def topicNamesToIds(): util.Map[String, Uuid] = _currentImage.topics.topicNameToIdView()
override def topicIdsToNames(): util.Map[Uuid, String] = _currentImage.topics.topicIdToNameView()
- override def topicIdInfo(): (util.Map[String, Uuid], util.Map[Uuid, String]) = {
+ override def topicIdInfo(): util.Map.Entry[util.Map[String, Uuid], util.Map[Uuid, String]] = {
val image = _currentImage
- (image.topics.topicNameToIdView(), image.topics.topicIdToNameView())
+ new util.AbstractMap.SimpleEntry(image.topics.topicNameToIdView(), image.topics.topicIdToNameView())
}
// if the leader is not known, return None;
// if the leader is known and corresponding node is available, return Some(node)
// if the leader is known but corresponding node with the listener name is not available, return Some(NO_NODE)
- override def getPartitionLeaderEndpoint(topicName: String, partitionId: Int, listenerName: ListenerName): Option[Node] = {
+ override def getPartitionLeaderEndpoint(topicName: String, partitionId: Int, listenerName: ListenerName): util.Optional[Node] = {
val image = _currentImage
Option(image.topics().getTopic(topicName)) match {
- case None => None
+ case None => util.Optional.empty()
case Some(topic) => Option(topic.partitions().get(partitionId)) match {
- case None => None
+ case None => util.Optional.empty()
case Some(partition) => Option(image.cluster().broker(partition.leader)) match {
- case None => Some(Node.noNode)
- case Some(broker) => Some(broker.node(listenerName.value()).orElse(Node.noNode()))
+ case None => util.Optional.of(Node.noNode)
+ case Some(broker) => util.Optional.of(broker.node(listenerName.value()).orElse(Node.noNode()))
}
}
}
}
- override def getPartitionReplicaEndpoints(tp: TopicPartition, listenerName: ListenerName): Map[Int, Node] = {
+ override def getPartitionReplicaEndpoints(tp: TopicPartition, listenerName: ListenerName): util.Map[Integer, Node] = {
val image = _currentImage
- val result = new mutable.HashMap[Int, Node]()
+ val result = new mutable.HashMap[Integer, Node]()
Review Comment:
could you please use java Map instead?
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -186,9 +188,9 @@ class AddPartitionsToTxnManager(
}
private def getTransactionCoordinator(partition: Int): Option[Node] = {
Review Comment:
we can use Java optional as it has only one usage
##########
metadata/src/main/java/org/apache/kafka/metadata/MetadataCache.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * 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.metadata;
+
+import org.apache.kafka.admin.BrokerMetadata;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.DescribeClientQuotasRequestData;
+import org.apache.kafka.common.message.DescribeClientQuotasResponseData;
+import org.apache.kafka.common.message.DescribeTopicPartitionsResponseData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsRequestData;
+import org.apache.kafka.common.message.DescribeUserScramCredentialsResponseData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.server.common.FinalizedFeatures;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public interface MetadataCache extends ConfigRepository {
+
+ /**
+ * Return topic metadata for a given set of topics and listener. See KafkaApis#handleTopicMetadataRequest for details
+ * on the use of the two boolean flags.
+ *
+ * @param topics The set of topics.
+ * @param listenerName The listener name.
+ * @param errorUnavailableEndpoints If true, we return an error on unavailable brokers. This is used to support
+ * MetadataResponse version 0.
+ * @param errorUnavailableListeners If true, return LEADER_NOT_AVAILABLE if the listener is not found on the leader.
+ * This is used for MetadataResponse versions 0-5.
+ * @return A collection of topic metadata.
+ */
+ List<MetadataResponseData.MetadataResponseTopic> getTopicMetadata(
+ Set<String> topics,
+ ListenerName listenerName,
+ boolean errorUnavailableEndpoints,
+ boolean errorUnavailableListeners);
+
+ Set<String> getAllTopics();
+
+ Set<TopicPartition> getTopicPartitions(String topicName);
+
+ boolean hasAliveBroker(int brokerId);
+
+ List<BrokerMetadata> getAliveBrokers();
+
+ Optional<Long> getAliveBrokerEpoch(int brokerId);
+
+ boolean isBrokerFenced(int brokerId);
+
+ boolean isBrokerShuttingDown(int brokerId);
+
+ Uuid getTopicId(String topicName);
+
+ Optional<String> getTopicName(Uuid topicId);
+
+ Optional<Node> getAliveBrokerNode(int brokerId, ListenerName listenerName);
+
+ List<Node> getAliveBrokerNodes(ListenerName listenerName);
+
+ List<Node> getBrokerNodes(ListenerName listenerName);
+
+ Optional<LeaderAndIsr> getLeaderAndIsr(String topic, int partitionId);
+
+ /**
+ * Return the number of partitions in the given topic, or None if the given topic does not exist.
+ */
+ Optional<Integer> numPartitions(String topic);
+
+ Map<String, Uuid> topicNamesToIds();
+
+ Map<Uuid, String> topicIdsToNames();
+
+ Map.Entry<Map<String, Uuid>, Map<Uuid, String>> topicIdInfo();
+
+ /**
+ * Get a partition leader's endpoint
+ *
+ * @return If the leader is known, and the listener name is available, return Some(node). If the leader is known,
+ * but the listener is unavailable, return Some(Node.NO_NODE). Otherwise, if the leader is not known,
+ * return None
+ */
+ Optional<Node> getPartitionLeaderEndpoint(String topic, int partitionId, ListenerName listenerName);
+
+ Map<Integer, Node> getPartitionReplicaEndpoints(TopicPartition tp, ListenerName listenerName);
+
+ Cluster getClusterMetadata(String clusterId, ListenerName listenerName);
Review Comment:
ditto
--
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-17565: Move MetadataCache interface to metadata module (wip) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18801:
URL: https://github.com/apache/kafka/pull/18801#discussion_r1941425411
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -345,68 +344,69 @@ class KRaftMetadataCache(
Option(_currentImage.cluster.broker(brokerId)).count(_.inControlledShutdown) == 1
}
- override def getAliveBrokers(): Iterable[BrokerMetadata] = getAliveBrokers(_currentImage)
+ override def getAliveBrokers(): util.List[BrokerMetadata] = getAliveBrokers(_currentImage)
- private def getAliveBrokers(image: MetadataImage): Iterable[BrokerMetadata] = {
+ private def getAliveBrokers(image: MetadataImage): util.List[BrokerMetadata] = {
image.cluster().brokers().values().asScala.filterNot(_.fenced()).
map(b => new BrokerMetadata(b.id, b.rack))
- }
+ }.toList.asJava
- override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): Option[Node] = {
- Option(_currentImage.cluster().broker(brokerId)).filterNot(_.fenced()).
- flatMap(_.node(listenerName.value()).toScala)
+ override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName): util.Optional[Node] = {
+ util.Optional.ofNullable(_currentImage.cluster().broker(brokerId))
+ .filter(broker => !broker.fenced())
+ .flatMap(broker => broker.node(listenerName.value()))
}
- override def getAliveBrokerNodes(listenerName: ListenerName): Seq[Node] = {
+ override def getAliveBrokerNodes(listenerName: ListenerName): util.List[Node] = {
_currentImage.cluster().brokers().values().asScala.filterNot(_.fenced()).
Review Comment:
We should remove `asScala` from here and use `stream` instead.
--
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