KAFKA-16480: ListOffsets change should have an associated API/IBP version update#16781
Conversation
showuon
left a comment
showuon
left a comment
There was a problem hiding this comment.
Thanks for the PR. Left some comments.
| .stream() | ||
| .anyMatch(key -> offsetTimestampsByPartition.get(key) == ListOffsetsRequest.MAX_TIMESTAMP); | ||
|
|
||
| boolean requireTiredTimestamp = keys |
There was a problem hiding this comment.
Maybe rename to requireTieredStorageTimestamp
| @ParameterizedTest | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testListOffsetsMaxTimeStampOldestVersion(quorum: String): Unit = { | ||
| @Test | ||
| def testListOffsetsMaxTimeStampOldestVersion(): Unit = { |
There was a problem hiding this comment.
Why do we remove the original ZK and kraft infra's tests?
There was a problem hiding this comment.
I add it back, but I think this test is unit test, so we don't need to test it on both zk and kraft modes.
There was a problem hiding this comment.
I know what you're saying, but let's move the refactor in a separate PR.
| } | ||
|
|
||
| @Test | ||
| def testListOffsetsRequestTiredTimestampOldestVersion(): Unit = { |
| val consumerRequestBuilder = ListOffsetsRequest.Builder | ||
| .forConsumer(false, IsolationLevel.READ_UNCOMMITTED, false, false) |
There was a problem hiding this comment.
This consumerRequestBuilder test is identical with the one in testListOffsetsMaxTimeStampOldestVersion. I think we can integrate these 2 tests into 1, and rename to something like: testListOffsetsRequestOldestVersion or something else. And I think (if we have time), we also need to test requireTimestamp and isolationLevel == IsolationLevel.READ_COMMITTED since I didn't see there's test covered them. WDYT?
There was a problem hiding this comment.
Yes, I add all cases to testListOffsetsRequestOldestVersion. Thanks.
…sion update Signed-off-by: PoAn Yang <payang@apache.org>
cdce631 to
83e5c3c
Compare
|
Hi @showuon, thanks for the review. I address all comments. |
showuon
left a comment
showuon
left a comment
There was a problem hiding this comment.
LGTM! Is this PR waiting for something before it moves to ready for review?
|
It's ready. I'm waiting for CI result. |
…sion update (#16781) 1. Use oldestAllowedVersion as 9 if using ListOffsetsRequest#EARLIEST_LOCAL_TIMESTAMP or ListOffsetsRequest#LATEST_TIERED_TIMESTAMP. 2. Add test cases to ListOffsetsRequestTest#testListOffsetsRequestOldestVersion to make sure requireTieredStorageTimestamp return 9 as minVersion. 3. Add EarliestLocalSpec and LatestTierSpec to OffsetSpec. 4. Add more cases to KafkaAdminClient#getOffsetFromSpec. 5. Add testListOffsetsEarliestLocalSpecMinVersion and testListOffsetsLatestTierSpecSpecMinVersion to KafkaAdminClientTest to make sure request builder has oldestAllowedVersion as 9. Signed-off-by: PoAn Yang <payang@apache.org> Reviewers: Luke Chen <showuon@gmail.com>
2 participants
oldestAllowedVersionas9if usingListOffsetsRequest#EARLIEST_LOCAL_TIMESTAMPorListOffsetsRequest#LATEST_TIERED_TIMESTAMP.ListOffsetsRequestTest#testListOffsetsRequestOldestVersionto make surerequireTieredStorageTimestampreturn9asminVersion.EarliestLocalSpecandLatestTierSpectoOffsetSpec.KafkaAdminClient#getOffsetFromSpec.testListOffsetsEarliestLocalSpecMinVersionandtestListOffsetsLatestTierSpecSpecMinVersiontoKafkaAdminClientTestto make sure request builder hasoldestAllowedVersionas9.Committer Checklist (excluded from commit message)