Ensure S3PartitionConsumer throws OffsetOutOfRangeException if position is earlier than earliest s3 offset - #45
Merged
Merged
Conversation
jeffxiang
marked this pull request as ready for review
November 26, 2025 21:04
vahidhashemian
approved these changes
Dec 8, 2025
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
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.
There is a bug in
S3PartitionConsumerwhere if the desired fetch offset is smaller than the earliest available offset on S3, it will automatically reset position to earliest on s3. This is due to the fact that we perform a binary search on the S3 log segments to locate the first S3 segment which might contain the desired offset, meaning that the found segment whendesiredOffset < earliestis just the earliest log segment. The effect of this is that S3PartitionConsumer does not honor the user-suppliedauto.offset.resetconfig, especially if the user has suppliedlatest. This undesirable behavior happens when theTieredStorageConsumeris constructed withauto.offset.reset=latestandgroup.idis set to a group where the committed offsets exist, but are expired.In the case that the
group.idis entirely new and there does not exist any previously-committed offsets for that group, this PR did not change any behavior. This is because AssignmentAwareConsumerRebalanceListener resets consumer positions according toauto.offset.resetconfig proactively when committed group offsets do not exist. However, if they do exist, the RebalanceListener will set fetch positions to the committed offsets even if they are expired. When these fetch positions are passed down toS3PartitionConsumer, the current behavior before this PR is that theS3PartitionConsumerwill skip the "lost" offsets and reset to earliest, no matter what theauto.offset.resetconfig is.This PR addresses the issue by throwing an
OffsetOutOfRangeExceptionwhen the above scenario is detected, and bubbling up the exception toTieredStorageConsumerwhere the offset reset logic already exists.TieredStorageConsumeris then responsible for handling the exception according toauto.offset.reset.This PR also includes additional refactoring, such as deduplicating some S3 poll logic code in
TieredStorageConsumer.