KAFKA-13228; Ensure ApiVersionRequest is properly handled KRaft co-resident mode by dengziming · Pull Request #11784 · apache/kafka

dengziming · 2022-02-17T09:47:41Z

More detailed description of your change

This is based on #12157

Normally, a broker connects to the active controller to learn its ApiVersions, and in co-resident KRaft mode, a broker gets the ApiVersions directly from binary code if the broker is also the activate controller.
But we only concerns ApiKeys.zkBrokerApis() when we call NodeApiVersions.create(), we should use ApiKeys.controllerApiVersions when in KRaft mode.

Summary of testing strategy (including rationale)

When I described quorum in Kraft mode I got org.apache.kafka.common.errors.UnsupportedVersionException: The broker does not support DESCRIBE_QUORUM.

After this change, the DESCRIBE_QUORUM request was property handled and got a correct response:

TopicData(topicName='__cluster_metadata', partitions=[PartitionData(partitionIndex=0, errorCode=0, leaderId=1, leaderEpoch=30, highWatermark=141, currentVoters=[ReplicaState(replicaId=1, logEndOffset=141)], observers=[])])

And I also add an integration test for this.

Committer Checklist (excluded from commit message)

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

dengziming · 2022-02-17T09:50:14Z

This is the source of the bug, we always assume currentNodeApiVersions=ApiKeys.zkBrokerApis

dengziming · 2022-02-17T09:51:40Z

I want to extend ClusterTestKit to add a new clusterType=Type.CO_KRAFT, but this is not a minor work.

jolshan · 2022-02-22T17:12:43Z

Looks like there are a few checkstyle issues in this file

jolshan · 2022-02-22T17:23:06Z

Are these changes just cleaning up? Did we not want to add an extra parameter to DefaultAlterIsrManager?

Yes, this is just a cleanup.
Currently, we have several BrokerToControllerChannel for AlterIsrManager, LifecycleManager, AutoTopicCreateManager and ForwardingManager. My cleanup here has 2 benifits:

Create and manage them all in BrokerServer or KafkaServer

Avoid creating different ControllerNodeProvider for each of them

jolshan · 2022-02-22T17:24:39Z

Unless I'm misreading the code, it seems like this was not None before. Are there implications for adding the None case?

The NodeApiVersions is used in BrokerToControllerChannelManager to get controller ApiVersion only if the current node is both the broker and the controller(which only happen in KRaft co-resident mode or zk mode), otherwise it will connect the controller and send ApiversionRequest to get it. in other cases, it is not used so I just set it to None, and it will make no difference if I change it to Some(controllerApiVersion)

hachikuji

@dengziming Thanks, this looks like a good fix. Would you mind rebasing?

…ent KRaft

hachikuji

Thanks for rebasing. Left a couple comments.

hachikuji · 2022-06-30T00:32:33Z

  def apply(
    config: KafkaConfig,
    metadataCache: MetadataCache,
+    alterPartitionChannelManager: BrokerToControllerChannelManager,


nit: I think alterPartition is redundant here. Could we call it channelManager?

hachikuji · 2022-06-30T00:37:13Z

    requestThread.activeControllerAddress().flatMap { activeController =>
      if (activeController.id == config.brokerId)
-        Some(currentNodeApiVersions)
+        currentNodeControllerApiVersions


I wonder if this optimization is worth preserving. Any harm using the same negotiation even when the controller is co-resident?

I didn't find any harm in using the same negotiation so I removed this optimization here. Yet we should notice one thing here, the NetworkClient will only send ApiversionRequest to a broker when connected to it and will not refresh it before reconnecting, this assumption is no longer absolutely right since the broker will intersect controller apiVersion and controller apiVersion would change.

If I understand correctly, you are saying that the range of supported api versions may change while a client is still connected to the broker. In general it's only a problem if the range of supported versions is reduced (e.g. in a downgrade). The way the code is supposed to handle this is by catching the unsupported version in the Envelope response and disconnecting the client in order to ensure that it reconnects and gets the latest range of api version support. Check KafkaApis.handleInvalidVersionsDuringForwarding and see if that logic makes sense. I do not know if we have testing for this scenario.

Yes, there may be a similar problem with finalized features since they will also change, I will check KafkaApis.handleInvalidVersionsDuringForwarding to inspect it.

hachikuji

Thanks, LGTM overall. Just one minor comment.

hachikuji · 2022-06-30T22:28:37Z

        // Start alter partition manager based on the IBP version
-        alterIsrManager = if (config.interBrokerProtocolVersion.isAlterPartitionSupported) {
+        alterPartitionManager = if (config.interBrokerProtocolVersion.isAlterPartitionSupported) {
+          val alterPartitionChannelManager = BrokerToControllerChannelManager(


nit: I think I favor creating the BrokerToControllerChannelManager in AlterPartitionManager.apply. The main thing is that it makes the ownership clearer. Currently DefaultAlterPartitionManager is responsible for starting and stopping the channel manager. We can revise BrokerServer to use the same apply method (I am not sure why we didn't do that).

I guess we create BrokerToControllerChannelManager in BrokerServer because we want to consolidate these channel to use only one, #10135 (comment)
but it's it's still early to consolidate channel now so I moved it to AlterPartitionManager.apply as you suggested.

Consolidation is a little difficult because of head-of-line blocking (sigh). It might be reasonable to combine the alter partition and broker lifecycle channel managers though. Anyway, this is definitely a separate patch.

hachikuji

LGTM. Thanks for the patch!

hachikuji · 2022-07-01T17:00:56Z

@dengziming Hope you don't mind, but I pushed a simple fix for an NPE.

dengziming · 2022-07-04T01:58:23Z

@dengziming Hope you don't mind, but I pushed a simple fix for an NPE.

Thanks, this fix is reasonable.👍

dengziming commented Feb 17, 2022

View reviewed changes

jolshan reviewed Feb 22, 2022

View reviewed changes

dengziming force-pushed the KAFKA-13228-kraft-api-version branch from 2b9ef1f to 69bbcbb Compare May 25, 2022 09:29

dengziming mentioned this pull request Jun 4, 2022

KAFKA-13888: Addition of Information in DescribeQuorumResponse about Voter Lag #12206

Merged

3 tasks

dengziming mentioned this pull request Jun 13, 2022

KAFKA-13228: ApiVersionRequest is not properly handled in KRaft co-resident mode #11261

Closed

hachikuji reviewed Jun 28, 2022

View reviewed changes

dengziming force-pushed the KAFKA-13228-kraft-api-version branch from aa7df09 to e972040 Compare June 29, 2022 02:33

dengziming added 3 commits June 29, 2022 17:27

Support co-resident in Test Kit

8f795a4

KAFKA-13228: Ensure ApiVersionRequest is properly handled in co-resid…

0e76ad9

…ent KRaft

Enable Kraft co-resident mode in ApiVersionTest

ac07a78

dengziming force-pushed the KAFKA-13228-kraft-api-version branch from e972040 to ac07a78 Compare June 29, 2022 09:29

hachikuji reviewed Jun 30, 2022

View reviewed changes

resolve comments

d4b9a04

hachikuji reviewed Jun 30, 2022

View reviewed changes

resolve comments

c631dae

hachikuji approved these changes Jul 1, 2022

View reviewed changes

Create controller node provider after metadata cache

58af486

hachikuji merged commit 448441a into apache:trunk Jul 5, 2022

dengziming deleted the KAFKA-13228-kraft-api-version branch November 24, 2022 06:46

Conversation

dengziming commented Feb 17, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dengziming Feb 23, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Jul 1, 2022

Uh oh!

dengziming commented Jul 4, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

dengziming commented Feb 17, 2022 •

edited

Loading

dengziming Feb 23, 2022 •

edited

Loading