KAFKA-17011: SupportedFeatures.MinVersion incorrectly blocks v0 (3.8) #16420

jolshan · 2024-06-21T20:19:53Z

For 3.8 we can't implement a new ApiVersions bump, so for now we will omit the feature in the response in order to be backwards compatible.

Without this change, when interacting with an old version of kafka, we see the following error:

java.lang.IllegalArgumentException: Expected minValue >= 1, maxValue >= 1 and maxValue >= minValue, but received minValue: 0, maxValue: 0
    at org.apache.kafka.common.feature.BaseVersionRange.<init>(BaseVersionRange.java:65)
    at org.apache.kafka.common.feature.SupportedVersionRange.<init>(SupportedVersionRange.java:32)
    at org.apache.kafka.clients.NodeApiVersions.<init>(NodeApiVersions.java:114)
    at org.apache.kafka.clients.NetworkClient.handleApiVersionsResponse(NetworkClient.java:960)
    at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:927)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:580)
    at kafka.common.InterBrokerSendThread.pollOnce(InterBrokerSendThread.scala:78)
    at kafka.common.InterBrokerSendThread.doWork(InterBrokerSendThread.scala:98)
    at org.apache.kafka.server.util.ShutdownableThread.run(ShutdownableThread.java:127)

jolshan · 2024-06-21T21:16:36Z

Ran the kraft upgrade tests and they passed with this change 👍

jolshan · 2024-06-21T22:28:10Z

I ran some group coordinator tests as well, and those are also passing.

chia7712

@jolshan thanks for this quick fix. two trivial comments are left. PTAL

chia7712 · 2024-06-24T10:00:46Z

clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java

+        Features<SupportedVersionRange> backwardsCompatibleFeatures = Features.supportedFeatures(latestSupportedFeatures.features().entrySet()
+            .stream().filter(entry -> {
+                SupportedVersionRange supportedVersionRange = entry.getValue();
+                return supportedVersionRange.min() != 0;


not sure whether I fail to catch your description.

so for now we will change 0 to 1 in the response in order to be backwards compatible.

the code looks like it gets rid of "0" instead of changing to from 0 to 1.

For another, does this PR mean Admin#describeFeatures can't see the feature "group.version" from the broker running in 3.8.0?

Hey sorry, the comment was from the old implementation, the new way is to omit as you mentioned.
I believe describeFeatures will not be able to show the version for this release.

Perhaps it's useful to add a comment on why we are filtering version 0 features.

chia7712 · 2024-06-24T12:34:55Z

BTW, there is a description in KIP-584

Each broker’s supported dictionary of {feature → version range} will be defined in the broker code. For each supported feature, the supported version range is defined by a min_version (an int64 starting always from 1) and max_version (an int64 >=1 and >= min_version).

not sure why #15671 changed the start version from 1 to 0

https://github.com/apache/kafka/pull/15671/files#r1576879728

jolshan · 2024-06-24T16:01:52Z

@chia7712 It was incorrect to set at 1 because we can not assume 0 level is supported. However, 0 level -- the feature being disabled is valid.

chia7712 · 2024-06-24T16:17:33Z

@jolshan I feel FeatureCommandTest and ApiVersionsRequestTest have some related failed tests

jolshan · 2024-06-24T17:53:31Z

Thanks @chia7712 I will take a look.

jolshan · 2024-06-25T00:04:54Z

Cleaned up the test failures. The new ones look unrelated. 👍

chia7712

LGTM

chia7712 · 2024-06-25T06:36:18Z

It was incorrect to set at 1 because we can not assume 0 level is supported. However, 0 level -- the feature being disabled is valid.

@jolshan thanks for explanation. Also, I have saw the discussion on the mail channel (https://lists.apache.org/thread/7wx3j788ztf1vl7xm9ryt8cwl3tk9lw3). Both are good resources to learn the root cause of this issue :)

chia7712 · 2024-06-25T09:53:11Z

@jolshan sorry that I have another question:

it seems the hotfix for 3.8 is to avoid propagating the min_version=0, so we remove the min_version=0 in generating ApiVersionsResponse. That is good, and for another: should we do similar guard for BrokerRegistrationRequest? Is it a potential bug that a "new" broker tries to register to "old" quorum controller which can't accept min_version=0?

jolshan · 2024-06-25T15:37:29Z

That is good, and for another: should we do similar guard for BrokerRegistrationRequest? Is it a potential bug that a "new" broker tries to register to "old" quorum controller which can't accept min_version=0?

Hmmm. I believe the handling is ok with 0. The issue is not with 0 itself. (We can handle controllers that don't have the same range of supported versions correctly.) The issue was with SupportedVersionRange which only seems to be used by NodeApiVersions.

@cmccabe correct me if I'm wrong.

jsancio · 2024-06-25T15:54:49Z

That is good, and for another: should we do similar guard for BrokerRegistrationRequest? Is it a potential bug that a "new" broker tries to register to "old" quorum controller which can't accept min_version=0?

@chia7712 the problem is specific to ApiVersionResponse. The deserializer for that RPC response doesn't allow 0 values for min supported version or max supported version. Broker registration doesn't have this problem.

jolshan · 2024-06-25T17:54:25Z

I will merge in the next hour or so unless there are further concerns.

junrao

@jolshan : Thanks for the PR. Just a minor comment. It will also be useful to include the exception caused by this issue in the jira or PR description.

junrao · 2024-06-25T17:17:43Z

clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java

+        Features<SupportedVersionRange> backwardsCompatibleFeatures = Features.supportedFeatures(latestSupportedFeatures.features().entrySet()
+            .stream().filter(entry -> {
+                SupportedVersionRange supportedVersionRange = entry.getValue();
+                return supportedVersionRange.min() != 0;


Perhaps it's useful to add a comment on why we are filtering version 0 features.

jolshan · 2024-06-25T18:44:43Z

Thanks @junrao! Done

chia7712 · 2024-06-25T18:55:19Z

@jolshan @jsancio thanks for your response. My concern was the "min version" of broker registration is used to create SupportedVersionRange. The one use case I noticed is shown below.

kafka/core/src/main/scala/kafka/cluster/Broker.scala

Line 53 in f995edc

name -> new SupportedVersionRange(range.min(), range.max())

That is used in migration, so not sure whether it will be a problem. For example, the quorum controller is running with defective SupportedVersionRange. After controller migration, one zk broker is rolling-upgrade to kraft mode with 3.8.0, and it will send BrokerRegistrationRequest with "group.version, minSupportedVersion=0". Hence, the quorum controller is possible to use the "min version=0" to build SupportedVersionRange ... ?

jolshan · 2024-06-25T23:22:26Z

@chia7712 The new code of SupportedVersionRange does support 0 version. The only issue is when a new broker sends a request to an older one.

Is it correct that in order to encounter this issue, we would need to have an migration ongoing at the same time as a version rolling upgrade? Is this typical for a kraft migration?

If there are tests that can confirm migration is safe, I'm also happy to run and double check.

cmccabe · 2024-06-25T23:39:23Z

It is not possible to do an inter.broker.protocol / MetadataVersion change during ZK-to-KRaft migration. The code has numerous checks that prevent anyone from doing this.

More importantly, ApiVersionsResponse.SupportedFeatures is not used in ZK migration, and never has been. And the QuorumController never had the buggy code which made feature ranges containing 0 fail. It's fine for the broker to register with a supported range including 0, both before and after this PR. The broker sets its features by filling out a BrokerRegistrationRequest, not by anything it does with ApiVersionsRequest.

chia7712 · 2024-06-26T04:07:10Z

@jolshan @cmccabe thanks for the confirmation. Learn a lot from you 😄

chia7712 · 2024-06-26T15:29:23Z

That may be not a common case, but the following error can be produced in migrating zk to kraft.

[2024-06-26 15:26:26,397] ERROR Encountered zk migration fault: Unhandled error in MetadataChangeEvent (org.apache.kafka.server.fault.LoggingFaultHandler)
java.lang.IllegalArgumentException: Expected minValue >= 1, maxValue >= 1 and maxValue >= minValue, but received minValue: 0, maxValue: 1
	at org.apache.kafka.common.feature.BaseVersionRange.<init>(BaseVersionRange.java:65)
	at org.apache.kafka.common.feature.SupportedVersionRange.<init>(SupportedVersionRange.java:32)
	at kafka.cluster.Broker$.$anonfun$supportedFeatures$1(Broker.scala:49)
	at scala.collection.StrictOptimizedMapOps.map(StrictOptimizedMapOps.scala:28)
	at scala.collection.StrictOptimizedMapOps.map$(StrictOptimizedMapOps.scala:27)
	at scala.collection.convert.JavaCollectionWrappers$AbstractJMapWrapper.map(JavaCollectionWrappers.scala:321)
	at kafka.cluster.Broker$.supportedFeatures(Broker.scala:48)
	at kafka.cluster.Broker$.fromBrokerRegistration(Broker.scala:58)
	at kafka.migration.MigrationControllerChannelContext.$anonfun$liveOrShuttingDownBrokers$1(MigrationControllerChannelContext.scala:78)
	at scala.collection.Iterator$$anon$9.next(Iterator.scala:584)
	at scala.collection.immutable.List.prependedAll(List.scala:153)
	at scala.collection.immutable.List$.from(List.scala:684)
	at scala.collection.immutable.List$.from(List.scala:681)
	at scala.collection.IterableFactory$Delegate.from(Factory.scala:288)
	at scala.collection.immutable.Iterable$.from(Iterable.scala:35)
	at scala.collection.immutable.Iterable$.from(Iterable.scala:32)
	at scala.collection.IterableFactory$Delegate.from(Factory.scala:288)
	at scala.collection.IterableOps.map(Iterable.scala:682)
	at scala.collection.IterableOps.map$(Iterable.scala:682)
	at scala.collection.AbstractIterable.map(Iterable.scala:933)
	at kafka.migration.MigrationControllerChannelContext.<init>(MigrationControllerChannelContext.scala:77)
	at kafka.migration.MigrationPropagator.metadataProvider(MigrationPropagator.scala:78)
	at kafka.migration.MigrationPropagator.$anonfun$requestBatch$1(MigrationPropagator.scala:71)
	at kafka.controller.AbstractControllerBrokerRequestBatch.newBatch(ControllerChannelManager.scala:399)
	at kafka.migration.MigrationPropagator.sendRPCsToBrokersFromMetadataDelta(MigrationPropagator.scala:117)
	at org.apache.kafka.metadata.migration.KRaftMigrationDriver$MetadataChangeEvent.run(KRaftMigrationDriver.java:559)
	at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:127)
	at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:210)
	at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:181)
	at java.base/java.lang.Thread.run(Thread.java:1583)

steps:

create zk broker cluster with this patch (3.8 branch)
create quorum controller with 3.6.2
migrate zk broker to kraft with IBP=3.6 and unstable.feature (in order to enable group.version to produce min=0)

chia7712 · 2024-06-26T15:33:52Z

That may be not a common case

Producing the error needs to enable unstable.feature, so I feel that is not a common use case.

jolshan · 2024-06-26T16:12:51Z

migrate zk broker to kraft with IBP=3.6 and unstable.feature

I see that the supported range is 0-1 in that case. Is it really the case you need to set that flag?

chia7712 · 2024-06-26T16:50:19Z

just to clarify: this is not a real case I have encountered. I just be over-engineering today 😄

Is it really the case you need to set that flag?

IBP=3.6 -> avoid UNSUPPORTED_VERSION
unstable.feature.versions.enable=true -> make kraft broker send broker registration with group.version=1

cmccabe · 2024-06-26T20:29:05Z

unstable feature versions should never be enabled outside JUnit tests (I have a KIP which formalized this). Exactly to avoid this kind of confusion.

cmccabe

LGTM

chia7712 · 2024-06-26T21:00:39Z

unstable feature versions should never be enabled outside JUnit tests (I have a KIP which formalized this). Exactly to avoid this kind of confusion.

Sorry that I should not use unstable.feature.versions.enable which can mislead this discussion. The broker registration encounters the issue of BaseVersionRange (min version=0) even though unstable.feature.versions.enable is not defined (see following error stack).

[2024-06-26 20:54:40,169] ERROR Encountered zk migration fault: Unhandled error in MetadataChangeEvent (org.apache.kafka.server.fault.LoggingFaultHandler)
java.lang.IllegalArgumentException: Expected minValue >= 1, maxValue >= 1 and maxValue >= minValue, but received minValue: 0, maxValue: 0
	at org.apache.kafka.common.feature.BaseVersionRange.<init>(BaseVersionRange.java:65)
	at org.apache.kafka.common.feature.SupportedVersionRange.<init>(SupportedVersionRange.java:32)
	at kafka.cluster.Broker$.$anonfun$supportedFeatures$1(Broker.scala:49)
	at scala.collection.StrictOptimizedMapOps.map(StrictOptimizedMapOps.scala:28)
	at scala.collection.StrictOptimizedMapOps.map$(StrictOptimizedMapOps.scala:27)
	at scala.collection.convert.JavaCollectionWrappers$AbstractJMapWrapper.map(JavaCollectionWrappers.scala:321)
	at kafka.cluster.Broker$.supportedFeatures(Broker.scala:48)
	at kafka.cluster.Broker$.fromBrokerRegistration(Broker.scala:58)
	at kafka.migration.MigrationControllerChannelContext.$anonfun$liveOrShuttingDownBrokers$1(MigrationControllerChannelContext.scala:78)
	at scala.collection.Iterator$$anon$9.next(Iterator.scala:584)
	at scala.collection.immutable.List.prependedAll(List.scala:153)
	at scala.collection.immutable.List$.from(List.scala:684)
	at scala.collection.immutable.List$.from(List.scala:681)
	at scala.collection.IterableFactory$Delegate.from(Factory.scala:288)
	at scala.collection.immutable.Iterable$.from(Iterable.scala:35)
	at scala.collection.immutable.Iterable$.from(Iterable.scala:32)
	at scala.collection.IterableFactory$Delegate.from(Factory.scala:288)
	at scala.collection.IterableOps.map(Iterable.scala:682)
	at scala.collection.IterableOps.map$(Iterable.scala:682)
	at scala.collection.AbstractIterable.map(Iterable.scala:933)
	at kafka.migration.MigrationControllerChannelContext.<init>(MigrationControllerChannelContext.scala:77)
	at kafka.migration.MigrationPropagator.metadataProvider(MigrationPropagator.scala:78)
	at kafka.migration.MigrationPropagator.$anonfun$requestBatch$1(MigrationPropagator.scala:71)
	at kafka.controller.AbstractControllerBrokerRequestBatch.newBatch(ControllerChannelManager.scala:399)
	at kafka.migration.MigrationPropagator.sendRPCsToBrokersFromMetadataDelta(MigrationPropagator.scala:117)
	at org.apache.kafka.metadata.migration.KRaftMigrationDriver$MetadataChangeEvent.run(KRaftMigrationDriver.java:559)
	at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:127)
	at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:210)
	at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:181)
	at java.base/java.lang.Thread.run(Thread.java:1583)

jolshan · 2024-06-26T21:27:32Z

@chia7712 Yeah, that's what I meant. I'm considering some options to fix this so that we don't see the issue during migration. I won't merge until we come to a good solution. We may just have to revert the group.version change for this release. I need to confirm with @dajac

cmccabe · 2024-06-26T22:17:02Z

@chia7712 @jolshan : Yes, after some offline discussion I now understand what you were trying to say. I guess the bad case is:

new broker registers with old controller.
old controller attempts to send out registration from #1 to brokers still in zk mode.
serde path crashes because of 0-1 supported range in the broker's registration.

Just as with ApiVersionsResponse, we can probably fix this with a new RPC version for BrokerRegistrationRequest. I will add that to my 3.9 PR.

dajac · 2024-06-27T07:07:24Z

@chia7712 Yeah, that's what I meant. I'm considering some options to fix this so that we don't see the issue during migration. I won't merge until we come to a good solution. We may just have to revert the group.version change for this release. I need to confirm with @dajac

Reverting the group.version feature flag seems to be the best option and we can live with going back to the static config for the preview in 3.8. I wonder if we should also revert it in trunk in order to not have it in 3.9. Then, we can bring it back for 4.0 when we GA the feature.

dajac · 2024-06-27T07:10:24Z

unstable feature versions should never be enabled outside JUnit tests (I have a KIP which formalized this). Exactly to avoid this kind of confusion.

I think that it would be nice to have a public config to run the broker in experimental mode in order to test non-production features in development clusters. Otherwise, I suppose that we can achieve this by manually upgrading the MV to an unreleased version with kafka-features or do we disallow setting a non-production MV?

dajac · 2024-06-27T15:15:54Z

Discussed offline with a few folks and reverting group.version seems to be the way to go. I will prepare the PR as soon as possible. cc @jlprat

jlprat · 2024-06-27T16:20:36Z

Thanks @dajac !

jolshan · 2024-06-27T16:35:54Z

I will close this PR.

jolshan added 2 commits June 21, 2024 13:16

Do not return 0 in apiversons response

1cd3967

Add unit test

38a8b07

Suppress 0 versions for older versions

6b897b4

jlprat added the Blocker This pull request is identified as solving a blocker for a release. label Jun 24, 2024

chia7712 reviewed Jun 24, 2024

View reviewed changes

test fixes

58a9cfc

chia7712 approved these changes Jun 25, 2024

View reviewed changes

junrao reviewed Jun 25, 2024

View reviewed changes

Add comment

1faf2ab

cmccabe approved these changes Jun 26, 2024

View reviewed changes

jolshan closed this Jun 27, 2024

KAFKA-17011: SupportedFeatures.MinVersion incorrectly blocks v0 (3.8) #16420

KAFKA-17011: SupportedFeatures.MinVersion incorrectly blocks v0 (3.8) #16420

Conversation

jolshan commented Jun 21, 2024 • edited Loading

jolshan commented Jun 21, 2024

jolshan commented Jun 21, 2024

chia7712 left a comment

Choose a reason for hiding this comment

chia7712 Jun 24, 2024

Choose a reason for hiding this comment

chia7712 Jun 24, 2024

Choose a reason for hiding this comment

jolshan Jun 24, 2024

Choose a reason for hiding this comment

junrao Jun 25, 2024

Choose a reason for hiding this comment

chia7712 commented Jun 24, 2024

jolshan commented Jun 24, 2024 • edited Loading

chia7712 commented Jun 24, 2024

jolshan commented Jun 24, 2024

jolshan commented Jun 25, 2024

chia7712 left a comment

Choose a reason for hiding this comment

chia7712 commented Jun 25, 2024

chia7712 commented Jun 25, 2024

jolshan commented Jun 25, 2024

jsancio commented Jun 25, 2024

jolshan commented Jun 25, 2024

junrao left a comment

Choose a reason for hiding this comment

junrao Jun 25, 2024

Choose a reason for hiding this comment

jolshan commented Jun 25, 2024

chia7712 commented Jun 25, 2024

jolshan commented Jun 25, 2024

cmccabe commented Jun 25, 2024 • edited Loading

chia7712 commented Jun 26, 2024

chia7712 commented Jun 26, 2024

chia7712 commented Jun 26, 2024

jolshan commented Jun 26, 2024

chia7712 commented Jun 26, 2024 • edited Loading

cmccabe commented Jun 26, 2024

cmccabe left a comment

Choose a reason for hiding this comment

chia7712 commented Jun 26, 2024

jolshan commented Jun 26, 2024 • edited Loading

cmccabe commented Jun 26, 2024 • edited Loading

dajac commented Jun 27, 2024

dajac commented Jun 27, 2024

dajac commented Jun 27, 2024 • edited Loading

jlprat commented Jun 27, 2024

jolshan commented Jun 27, 2024

jolshan commented Jun 21, 2024 •

edited

Loading

jolshan commented Jun 24, 2024 •

edited

Loading

cmccabe commented Jun 25, 2024 •

edited

Loading

chia7712 commented Jun 26, 2024 •

edited

Loading

jolshan commented Jun 26, 2024 •

edited

Loading

cmccabe commented Jun 26, 2024 •

edited

Loading

dajac commented Jun 27, 2024 •

edited

Loading