KAFKA-17331: Throw UnsupportedVersionException if the data in ListOffsetRequest does NOT fit EarliestLocalSpec and LatestTieredSpec. by frankvicky · Pull Request #16876 · apache/kafka

frankvicky · 2024-08-14T03:13:51Z

JIRA: KAFKA-17331

Add the version check to client side when building ListOffsetRequest for the specific timestamp:

the version must be >=8 if timestamp=-4L (EARLIEST_LOCAL_TIMESTAMP)
the version must be >=9 if timestamp=-5L (LATEST_TIERED_TIMESTAMP)

Committer Checklist (excluded from commit message)

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

chia7712

@frankvicky thanks for this patch

frankvicky · 2024-08-14T12:37:01Z

Hi @chia7712
I have addressed the comments, PTAL

FrankYang0529

Thanks for the PR. Leave one comment.

FrankYang0529 · 2024-08-15T14:24:40Z

        public ListOffsetsRequest build(short version) {
+            for (ListOffsetsTopic topic : data.topics()) {
+                for (ListOffsetsPartition partition : topic.partitions()) {
+                    checkVersion(version, topic.name(), partition);


The failed case ListOffsetsRequestTest#testResponseDefaultOffsetAndLeaderEpochForAllVersions can be reproduced on my laptop. Could you take a look? Thank you.

Thank you, I will take a look!

@frankvicky Could you please add a helper to build invalid request in order to run test?

Hi @FrankYang0529 and @chia7712
I have reviewed the code. Since this error is due to a version check in the build method, I will change assertEqual to assertThrows when the test tries to build an invalid request to align with the current behavior.

@frankvicky What we want to protect is the request from clients (admin and consumer). So maybe we should revise fromConsumer[0] rather than build.

WDYT?

[0]

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

Line 65 in 21dd5cd

public static Builder forConsumer(boolean requireTimestamp,

The benefit is that we can leverage the API_VERSION check of network layer. The disadvantage is that it requires the API_VERSION, and fortunately that is existent in consumer and admin.

Hi @chia7712
I have reviewed the code. Since forConsumer is a static method, if we aim to perform a version check in this method, we will need to add a new argument. I will try to implement it.

chia7712 · 2024-08-16T05:02:20Z

                                          boolean requireMaxTimestamp,
-                                          boolean requireTieredStorageTimestamp) {
+                                          boolean requireTieredStorageTimestamp,
+                                          List<ListOffsetsTopic> topics) {


Could you please follow the pattern? You can add a flag to pick up the min version. Current implementation does not consider the version=8 and I feel that is a kind of bug. @FrankYang0529 WDYT?

Hi @chia7712
I have add a requireEarliestLocalTimestamp parameter to forConsumer method, PTAL

Yes, we should update this. Thanks @chia7712 and @frankvicky

chia7712

@frankvicky thanks for this patch

chia7712 · 2024-08-17T15:32:20Z

-            return new Builder(minVersion, ApiKeys.LIST_OFFSETS.latestVersion(), CONSUMER_REPLICA_ID, isolationLevel);
+
+            short version = ApiKeys.LIST_OFFSETS.latestVersion();
+            topics.forEach(topic -> topic.partitions().forEach(partition -> checkVersion(version, topic.name(), partition)));


After setting the min version, the version checkcan handled by NetworkClient [0]. Hence, we don't need to add extra check here.

[0] https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L528

chia7712 · 2024-08-18T17:49:50Z

@frankvicky I feel the refactor is a bit overkill. Maybe we can add following two methods:

        public static Builder forConsumer(boolean requireTimestamp,
                                          IsolationLevel isolationLevel) {
            return forConsumer(requireTimestamp, isolationLevel, false, false, false);
        }

        public static Builder forConsumer(boolean requireTimestamp,
                                          IsolationLevel isolationLevel,
                                          boolean requireMaxTimestamp,
                                          boolean requireEarliestTimestamp,
                                          boolean requireTieredStorageTimestamp) {
            short minVersion = 0;
            if (requireTieredStorageTimestamp)
                minVersion = 9;
            else if (requireEarliestTimestamp)
                minVersion = 8;
            else if (requireMaxTimestamp)
                minVersion = 7;
            else if (isolationLevel == IsolationLevel.READ_COMMITTED)
                minVersion = 2;
            else if (requireTimestamp)
                minVersion = 1;
            return new Builder(minVersion, ApiKeys.LIST_OFFSETS.latestVersion(), CONSUMER_REPLICA_ID, isolationLevel);
        }

Those two methods should cover all use cases for now.

frankvicky · 2024-08-19T03:10:39Z

Hi @chia7712
I have refactor the forConsumer method, PTAL

chia7712

@frankvicky thanks for this patch. a couple of comments are left.

chia7712 · 2024-08-19T03:37:25Z

-        ListOffsetsRequest.Builder builder = ListOffsetsRequest.Builder
-                .forConsumer(requireTimestamps, isolationLevel, false, false)
-                .setTargetTimes(ListOffsetsRequest.toListOffsetsTopics(targetTimes));
+        List<ListOffsetsRequestData.ListOffsetsTopic> topics = ListOffsetsRequest.toListOffsetsTopics(targetTimes);


If there is no strong reason, could we keep fluent pattern?

ListOffsetsRequest.Builder builder = ListOffsetsRequest.Builder .forConsumer(requireTimestamps, isolationLevel) .setTargetTimes(ListOffsetsRequest.toListOffsetsTopics(targetTimes));

chia7712 · 2024-08-19T03:37:42Z

                    .setTargetTimes(topics)
                    .build(version);
            ListOffsetsResponse response = (ListOffsetsResponse) request.getErrorResponse(0, Errors.NOT_LEADER_OR_FOLLOWER.exception());
-    


please revert this unrelated change

chia7712 · 2024-08-19T03:37:53Z

-                    .forConsumer(false, IsolationLevel.READ_UNCOMMITTED, false, false)
-                    .setTargetTimes(singletonList(topic))
+                    .forConsumer(false, IsolationLevel.READ_UNCOMMITTED)
+                    .setTargetTimes(topics)


chia7712 · 2024-08-19T03:37:58Z

-                    .forConsumer(true, IsolationLevel.READ_UNCOMMITTED, false, false)
-                    .setTargetTimes(singletonList(topic))
+                    .forConsumer(true, IsolationLevel.READ_UNCOMMITTED)
+                    .setTargetTimes(topics)


chia7712 · 2024-08-19T03:38:01Z

-                    .forConsumer(true, IsolationLevel.READ_COMMITTED, false, false)
-                    .setTargetTimes(singletonList(topic))
+                    .forConsumer(true, IsolationLevel.READ_COMMITTED)
+                    .setTargetTimes(topics)


chia7712 · 2024-08-19T03:38:04Z

-          .setCurrentLeaderEpoch(27)).asJava)).asJava
-      ).
-      build()
+    val topics = List(new ListOffsetsTopic()


chia7712 · 2024-08-19T03:38:17Z

    val topicPartition = new TopicPartition("foo", 0)
-    val request = ListOffsetsRequest.Builder.forConsumer(false, IsolationLevel.READ_UNCOMMITTED, false, false)
-      .setTargetTimes(buildTargetTimes(topicPartition, ListOffsetsRequest.LATEST_TIMESTAMP, 10).asJava).build(0)
+    val targetTimes = buildTargetTimes(topicPartition, ListOffsetsRequest.LATEST_TIMESTAMP, 10).asJava


chia7712 · 2024-08-19T03:38:20Z

              .setCurrentLeaderEpoch(15)).asJava)
-          ListOffsetsRequest.Builder.forConsumer(false, IsolationLevel.READ_UNCOMMITTED, false, false)
-            .setTargetTimes(List(topic).asJava)
+          val targetTimes = List(topic).asJava


chia7712 · 2024-08-19T03:38:30Z

        public String toString() {
            return data.toString();
        }
+


chia7712

LGTM

chia7712 · 2024-08-20T01:06:31Z

            short minVersion = 0;
            if (requireTieredStorageTimestamp)
                minVersion = 9;
+            else if (requireEarliestTimestamp)


@FrankYang0529 Could you please take a look?

Hi @chia7712 and @frankvicky, What does requireEarliestTimestamp mean? EARLIEST_TIMESTAMP or EARLIEST_LOCAL_TIMESTAMP? If it's EARLIEST_LOCAL_TIMESTAMP, probably we should change the parameter name as requireEarliestLocalTimestamp, so users don't misunderstand it.

Sure, I will update it

chia7712 · 2024-08-20T19:39:56Z

@FrankYang0529 Could you please take a look?

FrankYang0529

LGTM. Thanks for the fix.

…dSpec (#16876) Add the version check to client side when building ListOffsetRequest for the specific timestamp: 1) the version must be >=8 if timestamp=-4L (EARLIEST_LOCAL_TIMESTAMP) 2) the version must be >=9 if timestamp=-5L (LATEST_TIERED_TIMESTAMP) Reviewers: PoAn Yang <payang@apache.org>, Chia-Ping Tsai <chia7712@gmail.com>

…dSpec (apache#16876) Add the version check to client side when building ListOffsetRequest for the specific timestamp: 1) the version must be >=8 if timestamp=-4L (EARLIEST_LOCAL_TIMESTAMP) 2) the version must be >=9 if timestamp=-5L (LATEST_TIERED_TIMESTAMP) Reviewers: PoAn Yang <payang@apache.org>, Chia-Ping Tsai <chia7712@gmail.com>

KAFKA-17331: Add version check in ListOffsetsRequest

c9c1c85

chia7712 mentioned this pull request Aug 14, 2024

KAFKA-17331: Throw unsupported version exception if the server does NOT support EarliestLocalSpec and LatestTieredSpec #16873

Merged

3 tasks

chia7712 reviewed Aug 14, 2024

View reviewed changes

KAFKA-17331: Address the comments

b446cca

FrankYang0529 requested changes Aug 15, 2024

View reviewed changes

frankvicky added 2 commits August 15, 2024 23:40

KAFKA-17331: Align assertion with the current behavior

81e18ef

KAFKA-17331: Move version check to forConsumer

ab74669

chia7712 reviewed Aug 16, 2024

View reviewed changes

KAFKA-17331: Add requireEarliestLocalTimestamp to forConsumer method

535f7ee

chia7712 reviewed Aug 17, 2024

View reviewed changes

frankvicky added 2 commits August 18, 2024 15:18

KAFKA-17331: refactor forConsumer

a4c268d

KAFKA-17331: Add unit test

709942b

chia7712 reviewed Aug 18, 2024

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java

KAFKA-17331: refactor

4d6e863

chia7712 reviewed Aug 19, 2024

View reviewed changes

frankvicky added 4 commits August 19, 2024 11:54

KAFKA-17331: revert unnecessary change

b3094e1

KAFKA-17331: revert unnecessary change

e7b50c0

KAFKA-17331: revert unnecessary change

3d97398

KAFKA-17331: revert unnecessary change

cc9533f

chia7712 approved these changes Aug 19, 2024

View reviewed changes

chia7712 reviewed Aug 20, 2024

View reviewed changes

KAFKA-17331: rename parameter and fix ut

d1158f1

FrankYang0529 approved these changes Aug 21, 2024

View reviewed changes

chia7712 merged commit d67c18b into apache:trunk Aug 25, 2024

chia7712 mentioned this pull request Oct 3, 2024

KAFKA-15859: Fixed the Unsupported version error when new admin connects to old broker #17358

Merged

3 tasks

Conversation

frankvicky commented Aug 14, 2024

Committer Checklist (excluded from commit message)

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

frankvicky commented Aug 14, 2024

Uh oh!

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

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!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

chia7712 commented Aug 18, 2024

Uh oh!

frankvicky commented Aug 19, 2024

Uh oh!

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

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!

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

chia7712 commented Aug 20, 2024

Uh oh!

FrankYang0529 left a comment

Choose a reason for hiding this comment