KAFKA-13935 Fix static usages of IBP in KRaft mode by mumrah · Pull Request #12250 · apache/kafka

mumrah · 2022-06-03T22:24:11Z

Set the minimum supported MetadataVersion to 3.0-IV1
Remove MetadataVersion.UNINITIALIZED
Relocate RPC version mapping for fetch protocols into MetadataVersion
Replace static IBP calls with dynamic calls to MetadataCache

A side effect of removing the UNINITIALIZED metadata version is that the FeatureControlManager and FeatureImage will initialize themselves with the minimum KRaft version (3.0-IV1).

The rationale for setting the minimum version to 3.0-IV1 is so that we can avoid any cases of KRaft mode running with an old log message format (KIP-724 was introduced in 3.0-IV1). As a side-effect of increasing this minimum version, the feature level values decreased by one.

…bp-broker-fixup

junrao

@mumrah : Thanks for the PR. One quick comment below.

junrao · 2022-06-06T21:35:04Z

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

-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_11_0_IV0)) 2
-    else if (interBrokerProtocolVersion.isAtLeast(IBP_0_10_1_IV2)) 1
-    else 0
+  val interBrokerProtocolVersion = if (processRoles.isEmpty) {


There are still a few places referencing this static value (Partition, GroupCoordinator, etc). Should we change them to use MetadataVersion?

Yes we should fix those as well. I believe all the remaining usages are for versions prior to 3.0-IV1, so in KRaft mode we can pass a static version 3.0-IV1 (regardless of the actual MetadataVersion). Setting the IBP statically to the minimum KRaft compatible version (when in KRaft mode) also guards against any incorrect/inadvertent future usages of config.interBrokerProtocolVersion.

Basically, the PR grew pretty large/complex when I was trying to include the remaining conversions. WDYT about completing those in a separate PR?

@mumrah : Thanks for the explanation. Covering that in a separate PR sounds good to me.

…bp-broker-fixup

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

…er-fixup

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

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

server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java

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

hachikuji · 2022-06-08T18:27:35Z

core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala

      metadataCache.setImage(newImage)

-      val metadataVersionLogMsg = newImage.features().metadataVersion() match {
-        case MetadataVersion.UNINITIALIZED => "un-initialized metadata.version"


Does broker startup have any guarantees with respect to the metadata version in the log? For example, does the controller require the latest metadata feature record to be consumed before a broker can be unfenced? I'm mainly trying to understand if we might revert to an older version and what the consequences of that would be.

The broker remains fenced until it catches up on the metadata log (up to the high watermark, I believe). The broker components that care about metadata (which is most of them) have some "initialize" or "startup" method called when we first publish metadata on a broker. This should mean that when a broker component is fully started, it will have the current metadata version of the cluster.

If a metadata version upgrade happened while a broker was initializing, I think it would either be included in the initial MetadataImage, or it would be handled separately. Either way, components which use the metadata version need to cope with the version changing at runtime.

Fair enough. I guess I was considering if we need any protection from an uninitialized metadata version in the code. We had the UNINITIALIZED sentinel before, but now it looks like we would use the min KRaft version or IBP. I don't know if there are any consequences to relying on an old version. Writes to the metadata log will go through the controller. Replication is protected by fencing. How about forwarding?

The IBP is just used for bootstrapping a metadata version when upgrading from an older KRaft cluster (see my other comment). Broker and controller components will assume the minimum KRaft version until we finish loading the metadata log.

Forwarding should be okay since we don't start accepting external connections until after we've caught up on metadata in the broker.

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

Lines 450 to 452 in f79528f

// Enable inbound TCP connections. Each endpoint will be started only once its matching

// authorizer future is completed.

socketServer.enableRequestProcessing(authorizerFutures)

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

hachikuji · 2022-06-09T04:59:02Z

core/src/main/scala/kafka/tools/StorageTool.scala

    formatParser.addArgument("--release-version", "-r").
      action(store()).
-      help(s"A release version to use for the initial metadata.version. The default is (${MetadataVersion.latest().version()})")
+      help(s"A release version (e.g., 3.2, 3.3) to use for the initial metadata version. The default is (${MetadataVersion.latest().version()})")


Should we mention the minimum version for KRaft? Also, do we enforce the minimum?

MetadataVersion-s have a isKRaftSupported which we check in a few places (including here in StorageTool)

hachikuji · 2022-06-09T05:03:12Z

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

+      if (!configuredVersion.isKRaftSupported) {
+        throw new ConfigException(s"A non-KRaft version ${interBrokerProtocolVersionString} given for ${KafkaConfig.InterBrokerProtocolVersionProp}")
+      } else {
+        warn(s"${KafkaConfig.InterBrokerProtocolVersionProp} is deprecated in KRaft mode as of 3.3. See kafka-storage.sh help for details.")


The message here says IBP is deprecated, but we're actually ignoring the value. Should we mention that in the message?

hachikuji · 2022-06-09T05:06:31Z

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

+    // Load the bootstrap metadata file or, in the case of an upgrade from older KRaft, bootstrap the
    // metadata.version corresponding to a user-configured IBP.
    val bootstrapMetadata = if (config.originals.containsKey(KafkaConfig.InterBrokerProtocolVersionProp)) {
      BootstrapMetadata.load(Paths.get(config.metadataLogDir), config.interBrokerProtocolVersion)


I'm a little confused on the use of the static IBP. In KafkaConfig, we ignore the value, but here we use it.

Ah, this should be reading the original config for IBP. I'll fix this.

The purpose of this is to allow a rolling upgrade from a previous KRaft version where the IBP is explicitly set. If someone is running 3.2 with IBP of 3.2 (or 3.1, 3.0), we want to keep that same compatibility level during the upgrade to 3.3+. We do this by bootstrapping the IBP supplied by the user.

If IBP is not explicitly set, users won't be able to do a rolling upgrade to 3.3 without side effects. We should probably guard against this to prevent upgraded brokers from falling back to 3.0 while un-upgraded brokers may be on a higher version.

hachikuji · 2022-06-09T05:26:26Z

core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala

      metadataCache.setImage(newImage)

-      val metadataVersionLogMsg = newImage.features().metadataVersion() match {
-        case MetadataVersion.UNINITIALIZED => "un-initialized metadata.version"


Fair enough. I guess I was considering if we need any protection from an uninitialized metadata version in the code. We had the UNINITIALIZED sentinel before, but now it looks like we would use the min KRaft version or IBP. I don't know if there are any consequences to relying on an old version. Writes to the metadata log will go through the controller. Replication is protected by fencing. How about forwarding?

core/src/main/scala/kafka/tools/StorageTool.scala

hachikuji · 2022-06-10T17:16:06Z

core/src/test/scala/unit/kafka/server/KafkaRaftServerTest.scala

  }

+  @Test
+  def testKRaftUpdateWithIBP(): Unit = {


Could we have a test case with an invalid IBP version?

If we have an invalid IBP here, KafkaConfig will throw a ConfigException right away. Basically, if we're in KRaft mode and IBP is user-defined as something less than 3.0 (or, 3.0-IV1 really), we won't even try to start up.

I'll see if I can add a test that initializes the controller with a bad metadata version.

A test case for KafkaConfig would be fine as well.

hachikuji

Thanks, LGTM

CONFLUENT: Sync from apache/kafka trunk to confluentinc/kafka master (13 Jun 2022) apache/trunk: (7 commits) KAFKA-13891: reset generation when syncgroup failed with REBALANCE_IN…(apache#12140) KAFKA-10000: Exactly-once source tasks (apache#11780) KAFKA-13436: Omitted BrokerTopicMetrics metrics in the documentation (apache#11473) MINOR: Use Exit.addShutdownHook instead of directly adding hooks to R…(apache#12283) KAFKA-13846: Adding overloaded metricOrElseCreate method (apache#12121) KAFKA-13935 Fix static usages of IBP in KRaft mode (apache#12250) HOTFIX: null check keys of ProducerRecord when computing sizeInBytes (apache#12288) Conflicts: None

mumrah added 5 commits June 3, 2022 11:39

Use MetadataCache for getting MetadataVersion in replica fetchers

e2e3ce3

WIP

dcac78c

Remove uninitialized MetadataVersion

f404e21

Merge remote-tracking branch 'apache-github/trunk' into KAFKA-13935-i…

d62d722

…bp-broker-fixup

Move RPC version mapping to MetadataVersion

49f34d0

mumrah marked this pull request as ready for review June 4, 2022 01:39

mumrah requested review from cmccabe and dajac June 4, 2022 01:40

Fix test failures

b903301

junrao reviewed Jun 7, 2022

View reviewed changes

Merge remote-tracking branch 'apache-github/trunk' into KAFKA-13935-i…

d9f398f

…bp-broker-fixup

dajac mentioned this pull request Jun 7, 2022

KAFKA-13916; Fenced replicas should not be allowed to join the ISR in KRaft (KIP-841, Part 1) #12240

Merged

3 tasks

mumrah added 3 commits June 7, 2022 12:24

fixup unit test

e9fc686

Merge remote-tracking branch 'apache-github/trunk' into KAFKA-13935-i…

9833a65

…bp-broker-fixup

Remove references to UNINITIALIZED metadata version

df839a3

hachikuji reviewed Jun 7, 2022

View reviewed changes

core/src/main/scala/kafka/server/RemoteLeaderEndPoint.scala Outdated Show resolved Hide resolved

mumrah added 3 commits June 7, 2022 20:41

fix failing test after merge from trunk

24006b6

Merge remote-tracking branch 'apache/trunk' into KAFKA-13935-ibp-brok…

eeaeb61

…er-fixup

Fix another test after updates from trunk

eaf680d

hachikuji reviewed Jun 8, 2022

View reviewed changes

mumrah added 2 commits June 8, 2022 16:13

PR feedback and additional IBP check on startup in KRaft mode

015349f

fix unit test

f79528f

hachikuji reviewed Jun 9, 2022

View reviewed changes

Protect against implicitly downgrading during rolling upgrade to 3.3+

0145d53

hachikuji reviewed Jun 10, 2022

View reviewed changes

PR feedback

aea22cc

hachikuji approved these changes Jun 13, 2022

View reviewed changes

mumrah merged commit cc38405 into apache:trunk Jun 13, 2022

	// Enable inbound TCP connections. Each endpoint will be started only once its matching
	// authorizer future is completed.
	socketServer.enableRequestProcessing(authorizerFutures)

Conversation

mumrah commented Jun 3, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

junrao 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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mumrah Jun 9, 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!

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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mumrah commented Jun 3, 2022 •

edited

Loading

mumrah Jun 9, 2022 •

edited

Loading