You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/12/07 03:01:44 UTC
[GitHub] [kafka] akhileshchg opened a new pull request, #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
akhileshchg opened a new pull request, #12961:
URL: https://github.com/apache/kafka/pull/12961
KAFKA-14446: API forwarding support from zkBrokers to the controller
**Changes**
* Enable `EnvelopeRequest` to support zkBroker to zkController forwarding.
* Changes to `BrokerToControllerChannelManager`, which we use in every broker to maintain a channel to the controller node. This node and security protocol to connect to the node can keep changing during zk -> kraft migration. Also, added support to reinitialize the request thread used in `BrokerToControllerChannelManager` when the controller switches from zkBroker to kraftController and vice-versa.
* Other peripheral changes required to support the above changes.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050047950
##########
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:
Yes.
--
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
[GitHub] [kafka] mumrah commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
mumrah commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050076979
##########
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:
Ah, i see, i missed `exists` part. This could still be cleaned up with pattern matching, but it's not critical for this PR
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046449684
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1665,6 +1665,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
def usesSelfManagedQuorum: Boolean = processRoles.nonEmpty
val migrationEnabled: Boolean = getBoolean(KafkaConfig.MigrationEnabledProp)
+ def enableZkApiForwarding: Boolean = migrationEnabled && interBrokerProtocolVersion.isApiForwardingEnabled
Review Comment:
please don't add a reference to `KafkaConfig.interBrokerProtocolVersion` here. That is going away since we are moving away from statically configuring IBP (or indeed calling it IBP)
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1042760995
##########
clients/src/main/resources/common/message/EnvelopeRequest.json:
##########
@@ -16,10 +16,11 @@
{
"apiKey": 58,
"type": "request",
- "listeners": ["controller"],
+ "listeners": ["controller", "zkBroker"],
"name": "EnvelopeRequest",
// Request struct for forwarding.
- "validVersions": "0",
+ // Version 1 adds the envelope request support to ZK brokers as well.
+ "validVersions": "0-1",
Review Comment:
hmm, I don't think this needs a version bump.
Even if we did the version bump, if a ZK-based broker sent a version 0 request it would work fine, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dengziming commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1042151200
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1665,6 +1665,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
def usesSelfManagedQuorum: Boolean = processRoles.nonEmpty
val migrationEnabled: Boolean = getBoolean(KafkaConfig.MigrationEnabledProp)
+ def enableZkApiForwarding: Boolean = migrationEnabled && interBrokerProtocolVersion.isApiForwardingSupported
Review Comment:
Why should we check `migrationEnabled` here? does this mean we will not enable forward in zk mode except for migration?
##########
clients/src/main/java/org/apache/kafka/common/requests/AbstractControlRequest.java:
##########
@@ -43,6 +49,8 @@ protected AbstractControlRequest(ApiKeys api, short version) {
public abstract int controllerId();
+ public abstract int kraftControllerId();
Review Comment:
I didn't see how is this field used in `KafkaApis`.
##########
core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala:
##########
@@ -148,13 +149,15 @@ class CreateTopicsRequestTest extends AbstractCreateTopicsRequestTest {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk"))
+ @ValueSource(strings = Array("zk", "zkMigration"))
def testNotController(quorum: String): Unit = {
// Note: we don't run this test when in KRaft mode, because KRaft doesn't have this
// behavior of returning NOT_CONTROLLER. Instead, the request is forwarded.
val req = topicsReq(Seq(topicReq("topic1")))
val response = sendCreateTopicRequest(req, notControllerSocketServer)
- assertEquals(1, response.errorCounts().get(Errors.NOT_CONTROLLER))
+ response.data().topics().asScala.foreach(topic => topic.errorCode())
Review Comment:
This line is unused.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1042776473
##########
clients/src/main/resources/common/message/EnvelopeRequest.json:
##########
@@ -16,10 +16,11 @@
{
"apiKey": 58,
"type": "request",
- "listeners": ["controller"],
+ "listeners": ["controller", "zkBroker"],
"name": "EnvelopeRequest",
// Request struct for forwarding.
- "validVersions": "0",
+ // Version 1 adds the envelope request support to ZK brokers as well.
+ "validVersions": "0-1",
Review Comment:
That makes sense. Let me revert the version to 0.
##########
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._
+case class ControllerInformation(node: Option[Node],
+ listenerName: ListenerName,
+ securityProtocol: SecurityProtocol,
+ saslMechanism: String,
+ isZkController: Boolean)
+
trait ControllerNodeProvider {
def get(): Option[Node]
- def listenerName: ListenerName
- def securityProtocol: SecurityProtocol
- def saslMechanism: String
-}
-
-object MetadataCacheControllerNodeProvider {
- def apply(
- config: KafkaConfig,
- metadataCache: kafka.server.MetadataCache
- ): MetadataCacheControllerNodeProvider = {
- val listenerName = config.controlPlaneListenerName
- .getOrElse(config.interBrokerListenerName)
-
- val securityProtocol = config.controlPlaneSecurityProtocol
- .getOrElse(config.interBrokerSecurityProtocol)
-
- new MetadataCacheControllerNodeProvider(
- metadataCache,
- listenerName,
- securityProtocol,
- config.saslMechanismInterBrokerProtocol
- )
- }
+ def getControllerInfo(): ControllerInformation
}
class MetadataCacheControllerNodeProvider(
- val metadataCache: kafka.server.MetadataCache,
- val listenerName: ListenerName,
- val securityProtocol: SecurityProtocol,
- val saslMechanism: String
+ val metadataCache: ZkMetadataCache,
+ val config: KafkaConfig
) extends ControllerNodeProvider {
+
+ def listenerName(isZkController: Boolean): ListenerName = {
Review Comment:
Whether this is part of the metadata cache or node provider, we still need to check for listername, security protocol and SASLMechanism needed to communicate with the controller based on if we're trying to connect to zkController or kraftController. I am not sure I completely understood your comment.
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -164,25 +184,63 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
- private val requestThread = newRequestThread
+ @volatile private var isZkControllerThread = true
+ @volatile private var requestThread = newRequestThread()
+ @volatile private var shutdownStarted = false
def start(): Unit = {
requestThread.start()
+ maybeScheduleReinitializeRequestThread()
}
def shutdown(): Unit = {
requestThread.shutdown()
+ shutdownStarted = false
info(s"Broker to controller channel manager for $channelName shutdown")
}
- private[server] def newRequestThread = {
+ def maybeScheduleReinitializeRequestThread(): Unit = {
+ // If migration is enabled for zkBroker, then we might see controller change from zk to kraft
+ // and vice-versa. This periodic task takes care of setting the right channel when such
+ // controller change is noticed.
+ if (config.migrationEnabled && config.requiresZookeeper) {
Review Comment:
I tried to swap out just the networkclient in interBrokerSendThread. Let me know how it looks now.
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046455192
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -355,6 +375,13 @@ class ZkMetadataCache(brokerId: Int, metadataVersion: MetadataVersion, brokerFea
case id if id < 0 => None
case id => Some(id)
}
+ val kraftControllerIdOpt = updateMetadataRequest.kraftControllerId() match {
+ case id if id < 0 => None
+ case id => Some(id)
+ }
+ if (controllerIdOpt.nonEmpty && kraftControllerIdOpt.nonEmpty) {
Review Comment:
This isn't an error, right? It's possible for both of them to be -1... Like sometimes we are in the middle of electing a new controller.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050037434
##########
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:
Okay. I re-read the comment again. Let me make the change and tests again.
--
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
[GitHub] [kafka] cmccabe merged pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe merged PR #12961:
URL: https://github.com/apache/kafka/pull/12961
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050065115
##########
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:
This is wrong. In Kraft brokers and Zk brokers with KRaft controller. It always returns controller information for any broker node query.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050074366
##########
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:
I did something like this before. But got a different comment then, saying we should explicitly handle that in KafkaApis.
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046446539
##########
clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java:
##########
@@ -51,7 +51,16 @@ public static class Builder extends AbstractControlRequest.Builder<LeaderAndIsrR
public Builder(short version, int controllerId, int controllerEpoch, long brokerEpoch,
List<LeaderAndIsrPartitionState> partitionStates, Map<String, Uuid> topicIds,
Collection<Node> liveLeaders) {
- super(ApiKeys.LEADER_AND_ISR, version, controllerId, controllerEpoch, brokerEpoch);
+ super(ApiKeys.LEADER_AND_ISR, version, controllerId, controllerEpoch, brokerEpoch, false);
Review Comment:
let's define this in terms of the other constructor to avoid duplicating the below code. something like:
```
this(version, controllerId, controllerEpoch, brokerEpoch, false)
```
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1042759981
##########
clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java:
##########
@@ -66,10 +75,14 @@ public LeaderAndIsrRequest build(short version) {
).collect(Collectors.toList());
LeaderAndIsrRequestData data = new LeaderAndIsrRequestData()
- .setControllerId(controllerId)
.setControllerEpoch(controllerEpoch)
.setBrokerEpoch(brokerEpoch)
.setLiveLeaders(leaders);
+ if (kraftController) {
+ data.setKRaftControllerId(controllerId).setControllerId(-1);
Review Comment:
Maybe this is just me, but this if block seems confusing
can we do something like
```
.setControllerId(kraftController ? -1 : controllerId)
.setKRaftControllerId(kraftController ? controllerId : -1)
```
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046452203
##########
core/src/main/scala/kafka/server/KafkaServer.scala:
##########
@@ -544,6 +548,7 @@ class KafkaServer(
private def controlledShutdown(): Unit = {
val socketTimeoutMs = config.controllerSocketTimeoutMs
+ // TODO (KAFKA-14447): Handle controlled shutdown for zkBroker when we have KRaft controller.
Review Comment:
Can we just log a message and `return` from this function early, when we are in this situation? Currently, it looks like we just get stuck and time out, which is not great
--
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
[GitHub] [kafka] mumrah commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
mumrah commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1042322800
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -37,42 +38,53 @@ import scala.collection.Seq
import scala.compat.java8.OptionConverters._
import scala.jdk.CollectionConverters._
+case class ControllerInformation(node: Option[Node], listenerName: ListenerName,
Review Comment:
style nit: one param per line
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -164,25 +182,54 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
- private val requestThread = newRequestThread
+ private var isZkControllerThread = true
+ @volatile private var requestThread = newRequestThread
def start(): Unit = {
requestThread.start()
+ // If migration is enabled for zkBroker, then we might see controller change from zk to kraft
+ // and vice-versa. This periodic task takes care of setting the right channel when such
+ // controller change is noticed.
+ if (config.migrationEnabled && config.requiresZookeeper) {
+ kafkaScheduler.schedule("maybeReinitializeRequestThread",
+ () => maybeReinitializeRequestThread(), 0L, 500, TimeUnit.MILLISECONDS)
+ }
}
def shutdown(): Unit = {
requestThread.shutdown()
info(s"Broker to controller channel manager for $channelName shutdown")
}
+ def maybeReinitializeRequestThread(): Unit = {
+ val controllerInfo = controllerNodeProvider.getControllerInfo()
+ if (isZkControllerThread == controllerInfo.isZkController)
+ return
+ val oldRequestThread = requestThread
+ // Close the old thread and start a new one if old one is not already in the process of
+ // shutting down.
+ if (oldRequestThread.isRunning) {
Review Comment:
If the old thread is not running, what would that mean here?
##########
core/src/main/scala/kafka/server/KafkaServer.scala:
##########
@@ -544,6 +549,7 @@ class KafkaServer(
private def controlledShutdown(): Unit = {
val socketTimeoutMs = config.controllerSocketTimeoutMs
+ // TODO: Handle controlled shutdown for zkBroker when we have KRaft controller.
Review Comment:
I just filed https://issues.apache.org/jira/browse/KAFKA-14447, can you add a comment here
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1665,6 +1665,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
def usesSelfManagedQuorum: Boolean = processRoles.nonEmpty
val migrationEnabled: Boolean = getBoolean(KafkaConfig.MigrationEnabledProp)
+ def enableZkApiForwarding: Boolean = migrationEnabled && interBrokerProtocolVersion.isApiForwardingSupported
Review Comment:
Good question @dengziming. We said in the KIP that forwarding is enabled in all cases with 3.4-IV0, though technically we only need to enable it if we're doing a migration. It might be safer to keep forwarding off when we haven't enabled the migration. Since we won't start the migration until all brokers come up with `zookeeper.metadata.migration.enable`, I think leaving it off like this will be okay.
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -164,25 +182,54 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
- private val requestThread = newRequestThread
+ private var isZkControllerThread = true
+ @volatile private var requestThread = newRequestThread
def start(): Unit = {
requestThread.start()
+ // If migration is enabled for zkBroker, then we might see controller change from zk to kraft
+ // and vice-versa. This periodic task takes care of setting the right channel when such
+ // controller change is noticed.
+ if (config.migrationEnabled && config.requiresZookeeper) {
+ kafkaScheduler.schedule("maybeReinitializeRequestThread",
+ () => maybeReinitializeRequestThread(), 0L, 500, TimeUnit.MILLISECONDS)
+ }
}
def shutdown(): Unit = {
requestThread.shutdown()
info(s"Broker to controller channel manager for $channelName shutdown")
}
+ def maybeReinitializeRequestThread(): Unit = {
+ val controllerInfo = controllerNodeProvider.getControllerInfo()
+ if (isZkControllerThread == controllerInfo.isZkController)
+ return
+ val oldRequestThread = requestThread
+ // Close the old thread and start a new one if old one is not already in the process of
+ // shutting down.
+ if (oldRequestThread.isRunning) {
+ val newThread = newRequestThread
+ newThread.start()
+ requestThread = newThread
+ // Shutdown the request thread. This fails the inflight requests to be failed with
+ // disconnection exception.
+ oldRequestThread.shutdown()
+ // There can still be unsent requests in the queue which need to be cleaned.
+ oldRequestThread.failAllUnsentRequests()
+ }
+ }
+
private[server] def newRequestThread = {
Review Comment:
Can we pass the `ControllerInfo` in here instead of re-reading it?
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -164,25 +182,54 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
- private val requestThread = newRequestThread
+ private var isZkControllerThread = true
Review Comment:
Should this be volatile as well?
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -243,6 +243,10 @@ public boolean isNoOpRecordSupported() {
return this.isAtLeast(IBP_3_3_IV1);
}
+ public boolean isApiForwardingSupported() {
Review Comment:
nit: Should we say "isApiForwardingEnabled" since we require API forwarding at this version?
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -315,7 +324,35 @@ class ZkMetadataCache(brokerId: Int, metadataVersion: MetadataVersion, brokerFea
}.getOrElse(Map.empty[Int, Node])
}
- def getControllerId: Option[Int] = metadataSnapshot.controllerId
+ def getControllerId: Option[Int] = {
+ val snapshot = metadataSnapshot
+ snapshot.controllerId.orElse(snapshot.kraftControllerId)
+ }
+
+ /**
+ * Return controller id. Additionally returns if the controller is zk or kraft controller.
+ * Returns true for zk controller, false otherwise.
+ * @return
+ */
+ def getZkOrKRaftControllerId: (Option[Int], Boolean) = {
+ val snapshot = metadataSnapshot
+ snapshot.kraftControllerId match {
+ case Some(controller) => (Some(controller), false)
+ case None => (snapshot.controllerId, true)
+ }
+ }
+
+ override def getControllerIdForExternalClient: Option[Int] = {
Review Comment:
Can you add a javadoc here explaining the behavior (i.e., random alive 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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1048902572
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -355,6 +375,13 @@ class ZkMetadataCache(brokerId: Int, metadataVersion: MetadataVersion, brokerFea
case id if id < 0 => None
case id => Some(id)
}
+ val kraftControllerIdOpt = updateMetadataRequest.kraftControllerId() match {
+ case id if id < 0 => None
+ case id => Some(id)
+ }
+ if (controllerIdOpt.nonEmpty && kraftControllerIdOpt.nonEmpty) {
Review Comment:
Removed this logic in favor of new change to the RPC message.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050026225
##########
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:
I think you're looking at an older patch.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050065115
##########
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:
This is wrong. In Kraft brokers and Zk brokers with the KRaft controller, it always returns controller information for any broker node query.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050049320
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -364,11 +399,13 @@ class BrokerToControllerRequestThread(
}
override def doWork(): Unit = {
Review Comment:
It is still the same code. Instead of calling in the `else` case, I'm just calling as soon as I get into the function. As long as the controller type is not changing, the function `maybeResetNetworkClient` does nothing.
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046447035
##########
clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java:
##########
@@ -45,14 +45,23 @@ public static class Builder extends AbstractControlRequest.Builder<StopReplicaRe
public Builder(short version, int controllerId, int controllerEpoch, long brokerEpoch,
boolean deletePartitions, List<StopReplicaTopicState> topicStates) {
- super(ApiKeys.STOP_REPLICA, version, controllerId, controllerEpoch, brokerEpoch);
+ super(ApiKeys.STOP_REPLICA, version, controllerId, controllerEpoch, brokerEpoch, false);
Review Comment:
this is another case where we should define this in terms of the other constructor to avoid duplication
```
this(version, controllerId, controllerEpoch, brokerEpoch, deletePartitions, topicStates, false)
```
--
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
[GitHub] [kafka] cmccabe commented on pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on PR #12961:
URL: https://github.com/apache/kafka/pull/12961#issuecomment-1341714694
I think the better approach to take here would be to make the `ControllerNodeProvider` return a tuple of `(isKraft, controller id)` and then have it check the metadata cache to see if it needs to use a kraft controller.
Then have the inter broker send thread do something useful if it sees a transition. I agree that we might want to use an external thread for this (or possibly the scheduler) but that should just be for reinitialization
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050026488
##########
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:
Older patch.
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046447468
##########
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:
this already has a default value of -1. It doesn't need to be stated twice...
##########
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:
this already has a default value of -1. It doesn't need to be stated twice...
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046450807
##########
core/src/main/scala/kafka/server/KafkaServer.scala:
##########
@@ -262,8 +263,11 @@ class KafkaServer(
_brokerState = BrokerState.RECOVERY
logManager.startup(zkClient.getAllTopicsInCluster())
- metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion, brokerFeatures)
- val controllerNodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache)
+ if (config.migrationEnabled) {
+ kraftControllerNodes = RaftConfig.voterConnectionsToNodes(RaftConfig.parseVoterConnections(config.quorumVoters)).asScala
+ }
+ metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion, brokerFeatures, kraftControllerNodes)
Review Comment:
can we put each parameter on its own line?
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046450254
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1665,6 +1665,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
def usesSelfManagedQuorum: Boolean = processRoles.nonEmpty
val migrationEnabled: Boolean = getBoolean(KafkaConfig.MigrationEnabledProp)
+ def enableZkApiForwarding: Boolean = migrationEnabled && interBrokerProtocolVersion.isApiForwardingEnabled
Review Comment:
to clarify, it's fine to do this static check for now but it should be done in `Kafka.scala` and we should comment that it's OK because IBP is static in ZK mode
--
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
[GitHub] [kafka] dengziming commented on pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
dengziming commented on PR #12961:
URL: https://github.com/apache/kafka/pull/12961#issuecomment-1340958183
We can close #11042 and #11281 after merging this, but we can take a look to see whether we can port some useful 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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1042536606
##########
clients/src/main/java/org/apache/kafka/common/requests/AbstractControlRequest.java:
##########
@@ -43,6 +49,8 @@ protected AbstractControlRequest(ApiKeys api, short version) {
public abstract int controllerId();
+ public abstract int kraftControllerId();
Review Comment:
This field is used currently in handling `UpdateMetadataRequest` in `ZkMetadataCache#updateMetadata`
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -355,6 +392,13 @@ class ZkMetadataCache(brokerId: Int, metadataVersion: MetadataVersion, brokerFea
case id if id < 0 => None
case id => Some(id)
}
+ val kraftControllerIdOpt = updateMetadataRequest.kraftControllerId() match {
Review Comment:
@dengziming is used here. But will be used in more places as we get more migration-related code.
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -315,7 +324,35 @@ class ZkMetadataCache(brokerId: Int, metadataVersion: MetadataVersion, brokerFea
}.getOrElse(Map.empty[Int, Node])
}
- def getControllerId: Option[Int] = metadataSnapshot.controllerId
+ def getControllerId: Option[Int] = {
+ val snapshot = metadataSnapshot
+ snapshot.controllerId.orElse(snapshot.kraftControllerId)
+ }
+
+ /**
+ * Return controller id. Additionally returns if the controller is zk or kraft controller.
+ * Returns true for zk controller, false otherwise.
+ * @return
+ */
+ def getZkOrKRaftControllerId: (Option[Int], Boolean) = {
+ val snapshot = metadataSnapshot
+ snapshot.kraftControllerId match {
+ case Some(controller) => (Some(controller), false)
+ case None => (snapshot.controllerId, true)
+ }
+ }
+
+ override def getControllerIdForExternalClient: Option[Int] = {
Review Comment:
Added to the `MetadataCache` trait.
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -164,25 +182,54 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
- private val requestThread = newRequestThread
+ private var isZkControllerThread = true
Review Comment:
I thought about it. It is just getting accessed in the `KafkaScheduler` thread. So, I thought it is not needed. But as a caution, I can add it.
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -164,25 +182,54 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
- private val requestThread = newRequestThread
+ private var isZkControllerThread = true
+ @volatile private var requestThread = newRequestThread
def start(): Unit = {
requestThread.start()
+ // If migration is enabled for zkBroker, then we might see controller change from zk to kraft
+ // and vice-versa. This periodic task takes care of setting the right channel when such
+ // controller change is noticed.
+ if (config.migrationEnabled && config.requiresZookeeper) {
+ kafkaScheduler.schedule("maybeReinitializeRequestThread",
+ () => maybeReinitializeRequestThread(), 0L, 500, TimeUnit.MILLISECONDS)
+ }
}
def shutdown(): Unit = {
requestThread.shutdown()
info(s"Broker to controller channel manager for $channelName shutdown")
}
+ def maybeReinitializeRequestThread(): Unit = {
+ val controllerInfo = controllerNodeProvider.getControllerInfo()
+ if (isZkControllerThread == controllerInfo.isZkController)
+ return
+ val oldRequestThread = requestThread
+ // Close the old thread and start a new one if old one is not already in the process of
+ // shutting down.
+ if (oldRequestThread.isRunning) {
Review Comment:
Please let me know if this looks more clear to you.
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -164,25 +182,54 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
- private val requestThread = newRequestThread
+ private var isZkControllerThread = true
+ @volatile private var requestThread = newRequestThread
def start(): Unit = {
requestThread.start()
+ // If migration is enabled for zkBroker, then we might see controller change from zk to kraft
+ // and vice-versa. This periodic task takes care of setting the right channel when such
+ // controller change is noticed.
+ if (config.migrationEnabled && config.requiresZookeeper) {
+ kafkaScheduler.schedule("maybeReinitializeRequestThread",
+ () => maybeReinitializeRequestThread(), 0L, 500, TimeUnit.MILLISECONDS)
+ }
}
def shutdown(): Unit = {
requestThread.shutdown()
info(s"Broker to controller channel manager for $channelName shutdown")
}
+ def maybeReinitializeRequestThread(): Unit = {
+ val controllerInfo = controllerNodeProvider.getControllerInfo()
+ if (isZkControllerThread == controllerInfo.isZkController)
+ return
+ val oldRequestThread = requestThread
+ // Close the old thread and start a new one if old one is not already in the process of
+ // shutting down.
+ if (oldRequestThread.isRunning) {
Review Comment:
It basically is a check for seeing if `shutdown` ran first. While thinking about this I realized, we might just keep running the scheduler task indefinitely. Let me fix that.
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1042763929
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -164,25 +184,63 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
- private val requestThread = newRequestThread
+ @volatile private var isZkControllerThread = true
+ @volatile private var requestThread = newRequestThread()
+ @volatile private var shutdownStarted = false
def start(): Unit = {
requestThread.start()
+ maybeScheduleReinitializeRequestThread()
}
def shutdown(): Unit = {
requestThread.shutdown()
+ shutdownStarted = false
info(s"Broker to controller channel manager for $channelName shutdown")
}
- private[server] def newRequestThread = {
+ def maybeScheduleReinitializeRequestThread(): Unit = {
+ // If migration is enabled for zkBroker, then we might see controller change from zk to kraft
+ // and vice-versa. This periodic task takes care of setting the right channel when such
+ // controller change is noticed.
+ if (config.migrationEnabled && config.requiresZookeeper) {
Review Comment:
Shouldn't this just be something we check in the `InterBrokerSendThread` itself? Having a separate thread to do this seems very messy.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050036052
##########
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:
The name makes sense. But the caching the `ControllerInformation` doesn't give any additional benefit. In fact, it is going to be same exact condition. So, I think it is okay to be just a boolean.
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050063855
##########
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:
Returning an `Option[Node]` doesn't really solve the problem of checking the type of controller. The real problem is this: We query metadataCache 1) either for a real kraft controller (to build communication channels, for ex.) or 2) for a random broker if we have a kraft controller (to respond to metadata and describe cluster requests).
I initially had two different functions returning either real controller id or random broker id in the case of kraft controller. This will help reduce type-checking outside this class. But Coling advised against this.
--
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
[GitHub] [kafka] mumrah commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
mumrah commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050096386
##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -364,11 +399,13 @@ class BrokerToControllerRequestThread(
}
override def doWork(): Unit = {
Review Comment:
As discussed offline, by proactively checking if the controller has changed (L402), we avoid falling into the case where an authentication of version mismatch error prevents the channel manager from finding a new controller.
--
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
[GitHub] [kafka] mumrah commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
mumrah commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050078269
##########
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:
I see, ok we can revisit this later. I suppose it would be a bit confusing if `getControllerNode()` returned something other than the Node of `getControllerId()`
--
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
[GitHub] [kafka] akhileshchg commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
akhileshchg commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1050028258
##########
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 thought about this. This unnecessarily changes all the other implementations and places where InterBrokerSendThread. Let me know your thoughts.
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046452882
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -60,7 +61,8 @@ 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,
Review Comment:
can we do an
```
abc(
a
b
c
) {
}
```
thing here? Sorry, these long scala declarations are ugly otherwise...
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046447181
##########
clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java:
##########
@@ -51,7 +51,16 @@ public static class Builder extends AbstractControlRequest.Builder<UpdateMetadat
public Builder(short version, int controllerId, int controllerEpoch, long brokerEpoch,
List<UpdateMetadataPartitionState> partitionStates, List<UpdateMetadataBroker> liveBrokers,
Map<String, Uuid> topicIds) {
- super(ApiKeys.UPDATE_METADATA, version, controllerId, controllerEpoch, brokerEpoch);
+ super(ApiKeys.UPDATE_METADATA, version, controllerId, controllerEpoch, brokerEpoch, false);
Review Comment:
same, define this in terms of the other ctor
--
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
[GitHub] [kafka] mumrah commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
mumrah commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1046370590
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1317,14 +1316,23 @@ 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 match {
+ case cache: KRaftMetadataCache => cache.getControllerId.map(_.id)
+ case cache => cache.getControllerId.flatMap {
Review Comment:
Can we add the class for this case as well?
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3332,13 +3340,21 @@ class KafkaApis(val requestChannel: RequestChannel,
}
val brokers = metadataCache.getAliveBrokerNodes(request.context.listenerName)
- val controllerId = metadataCache.getControllerId.getOrElse(MetadataResponse.NO_CONTROLLER_ID)
+ val controllerId = {
+ metadataCache match {
+ case cache: KRaftMetadataCache => cache.getControllerId.map(_.id)
+ case cache => cache.getControllerId.flatMap {
Review Comment:
And here
##########
core/src/main/scala/kafka/server/metadata/ZkMetadataCache.scala:
##########
@@ -315,7 +324,18 @@ class ZkMetadataCache(brokerId: Int, metadataVersion: MetadataVersion, brokerFea
}.getOrElse(Map.empty[Int, Node])
}
- def getControllerId: Option[Int] = metadataSnapshot.controllerId
+ def getControllerId: Option[CachedControllerId] = {
+ val snapshot = metadataSnapshot
+ snapshot.kraftControllerId match {
+ case Some(id) => Some(KRaftCachedControllerId(id))
+ case None => snapshot.controllerId.map(ZkCachedControllerId)
+ }
+ }
+
+ def getRandomAliveBrokerId: Option[Int] = {
Review Comment:
errant whitespace
--
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
[GitHub] [kafka] cmccabe commented on a diff in pull request #12961: KAFKA-14446: API forwarding support from zkBrokers to the Controller
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12961:
URL: https://github.com/apache/kafka/pull/12961#discussion_r1042763145
##########
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._
+case class ControllerInformation(node: Option[Node],
+ listenerName: ListenerName,
+ securityProtocol: SecurityProtocol,
+ saslMechanism: String,
+ isZkController: Boolean)
+
trait ControllerNodeProvider {
def get(): Option[Node]
- def listenerName: ListenerName
- def securityProtocol: SecurityProtocol
- def saslMechanism: String
-}
-
-object MetadataCacheControllerNodeProvider {
- def apply(
- config: KafkaConfig,
- metadataCache: kafka.server.MetadataCache
- ): MetadataCacheControllerNodeProvider = {
- val listenerName = config.controlPlaneListenerName
- .getOrElse(config.interBrokerListenerName)
-
- val securityProtocol = config.controlPlaneSecurityProtocol
- .getOrElse(config.interBrokerSecurityProtocol)
-
- new MetadataCacheControllerNodeProvider(
- metadataCache,
- listenerName,
- securityProtocol,
- config.saslMechanismInterBrokerProtocol
- )
- }
+ def getControllerInfo(): ControllerInformation
}
class MetadataCacheControllerNodeProvider(
- val metadataCache: kafka.server.MetadataCache,
- val listenerName: ListenerName,
- val securityProtocol: SecurityProtocol,
- val saslMechanism: String
+ val metadataCache: ZkMetadataCache,
+ val config: KafkaConfig
) extends ControllerNodeProvider {
+
+ def listenerName(isZkController: Boolean): ListenerName = {
Review Comment:
the metadata cache should know whether we have a zk controller or a kraft controller, right? So we shouldn't need to pass `isZkController` in to all these functions?
--
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