Group and Txn metadata topics should be queried directly from the controller #7716

viktorsomogyi · 2019-11-19T22:19:48Z

GroupMetadataManager and TransactionStateManager should use a direct broker-to-controller channel to query the number of partitions instead of relying on Zookeeper.

This change introduces a new class that always sends the request to the active controller. In case the cached controller isn't available or not the controller it closes the connection and tries to refresh itself from the local metadataCache until it finds the active controller.

BrokerToControllerMetadataManager manages the request queue that is consumed by the request thread and also controls its lifecycle. Lazy initialization is used as the means of creating the thread so it won't try to create it before there is an actual need for it. The public methods of this class supposed to implement the high level functions that are queried by various classes (in this case GroupMetadataManager and TransactionStateManager) and return KafkaFuture so that the users of this class can work asynchronously over a blocking connection that the BrokerToControllerRequestThread implements.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

viktorsomogyi · 2019-11-19T22:23:20Z

@dhruvilshah3, @cmccabe, @ijuma I've published the broker-to-controller channel I've been working on. There are still a few things that I'd like to do (mainly on the testing side) hence the PR is a draft for now but wanted to get your opinion on this approach if it's in sync with what you're thinking of. (And if the implementation is in line with your thinking we could get into a deeper review.)

ijuma · 2019-11-20T04:00:05Z

InterBrokerSendThread was meant to be a building block for this kind of thing. Having said that, was a decision made that we should make the call to the Controller for this instead fo relying on the metadata cache? There are some issues with the latter and maybe this is the right approach as we move to a pull model, but I didn't see much discussion. cc @cmccabe @hachikuji

viktorsomogyi · 2019-11-20T10:48:49Z

Yea I was looking at InterBrokerSendThread but I thought we'd be OK with a less complex solution. For instance I think it's enough to use one queue instead of per node as handling controller failovers would be easier. Also a blocking call might be enough similarly to the controller-to-broker communication. Although if you think it would be better to do this with the InterBrokerSendThread I'm fine with rewriting that part (I don't expect it to add too much overhead).
Regarding the pull vs metadata cache decision: I was inferring this from KIP-500's "New Controller APIs" section as it says in some cases we'd need new API to replace an operation that was formerly done via ZooKeeper. It brings up ISR altering as an example but I think it will need to be applied for other protocols as well, such as broker bootstrapping and registration, log dir failure handling, producer ID management. I can write a KIP too if you think this should be discussed more elaborately.

viktorsomogyi · 2019-11-22T09:31:53Z

Had a chat with @satishd yesterday and it seems like continuing with the InterBrokerSendThread would be indeed better. Will update this PR with the change using IBST.

viktorsomogyi · 2019-12-14T09:07:06Z

retest this please

viktorsomogyi · 2020-02-27T16:56:45Z

@ijuma @cmccabe @hachikuji I rebased this and actualized it a bit. Would you please review this? Does this need more elaborate discussion such as a KIP?

viktorsomogyi · 2020-02-27T17:07:30Z

retest this please

abbccdda · 2020-02-27T17:29:34Z

If this change adds or mutates jmx metrics, I think we should do a KIP.

viktorsomogyi · 2020-02-28T16:05:36Z

@abbccdda it shouldn't modify JMX metrics

viktorsomogyi · 2020-02-28T16:10:40Z

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

   */
  private def getGroupMetadataTopicPartitionCount: Int = {
-    zkClient.getTopicPartitionCount(Topic.GROUP_METADATA_TOPIC_NAME).getOrElse(config.offsetsTopicNumPartitions)
+    controllerChannel.getPartitionCount(Topic.GROUP_METADATA_TOPIC_NAME).get


This one needs to be more robust, now it's prone to timeout related errors. Working on this.

abbccdda

Thanks for the PR, I got a high level question, which is whether the change considers the controller broker version, which means whether the targeted controller could answer the topic metadata request in all scenarios?

core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala

abbccdda · 2020-02-28T17:04:24Z

core/src/main/scala/kafka/server/KafkaServer.scala

@@ -169,6 +169,8 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
  var metadataCache: MetadataCache = null
  var quotaManagers: QuotaFactory.QuotaManagers = null

+  var controllerChannel: BrokerToControllerChannelManager = _


What's the benefit of initializing as default value here vs starting from null?

core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala

abbccdda · 2020-02-28T17:14:49Z

core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala

+        }
+      }
+    } catch {
+      case e: Exception =>


Should we be more strict here about exception handling? I don't think we shall continue in every possible exceptions, if we could brainstorm :)

Of course, let's brainstorm! :)

My main goal with this was that ideally we should catch network related exceptions when the request fails due to disconnect events. Do you have specific ideas about what to catch here?

Yea, I agree. We could just stay here as long as we are not detecting any other fatal exceptions.

Tomorrow I'll write some test cases for this for some specific scenarios.
I think though that usually we should try and handle most exceptions and reconnect to a controller when possible. On the other hand we likely don't want to catch any non-Exception Throwables as those are usually more serious cases (for instance OOM).

abbccdda · 2020-03-03T19:16:31Z

@hachikuji @cmccabe Could you also take a look?

…oControllerChannel instead of zkClient

viktorsomogyi · 2020-04-02T15:46:39Z

@hachikuji , @cmccabe I rebased my solution. Would you please look at this and review it and suggest if it's fine, needs more tests or how can we proceed with this?

viktorsomogyi · 2020-04-30T09:06:53Z

Hey folks, I'm closing this PR due to the lack of interest. If anyone interested, please feel free to pick up the related jira.

viktorsomogyi force-pushed the broker-request-channel branch from f7b4ed4 to 722fe08 Compare November 25, 2019 16:34

viktorsomogyi marked this pull request as ready for review February 27, 2020 13:53

viktorsomogyi force-pushed the broker-request-channel branch from 393a2d5 to 3d20888 Compare February 27, 2020 16:43

viktorsomogyi commented Feb 28, 2020

View reviewed changes

abbccdda reviewed Feb 28, 2020

View reviewed changes

viktorsomogyi added 7 commits April 2, 2020 11:23

Broker to controller request channel

ff1d0d4

Change GroupMetadataManager and TransactionStateManage to use BrokerT…

dc0ae29

…oControllerChannel instead of zkClient

Use InterBrokerSendThread

3d05e7a

Add license headers

4cd9b1f

Remove some unused imports

93ba030

Add missing logContext

d70e2ab

Address review comments

a9ec2bd

viktorsomogyi force-pushed the broker-request-channel branch from 3e1a304 to a9ec2bd Compare April 2, 2020 09:41

viktorsomogyi closed this Apr 30, 2020

This was referenced Jul 13, 2020

KAFKA-10270: A broker to controller channel manager #9012

Merged

(Back-up draft) redirection with version bump #9042

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Group and Txn metadata topics should be queried directly from the controller #7716

Group and Txn metadata topics should be queried directly from the controller #7716

viktorsomogyi commented Nov 19, 2019

viktorsomogyi commented Nov 19, 2019

ijuma commented Nov 20, 2019

viktorsomogyi commented Nov 20, 2019

viktorsomogyi commented Nov 22, 2019

viktorsomogyi commented Dec 14, 2019

viktorsomogyi commented Feb 27, 2020

viktorsomogyi commented Feb 27, 2020

abbccdda commented Feb 27, 2020

viktorsomogyi commented Feb 28, 2020

viktorsomogyi Feb 28, 2020

abbccdda left a comment

abbccdda Feb 28, 2020

abbccdda Feb 28, 2020

viktorsomogyi Mar 2, 2020

abbccdda Mar 3, 2020

viktorsomogyi Mar 3, 2020

abbccdda commented Mar 3, 2020

viktorsomogyi commented Apr 2, 2020

viktorsomogyi commented Apr 30, 2020

Group and Txn metadata topics should be queried directly from the controller #7716

Group and Txn metadata topics should be queried directly from the controller #7716

Conversation

viktorsomogyi commented Nov 19, 2019

Committer Checklist (excluded from commit message)

viktorsomogyi commented Nov 19, 2019

ijuma commented Nov 20, 2019

viktorsomogyi commented Nov 20, 2019

viktorsomogyi commented Nov 22, 2019

viktorsomogyi commented Dec 14, 2019

viktorsomogyi commented Feb 27, 2020

viktorsomogyi commented Feb 27, 2020

abbccdda commented Feb 27, 2020

viktorsomogyi commented Feb 28, 2020

viktorsomogyi Feb 28, 2020

Choose a reason for hiding this comment

abbccdda left a comment

Choose a reason for hiding this comment

abbccdda Feb 28, 2020

Choose a reason for hiding this comment

abbccdda Feb 28, 2020

Choose a reason for hiding this comment

viktorsomogyi Mar 2, 2020

Choose a reason for hiding this comment

abbccdda Mar 3, 2020

Choose a reason for hiding this comment

viktorsomogyi Mar 3, 2020

Choose a reason for hiding this comment

abbccdda commented Mar 3, 2020

viktorsomogyi commented Apr 2, 2020

viktorsomogyi commented Apr 30, 2020