KAFKA-16310 ListOffsets doesn't report the offset with maxTimestamp a… by chia7712 · Pull Request #15621 · apache/kafka

chia7712 · 2024-03-28T20:31:50Z

We do iterate the records to find the offsetOfMaxTimestamp instead of returning the cached one when handling ListOffsetsRequest.MAX_TIMESTAMP, since it is hard to align all paths to get correct offsetOfMaxTimestamp. The known paths are shown below.

convertAndAssignOffsetsNonCompressed -> we CAN get correct offsetOfMaxTimestamp when validating all records
assignOffsetsNonCompressed -> ditto
validateMessagesAndAssignOffsetsCompressed -> ditto
validateMessagesAndAssignOffsetsCompressed#buildRecordsAndAssignOffsets -> ditto
appendAsFollow#append#analyzeAndValidateRecords -> we CAN'T get correct offsetOfMaxTimestamp as iterating all records is expensive when fetching records from leader
LogSegment#recover -> ditto

https://issues.apache.org/jira/browse/KAFKA-16310

Committer Checklist (excluded from commit message)

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

…nymore

junrao

@chia7712 : Thanks for the PR.

We also need to
(1) revert all offsetForMaxTimestamp to shallowOffsetMaxTimestamp
(2) change/revert the implementation to set shallowOffsetMaxTimestamp accordingly.
(3) add tests for follower appends

chia7712 · 2024-03-29T04:47:43Z

(1) revert all offsetForMaxTimestamp to shallowOffsetMaxTimestamp
(2) change/revert the implementation to set shallowOffsetMaxTimestamp accordingly.

Do we need to revert all of them? the paths we had fixed works well now.

It seems to me adding comments for both "recover" and "follower" cases can remind readers that this offsetOfMaxTimestampMs is shallow.
or we can only rename offsetForMaxTimestamp back to shallowOffsetMaxTimestamp but we keep the implementation.

@junrao WDYT?

(3) add tests for follower appends

will complete it later

chia7712 · 2024-03-29T09:28:12Z

+    // case 0: test the offsets from leader's append path
+    check()
+
+    // case 1: test the offsets from follower's append path.


@junrao the extra tests are added. please take a look

junrao · 2024-03-29T17:30:31Z

Do we need to revert all of them? the paths we had fixed works well now.

It seems to me adding comments for both "recover" and "follower" cases can remind readers that this offsetOfMaxTimestampMs is shallow.

or we can only rename offsetForMaxTimestamp back to shallowOffsetMaxTimestamp but we keep the implementation.

@chia7712 : I think both are important. First, it's important to be able to derive the same thing consistently from the leader and the follower log. This affects things like the time indexing entries. It will be confusing if the leader adds an offset in the middle of a batch while the follower adds an offset at the end of the batch. Second, it's important to name things as accurately as possible. Otherwise, future developers could make inaccurate assumptions.

chia7712 · 2024-03-30T03:59:29Z

I think both are important. First, it's important to be able to derive the same thing consistently from the leader and the follower log. This affects things like the time indexing entries. It will be confusing if the leader adds an offset in the middle of a batch while the follower adds an offset at the end of the batch. Second, it's important to name things as accurately as possible. Otherwise, future developers could make inaccurate assumptions.

you are right. I have reverted the impl and naming. Also, I add extra comments for the "spec" of offsetOfMaxTimestamp

chia7712 · 2024-04-01T05:58:40Z

rebase code and apply Luke's patch from https://github.com/chia7712/kafka/pull/3/files

junrao

@chia7712 : Thanks for the PR. Added a few comments.

chia7712 · 2024-04-01T19:54:34Z

@junrao thanks for all your reviews and patience. all comments are addressed

junrao

@chia7712 : Thanks for the updated PR. A few more comments.

junrao · 2024-04-07T23:48:11Z

@chia7712 : There are quite a few test failures on kafka.server.ListOffsetsRequestTest.testResponseIncludesLeaderEpoch().

chia7712 · 2024-04-07T23:50:49Z

There are quite a few test failures on kafka.server.ListOffsetsRequestTest.testResponseIncludesLeaderEpoch().

yep, I have fixed it on my local. will update PR later. thanks for the reminder :)

junrao

@chia7712 : Thanks for the updated PR. Just a couple of minor comments.

chia7712 · 2024-04-08T17:17:45Z

@junrao thanks for reviews. both comments get addressed in 581242c

junrao · 2024-04-08T23:10:14Z

@chia7712 : Thanks for the updated PR. The code looks good to me. There were 50 failed tests. Is any of them related to the PR? If not, have they all been tracked?

chia7712 · 2024-04-09T11:32:09Z

Thanks for the updated PR. The code looks good to me. There were 50 failed tests. Is any of them related to the PR? If not, have they all been tracked?

there are many timeout exception, and so I feel that could be caused by busy server. I will trigger QA again instead of creating a bunch of flaky issues

chia7712 · 2024-04-09T18:16:18Z

Build / JDK 11 and Scala 2.13 / testLowMaxFetchSizeForRequestAndPartition(String, String).quorum=kraft+kip848.groupProtocol=consumer – kafka.api.PlaintextConsumerFetchTest

https://issues.apache.org/jira/browse/KAFKA-16494

Build / JDK 11 and Scala 2.13 / testFenceMultipleBrokers() – org.apache.kafka.controller.QuorumControllerTest

https://issues.apache.org/jira/browse/KAFKA-15898

Build / JDK 17 and Scala 2.13 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest
Build / JDK 8 and Scala 2.12 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest

https://issues.apache.org/jira/browse/KAFKA-15945

Build / JDK 17 and Scala 2.13 / testReplicateFromLatest() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest

https://issues.apache.org/jira/browse/KAFKA-16383

Build / JDK 8 and Scala 2.12 / testCoordinatorFailover(String, String).quorum=kraft.groupProtocol=classic – kafka.api.SslConsumerTest

https://issues.apache.org/jira/browse/KAFKA-16024

Build / JDK 8 and Scala 2.12 / "testCommitTransactionTimeout(String).quorum=kraft+kip848" – org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest

https://issues.apache.org/jira/browse/KAFKA-16495

Build / JDK 8 and Scala 2.12 / testDescribeQuorumReplicationSuccessful [2] Type=Raft-Isolated, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest

https://issues.apache.org/jira/browse/KAFKA-15104

@junrao those failed tests pass on my local, and they have jira now. Please review this PR again. thanks!

junrao · 2024-04-09T18:32:14Z

@chia7712 : Thanks for triaging the failed tests. In the last run, it seem that JDK 21 and Scala 2.13 didn't complete. Could you trigger another build? Typically, this could be done by closing the PR, waiting for 20 secs and opening it again.

chia7712 · 2024-04-09T19:29:46Z

In the last run, it seem that JDK 21 and Scala 2.13 didn't complete. Could you trigger another build? Typically, this could be done by closing the PR, waiting for 20 secs and opening it again.

thanks for the tip. I rebase the code to trigger QA in order to make sure this PR works well with latest code :)

chia7712 · 2024-04-10T02:57:07Z

Build / JDK 21 and Scala 2.13 / testInvalidPasswordSaslScram() – org.apache.kafka.common.security.authenticator.SaslAuthenticatorFailureNoDelayTest

https://issues.apache.org/jira/browse/KAFKA-16497

Build / JDK 21 and Scala 2.13 / testReplicateFromLatest() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationSSLTest

https://issues.apache.org/jira/browse/KAFKA-16383

Build / JDK 21 and Scala 2.13 / testAlterSinkConnectorOffsetsZombieSinkTasks – org.apache.kafka.connect.integration.OffsetsApiIntegrationTest

https://issues.apache.org/jira/browse/KAFKA-15917

Build / JDK 21 and Scala 2.13 / testGetSinkConnectorOffsets – org.apache.kafka.connect.integration.OffsetsApiIntegrationTest

https://issues.apache.org/jira/browse/KAFKA-16498

Build / JDK 21 and Scala 2.13 / testResetSinkConnectorOffsetsOverriddenConsumerGroupId – org.apache.kafka.connect.integration.OffsetsApiIntegrationTest

https://issues.apache.org/jira/browse/KAFKA-15891

Build / JDK 21 and Scala 2.13 / testCacheEviction() – org.apache.kafka.server.ClientMetricsManagerTest

https://issues.apache.org/jira/browse/KAFKA-16499

Build / JDK 21 and Scala 2.13 / testTaskRequestWithOldStartMsGetsUpdated() – org.apache.kafka.trogdor.coordinator.CoordinatorTest

https://issues.apache.org/jira/browse/KAFKA-16136

Build / JDK 17 and Scala 2.13 / "testTrustStoreAlter(String).quorum=kraft" – kafka.server.DynamicBrokerReconfigurationTest

https://issues.apache.org/jira/browse/KAFKA-16500

Build / JDK 8 and Scala 2.12 / testConsumptionWithBrokerFailures() – kafka.api.ConsumerBounceTest

https://issues.apache.org/jira/browse/KAFKA-15146

Build / JDK 11 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest

https://issues.apache.org/jira/browse/KAFKA-15927

Build / JDK 11 and Scala 2.13 / testSeparateOffsetsTopic – org.apache.kafka.connect.integration.ExactlyOnceSourceIntegrationTest

https://issues.apache.org/jira/browse/KAFKA-14089

Build / JDK 11 and Scala 2.13 / testConsumptionWithBrokerFailures() – kafka.api.ConsumerBounceTest

https://issues.apache.org/jira/browse/KAFKA-15146

Build / JDK 11 and Scala 2.13 / "testCreateUserWithDelegationToken(String).quorum=kraft" – kafka.api.DelegationTokenEndToEndAuthorizationWithOwnerTest

https://issues.apache.org/jira/browse/KAFKA-16501

Build / JDK 11 and Scala 2.13 / "testBrokerHeartbeatDuringMigration(MetadataVersion).metadataVersion=3.4-IV0" – org.apache.kafka.controller.QuorumControllerTest

https://issues.apache.org/jira/browse/KAFKA-15963

Build / JDK 11 and Scala 2.13 / shouldWorkWithUncleanShutdownWipeOutStateStore[exactly_once_v2] – org.apache.kafka.streams.integration.EOSUncleanShutdownIntegrationTest

https://issues.apache.org/jira/browse/KAFKA-16502

Build / JDK 11 and Scala 2.13 / testDescribeQuorumReplicationSuccessful [1] Type=Rt-Combined, MetadataVersion=3.8-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest

https://issues.apache.org/jira/browse/KAFKA-15104

ok, all pass on my local and they have jira.

junrao

@chia7712 : Thanks for triaging the tests. LGTM

chia7712 · 2024-04-10T03:43:48Z

@junrao @showuon thanks for all your reviews and help!

junrao · 2024-05-07T17:48:22Z

-    val expectedOffsetOfMaxTimestamp = 1
-    assertEquals(expectedOffsetOfMaxTimestamp, validatingResults.offsetOfMaxTimestampMs,
-      s"Offset of max timestamp should be 1")
+    assertEquals(2, validatingResults.shallowOffsetOfMaxTimestamp)


@chia7712 : There seems to be an existing bug. The method is checkNonCompressed(), but in line 370, we set the compression codec to GZIP.

yep, that is a "TYPO" but it does not change the test. We do pass the "NONE" to create LogValidator so it will run the path assignOffsetsNonCompressed

kafka/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala

Line 377 in 525b9b1

CompressionType.NONE,

However, I do observe a potential bug.

context

Those batches can have different compression

We take the compression from last batch

kafka/core/src/main/scala/kafka/log/UnifiedLog.scala

Line 1180 in 525b9b1

if (batchCompression != CompressionType.NONE)

potential bug

topic-level compression = GZIP
batch_0 = NONE
batch_1 = GZIP

In this case, we don't rebuild records according to topic-level compression since the compression of "last batch" is equal to GZIP. Hence, it results in batch_0 having incorrect compression.

This bug does not produce corrupt records, so we can add comments/docs to describe that issue. Or we can fix it by changing the sourceCompression to be a "collection" of all batches' compression, and then do conversion if one of them is mismatched.

for another, LogValidator is moved to storage module already but its unit test is still in core module. That is a bit weird. We can rewrite it by java with bug fix and then move it to storage module. I have filed https://issues.apache.org/jira/browse/KAFKA-16689

yep, that is a "TYPO" but it does not change the test.

The issue is that the test is testing the wrong expected value. For magic of 1, the offset for max timestamp should be 1 instead of 2.

However, I do observe a potential bug.

Yes, this can lead to inaccurate LogAppendInfo.sourceCompression. But it doesn't seem to have real impact now. LogAppendInfo.sourceCompression is only used in LogValidator, which is only called by the leader. In the leader, currently, we expect only 1 batch per producer.

apache#15621) We do iterate the records to find the offsetOfMaxTimestamp instead of returning the cached one when handling ListOffsetsRequest.MAX_TIMESTAMP, since it is hard to align all paths to get correct offsetOfMaxTimestamp. The known paths are shown below. 1. convertAndAssignOffsetsNonCompressed -> we CAN get correct offsetOfMaxTimestamp when validating all records 2. assignOffsetsNonCompressed -> ditto 3. validateMessagesAndAssignOffsetsCompressed -> ditto 4. validateMessagesAndAssignOffsetsCompressed#buildRecordsAndAssignOffsets -> ditto 5. appendAsFollow#append#analyzeAndValidateRecords -> we CAN'T get correct offsetOfMaxTimestamp as iterating all records is expensive when fetching records from leader 6. LogSegment#recover -> ditto Reviewers: Jun Rao <junrao@gmail.com>

KAFKA-16310 ListOffsets doesn't report the offset with maxTimestamp a…

a7d6288

…nymore

chia7712 requested review from ijuma, junrao and showuon March 28, 2024 20:31

Merge branch 'trunk' into KAFKA-16310

bd32d03

junrao reviewed Mar 28, 2024

View reviewed changes

Comment thread core/src/main/scala/kafka/log/UnifiedLog.scala Outdated

Merge branch 'trunk' into KAFKA-16310

0f18cae

add more tests

2c00cee

chia7712 commented Mar 29, 2024

View reviewed changes

chia7712 added 3 commits March 30, 2024 02:05

Merge branch 'trunk' into KAFKA-16310

5278cd1

revert code and add comments

90d33e0

Merge branch 'trunk' into KAFKA-16310

f11b877

chia7712 added 7 commits March 30, 2024 12:09

adjust comments

5180f2d

adjust comments

79561cd

adjust comments

726de99

ranem offsetOfMaxTimestampSoFar to shallowOffsetOfMaxTimestampSoFar

29137ae

Merge branch 'trunk' into KAFKA-16310

1cf8ca4

Merge branch 'trunk' into KAFKA-16310

ac729a6

apply luke patch

2297b52

Merge branch 'trunk' into KAFKA-16310

a1fbc6c

junrao reviewed Apr 1, 2024

View reviewed changes

chia7712 added 2 commits April 2, 2024 03:10

Merge branch 'trunk' into KAFKA-16310

9505115

address comments

b4b1332

junrao reviewed Apr 1, 2024

View reviewed changes

address comments

609b6ab

chia7712 added 6 commits April 7, 2024 00:01

Merge branch 'trunk' into KAFKA-16310

9af7513

address comments

8b1005e

Merge branch 'trunk' into KAFKA-16310

58b5e2f

fix failed tests

728e7bb

Merge branch 'trunk' into KAFKA-16310

e21447e

Merge branch 'trunk' into KAFKA-16310

9e22f2d

chia7712 added 2 commits April 8, 2024 07:51

fix testResponseIncludesLeaderEpoch

fe2d7c9

fix failed test

2dee703

junrao reviewed Apr 8, 2024

View reviewed changes

Comment thread core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala Outdated

Comment thread core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala

chia7712 added 2 commits April 9, 2024 01:14

Merge branch 'trunk' into KAFKA-16310

b0afa6d

address comments

581242c

Merge branch 'trunk' into KAFKA-16310

d42fe26

Merge branch 'trunk' into KAFKA-16310

d12fce2

junrao approved these changes Apr 10, 2024

View reviewed changes

chia7712 merged commit 9a6760f into apache:trunk Apr 10, 2024

chia7712 deleted the KAFKA-16310 branch April 26, 2024 22:38

junrao reviewed May 7, 2024

View reviewed changes

chia7712 mentioned this pull request May 9, 2024

MINOR: fix LogValidatorTest#checkNonCompressed #15904

Merged

3 tasks

kamalcph mentioned this pull request Sep 14, 2024

KAFKA-15859: Make RemoteListOffsets call an async operation #16602

Merged

3 tasks

Conversation

chia7712 commented Mar 28, 2024

Committer Checklist (excluded from commit message)

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

chia7712 commented Mar 29, 2024

Uh oh!

chia7712 Mar 29, 2024

Choose a reason for hiding this comment

Uh oh!

junrao commented Mar 29, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chia7712 commented Mar 30, 2024

Uh oh!

chia7712 commented Apr 1, 2024

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

chia7712 commented Apr 1, 2024

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

junrao commented Apr 7, 2024

Uh oh!

chia7712 commented Apr 7, 2024

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

chia7712 commented Apr 8, 2024

Uh oh!

junrao commented Apr 8, 2024

Uh oh!

chia7712 commented Apr 9, 2024

Uh oh!

chia7712 commented Apr 9, 2024

Uh oh!

junrao commented Apr 9, 2024

Uh oh!

chia7712 commented Apr 9, 2024

Uh oh!

chia7712 commented Apr 10, 2024

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

chia7712 commented Apr 10, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

junrao May 7, 2024

Choose a reason for hiding this comment

Uh oh!

chia7712 May 8, 2024

Choose a reason for hiding this comment

Uh oh!

chia7712 May 8, 2024

Choose a reason for hiding this comment

Uh oh!

junrao May 8, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

junrao commented Mar 29, 2024 •

edited

Loading

chia7712 commented Apr 10, 2024 •

edited

Loading

junrao May 8, 2024 •

edited

Loading