hachikuji commented on a change in pull request #10887:
URL: https://github.com/apache/kafka/pull/10887#discussion_r660883703
##########
File path: core/src/main/scala/kafka/server/MetadataCache.scala
##########
@@ -62,17 +46,22 @@ trait MetadataCache {
def getAllTopics(): collection.Set[String]
- def getAllPartitions(): collection.Set[TopicPartition]
+ def getTopicPartitions(topicName: String): collection.Set[TopicPartition]
- def getNonExistingTopics(topics: collection.Set[String]):
collection.Set[String]
+ def hasAliveBroker(brokerId: Int): Boolean
- def getAliveBroker(brokerId: Int): Option[MetadataBroker]
+ def getAliveBrokers(): Iterable[BrokerMetadata]
- def getAliveBrokers: collection.Seq[MetadataBroker]
+ def getAliveBrokerNode(brokerId: Int, listenerName: String): Option[Node]
+
+ def getAliveBrokerNodes(listenerName: String): Iterable[Node]
Review comment:
Would it make sense to use `ListenerName` in these APIs? As far as I can
tell, all of the usages are just using `ListenerName.value`. Also, in
`ZkMetadataCache`, we end up constructing a new `ListenerName` from the string.
##########
File path: core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala
##########
@@ -0,0 +1,406 @@
+/**
+ * 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 kafka.server
+
+import java.util
+import java.util.Collections
+import java.util.concurrent.locks.ReentrantReadWriteLock
+
+import kafka.admin.BrokerMetadata
+
+import scala.collection.{Seq, Set, mutable}
+import scala.jdk.CollectionConverters._
+import kafka.cluster.{Broker, EndPoint}
+import kafka.api._
+import kafka.controller.StateChangeLogger
+import kafka.utils.CoreUtils._
+import kafka.utils.Logging
+import kafka.utils.Implicits._
+import org.apache.kafka.common.internals.Topic
+import
org.apache.kafka.common.message.UpdateMetadataRequestData.UpdateMetadataPartitionState
+import org.apache.kafka.common.{Cluster, Node, PartitionInfo, TopicPartition,
Uuid}
+import
org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic
+import
org.apache.kafka.common.message.MetadataResponseData.MetadataResponsePartition
+import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.requests.{MetadataResponse,
UpdateMetadataRequest}
+import org.apache.kafka.common.security.auth.SecurityProtocol
+
+/**
+ * A cache for the state (e.g., current leader) of each partition. This cache
is updated through
+ * UpdateMetadataRequest from the controller. Every broker maintains the same
cache, asynchronously.
+ */
+class ZkMetadataCache(brokerId: Int) extends MetadataCache with Logging {
Review comment:
I'm wondering if we could leave this class in its current place just for
this PR. It is a little hard to see the diffs. In a trivial follow-up, we can
factor it out.
##########
File path: core/src/main/scala/kafka/server/MetadataCache.scala
##########
@@ -62,17 +46,22 @@ trait MetadataCache {
def getAllTopics(): collection.Set[String]
- def getAllPartitions(): collection.Set[TopicPartition]
+ def getTopicPartitions(topicName: String): collection.Set[TopicPartition]
- def getNonExistingTopics(topics: collection.Set[String]):
collection.Set[String]
+ def hasAliveBroker(brokerId: Int): Boolean
- def getAliveBroker(brokerId: Int): Option[MetadataBroker]
+ def getAliveBrokers(): Iterable[BrokerMetadata]
- def getAliveBrokers: collection.Seq[MetadataBroker]
+ def getAliveBrokerNode(brokerId: Int, listenerName: String): Option[Node]
+
+ def getAliveBrokerNodes(listenerName: String): Iterable[Node]
def getPartitionInfo(topic: String, partitionId: Int):
Option[UpdateMetadataRequestData.UpdateMetadataPartitionState]
- def numPartitions(topic: String): Option[Int]
+ /**
+ * Return the number of partitions in the given topic, or 0 if the given
topic does not exist.
+ */
+ def numPartitions(topic: String): Int
Review comment:
An alternative might be to raise an exception if the topic does not
exist? I agree with Jose that we are likely to have bugs resulting from the
caller forgetting to handle the case of 0.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -1721,8 +1720,10 @@ class ReplicaManager(val config: KafkaConfig,
} else {
// we do not need to check if the leader exists again since this has
been done at the beginning of this process
val partitionsToMakeFollowerWithLeaderAndOffset =
partitionsToMakeFollower.map { partition =>
- val leader = metadataCache.getAliveBrokers.find(_.id ==
partition.leaderReplicaIdOpt.get).get
- .brokerEndPoint(config.interBrokerListenerName)
+ val leaderNode = metadataCache.getAliveBrokerNode(
+ partition.leaderReplicaIdOpt.getOrElse(-1),
config.interBrokerListenerName.value()).
Review comment:
It's a little strange to assume `getAliveBrokerNode` works with -1 as a
broker id. Are we assuming that the implementation returns `noNode` in that
case? I would probably try to write this like this:
```scala
partition.leaderReplicaIdOpt.flatMap(leaderId =>
metadataCache.getAliveBrokerNode(leaderId,
config.interBrokerListenerName.value))
.getOrElse(Node.noNode())
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
