KAFKA-17302: ReplicaFetcher changes for fetching from tiered offset #20428
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
github-actions
bot
added
core
abhijeetk88
force-pushed
the
KAFKA-17302
branch
2 times, most recently
from
c125bd0 to
c07a3f6
Compare
abhijeetk88
changed the title
abhijeetk88
force-pushed
the
KAFKA-17302
branch
8 times, most recently
from
408fb46 to
72a4dbf
Compare
abhijeetk88
force-pushed
the
KAFKA-17302
branch
from
72a4dbf to
446efe0
Compare
kamalcph
reviewed
Sep 3, 2025
There was a problem hiding this comment.
Thanks @abhijeetk88 for the patch! Went over the first pass.
Could you explain why the TierStateMachine#buildRemoteLogAuxState is not used to decide the offset-to-fetch either leaderLocalStartOffset or lastTieredOffset?
It fits nicely:
- Minimal code change and maintenance.
- If this code change is to handle the case when leader-log-start-offset == leader-local-log-start-offset == 0 and earliestUploadOffset > 0 (yet to be approved), then we may have to discuss it before proceeding with the patch. It is one of the edge-case but not sure how useful it would be in practice.
| @@ -182,7 +179,7 @@ private void buildProducerSnapshotFile(UnifiedLog unifiedLog, | |||
| * fetching records from the leader. The return value is the next offset to fetch from the leader, which is the | |||
| * next offset following the end offset of the remote log portion. | |||
| */ | |||
| private Long buildRemoteLogAuxState(TopicPartition topicPartition, | |||
| protected Long buildRemoteLogAuxState(TopicPartition topicPartition, | |||
| val earliestPendingUploadOffset = Math.max(highestRemoteOffset + 1, logStartOffset.offset()) | ||
| val epoch = log.leaderEpochCache.epochForOffset(earliestPendingUploadOffset) | ||
|
|
||
| new OffsetAndEpoch(earliestPendingUploadOffset, epoch.orElse(0)) |
There was a problem hiding this comment.
epoch 0 is vaild. If the epoch is empty, shall we return -1?
| val leaderState = leaderPartitionState(topicPartition) | ||
| checkLeaderEpochAndThrow(leaderEpoch, leaderState) | ||
| val offsetAndEpoch = leaderState.earliestPendingUploadOffset match { | ||
| case -1L => (-1L, -1) |
There was a problem hiding this comment.
can the OffsetAndEpoch be returned directly?
leaderState.earliestPendingUploadOffset match {
case -1L => new OffsetAndEpoch(-1L, -1)
case _ => new OffsetAndEpoch(math.max(leaderState.earliestPendingUploadOffset, leaderState.logStartOffset), leaderState.leaderEpoch)
}
| expected: Boolean): Unit = { | ||
| val tp = new TopicPartition("t", 0) | ||
|
|
||
| def runScenario(enableLastTieredOffsetFetch: Boolean, |
There was a problem hiding this comment.
do we need this internal runScenario method? Can this be flattened?
assertEquals(
expected,
thread.callShouldFetchFromLastTieredOffset(tp, leaderEndOffset, replicaEndOffset)
)| ) | ||
| } | ||
|
|
||
| private class MockReplicaFetcherThread(name: String, |
There was a problem hiding this comment.
can we remove this MockReplicaFetcherThread and expose the shouldFetchFromLastTieredOffset in RFT to server package?
override protected[server] def shouldFetchFromLastTieredOffset(topicPartition: TopicPartition, leaderEndOffset: Long, replicaEndOffset: Long): Boolean
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Delete this text and replace it with a detailed description of your change. The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are expected
for any behavior change and system tests should be considered for larger
changes.