mumrah commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046431404
##########
clients/src/main/resources/common/message/StopReplicaRequest.json:
##########
@@ -31,7 +31,7 @@
"fields": [
{ "name": "ControllerId", "type": "int32", "versions": "0+", "entityType":
"brokerId",
"about": "The controller id." },
- { "name": "KRaftControllerId", "type": "int32", "versions": "4+",
"entityType": "brokerId", "default": "-1",
+ { "name": "KRaftControllerId", "type": "int32", "versions": "4+",
"default": "-1", "entityType": "brokerId", "default": "-1",
Review Comment:
Same here
##########
clients/src/main/resources/common/message/LeaderAndIsrRequest.json:
##########
@@ -36,7 +36,7 @@
"fields": [
{ "name": "ControllerId", "type": "int32", "versions": "0+", "entityType":
"brokerId",
"about": "The current controller ID." },
- { "name": "KRaftControllerId", "type": "int32", "versions": "7+",
"entityType": "brokerId", "default": "-1",
+ { "name": "KRaftControllerId", "type": "int32", "versions": "7+",
"default": "-1", "entityType": "brokerId", "default": "-1",
Review Comment:
You've got "default" twice here
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -176,13 +190,13 @@ class BrokerToControllerChannelManagerImpl(
}
private[server] def newRequestThread = {
- val networkClient = {
+ def networkClient(controllerInfo: ControllerInformation) = {
Review Comment:
Can we name this "buildNetworkClient" or "createNetworkClient" to make it
clear we are actually making a new one?
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -276,17 +290,38 @@ case class BrokerToControllerQueueItem(
)
class BrokerToControllerRequestThread(
- networkClient: KafkaClient,
+ networkClientFactory: ControllerInformation => KafkaClient,
metadataUpdater: ManualMetadataUpdater,
controllerNodeProvider: ControllerNodeProvider,
config: KafkaConfig,
time: Time,
threadName: String,
retryTimeoutMs: Long
-) extends InterBrokerSendThread(threadName, networkClient,
Math.min(Int.MaxValue, Math.min(config.controllerSocketTimeoutMs,
retryTimeoutMs)).toInt, time, isInterruptible = false) {
+) extends InterBrokerSendThread(threadName, null, Math.min(Int.MaxValue,
Math.min(config.controllerSocketTimeoutMs, retryTimeoutMs)).toInt, time,
isInterruptible = false) {
+
+ var isZkController = false
Review Comment:
Let's pick a more descriptive name, maybe "isNetworkClientForZkController"
or something, or we could cache the ControllerInformation instead of just this
boolean
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1318,14 +1317,20 @@ class KafkaApis(val requestChannel: RequestChannel,
trace("Sending topic metadata %s and brokers %s for correlation id %d to
client %s".format(completeTopicMetadata.mkString(","),
brokers.mkString(","), request.header.correlationId,
request.header.clientId))
+ val controllerId = {
+ metadataCache.getControllerId.flatMap {
+ case ZkCachedControllerId(id) => Some(id)
+ case KRaftCachedControllerId(_) => metadataCache.getRandomAliveBrokerId
Review Comment:
See my note in MetadataCache. We could simplify this to
`metadataCache.getControllerNode.map(_.id)`
##########
core/src/main/scala/kafka/server/MetadataCache.scala:
##########
@@ -32,6 +32,13 @@ case class FinalizedFeaturesAndEpoch(features: Map[String,
Short], epoch: Long)
}
}
+sealed trait CachedControllerId {
Review Comment:
A short javadoc might be useful here.
##########
core/src/main/scala/kafka/server/KafkaServer.scala:
##########
@@ -147,6 +147,7 @@ class KafkaServer(
var kafkaScheduler: KafkaScheduler = _
+ var kraftControllerNodes: Seq[Node] = Seq.empty
Review Comment:
Let's just initialize this to `_` like the other class members here. It's
not pretty, but it's better to be consistent
##########
core/src/main/scala/kafka/server/MetadataCache.scala:
##########
@@ -103,13 +110,18 @@ trait MetadataCache {
def metadataVersion(): MetadataVersion
def features(): FinalizedFeaturesAndEpoch
+
+ def getRandomAliveBrokerId: Option[Int]
}
object MetadataCache {
def zkMetadataCache(brokerId: Int,
metadataVersion: MetadataVersion,
- brokerFeatures: BrokerFeatures =
BrokerFeatures.createEmpty()): ZkMetadataCache = {
- new ZkMetadataCache(brokerId, metadataVersion, brokerFeatures)
+ brokerFeatures: BrokerFeatures =
BrokerFeatures.createEmpty(),
+ kraftControllerNodes: collection.Seq[Node] = null)
Review Comment:
Can we initialize this as an empty Seq instead of null?
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -248,7 +258,12 @@ class ZkMetadataCache(brokerId: Int, metadataVersion:
MetadataVersion, brokerFea
}
override def getAliveBrokerNode(brokerId: Int, listenerName: ListenerName):
Option[Node] = {
-
metadataSnapshot.aliveBrokers.get(brokerId).flatMap(_.getNode(listenerName))
+ val snapshot = metadataSnapshot
+ brokerId match {
+ case id if
snapshot.controllerId.filter(_.isInstanceOf[KRaftCachedControllerId]).exists(_.id
== id) =>
+ kraftControllerNodeMap.get(id)
+ case _ =>
snapshot.aliveBrokers.get(brokerId).flatMap(_.getNode(listenerName))
+ }
Review Comment:
We should be able to use pattern matching a little more cleanly like:
```suggestion
val snapshot = metadataSnapshot
snapshot.controllerId match {
case Some(controllerId: KRaftCachedControllerId) =>
kraftControllerNodeMap.get(controllerId.id)
case _ =>
snapshot.aliveBrokers.get(brokerId).flatMap(_.getNode(listenerName))
}
```
##########
core/src/main/scala/kafka/server/KafkaServer.scala:
##########
@@ -668,7 +684,7 @@ class KafkaServer(
// 1. Find the controller and establish a connection to it.
// If the controller id or the broker registration are missing, we
sleep and retry (if there are remaining retries)
- metadataCache.getControllerId match {
+
metadataCache.getControllerId.filter(_.isInstanceOf[ZkCachedControllerId]).map(_.id)
match {
Review Comment:
If we leave this as `metadataCache.getControllerId match`, we can just add
the type to the cases below. E.g.,
```
case Some(controllerId: ZkCachedControllerId) => ...
```
##########
core/src/main/scala/kafka/server/MetadataCache.scala:
##########
@@ -103,13 +110,18 @@ trait MetadataCache {
def metadataVersion(): MetadataVersion
def features(): FinalizedFeaturesAndEpoch
+
+ def getRandomAliveBrokerId: Option[Int]
Review Comment:
Would it make sense to add `def getControllerNode: Option[Node]` here? It
might allow for less complex usage outside of this class (e.g., maybe we can
avoid checking kraft/vs in some cases?)
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -276,17 +290,38 @@ case class BrokerToControllerQueueItem(
)
class BrokerToControllerRequestThread(
- networkClient: KafkaClient,
+ networkClientFactory: ControllerInformation => KafkaClient,
metadataUpdater: ManualMetadataUpdater,
controllerNodeProvider: ControllerNodeProvider,
config: KafkaConfig,
time: Time,
threadName: String,
retryTimeoutMs: Long
-) extends InterBrokerSendThread(threadName, networkClient,
Math.min(Int.MaxValue, Math.min(config.controllerSocketTimeoutMs,
retryTimeoutMs)).toInt, time, isInterruptible = false) {
+) extends InterBrokerSendThread(threadName, null, Math.min(Int.MaxValue,
Math.min(config.controllerSocketTimeoutMs, retryTimeoutMs)).toInt, time,
isInterruptible = false) {
Review Comment:
I wonder if we could do this slightly differently. Instead of passing a
`null` and using the `maybeResetNetworkClient` call in the class body, could we
make a companion object that does the initialization and passes in an
initialized NetworkClient?
e.g.,
```scala
class BrokerToControllerRequestThread(
initialNetworkClient: KafkaClient,
networkClientFactory: ControllerInformation => KafkaClient,
metadataUpdater: ManualMetadataUpdater,
controllerNodeProvider: ControllerNodeProvider,
config: KafkaConfig,
time: Time,
threadName: String,
retryTimeoutMs: Long
) extends InterBrokerSendThread(threadName, initialNetworkClient,
Math.min(Int.MaxValue, Math.min(config.controllerSocketTimeoutMs,
retryTimeoutMs)).toInt, time, isInterruptible = false) {
```
This way we never have a `null` reference for
InterBrokerSendThread#networkClient.
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -60,16 +60,24 @@ trait ZkFinalizedFeatureCache {
* 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, metadataVersion: MetadataVersion,
brokerFeatures: BrokerFeatures)
+class ZkMetadataCache(
+ brokerId: Int,
+ metadataVersion: MetadataVersion,
+ brokerFeatures: BrokerFeatures,
+ kraftControllerNodes: Seq[Node] = Seq.empty)
Review Comment:
Do we need the default value here? (maybe it's there just for tests?)
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -37,42 +38,55 @@ import scala.collection.Seq
import scala.compat.java8.OptionConverters._
import scala.jdk.CollectionConverters._
-trait ControllerNodeProvider {
- def get(): Option[Node]
- def listenerName: ListenerName
- def securityProtocol: SecurityProtocol
- def saslMechanism: String
-}
+case class ControllerInformation(node: Option[Node],
+ listenerName: ListenerName,
+ securityProtocol: SecurityProtocol,
+ saslMechanism: String,
+ isZkController: Boolean)
Review Comment:
nit: prefer the new continuation style:
```
class Foo(
arg1: T,
arg2: T
)
```
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -364,11 +399,13 @@ class BrokerToControllerRequestThread(
}
override def doWork(): Unit = {
Review Comment:
Previously, we would only check ControllerNodeProvider for new connection
information if we received NOT_CONTROLLER or got disconnected. Can we keep with
this paradigm and only check ControllerNodeProvider in these cases?
If the controller has moved from ZK to ZK, or ZK to KRaft, we'll likely
learn about it first via NOT_CONTROLLER anyways.
My main concern is not changing too much behavior in this class. Ideally, in
non-migration ZK mode, things here will behave exactly the same as they are
today.
--
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
