Skip to content

Consolidate auto.offset.reset config - #44

Merged
jeffxiang merged 7 commits into
mainfrom
consumer_improvements
Nov 24, 2025
Merged

Consolidate auto.offset.reset config#44
jeffxiang merged 7 commits into
mainfrom
consumer_improvements

Conversation

@jeffxiang

@jeffxiang jeffxiang commented Nov 20, 2025

Copy link
Copy Markdown
Contributor

This PR removes the offset.reset.strategy TieredStorageConsumer config and consolidates it with native KafkaConsumer's auto.offset.reset config. The old model of having both configs makes it confusing and difficult to understand the consumer's offset reset behavior, especially with different TieredStorageModes.

Now, the TieredStorageConsumer will only accept the native KafkaConsumer's auto.offset.reset config, and use it throughout the consumer logic. The previous TieredStorageConfig offset.reset.strategy (if supplied) will be ignored.

The way we use this config is as follows:

mode = KAFKA_ONLY

  • The auto.offset.reset config will be passed directly into the native KafkaConsumer, resulting in the same behavior as regular KafkaConsumer

mode = KAFKA_PREFERRED

  • The auto.offset.reset config provided by the user-supplied Properties will be first extracted and stored as a field
  • We will internally override this config to be none when passing it into the KafkaConsumer, to ensure that KafkaConsumer.poll() will throw OffsetOutOfRangeException if the offset is out of range, instead of automatically resetting offsets within the KafkaConsumer
  • If KafkaConsumer.poll() throws an OffsetOutOfRangeException, we fall back to S3Consumer.poll() as we did previously
  • If any S3PartitionConsumer.poll() encounters a NoSuchKeyException (the S3 version of OffsetOutOfRange), we will reset offsets based on the auto.offset.reset config supplied by the user, or throw the exception as an OffsetOutOfRangeException if the user supplied auto.offset.reset=none

mode = TIERED_STORAGE_ONLY

  • This mode is untested / under development

@jeffxiang jeffxiang changed the title Consumer improvements Consolidate auto.offset.reset config Nov 21, 2025
@jeffxiang
jeffxiang marked this pull request as ready for review November 21, 2025 21:00
@jeffxiang
jeffxiang requested a review from a team as a code owner November 21, 2025 21:00

@vahidhashemian vahidhashemian left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor nit inside.

Since this is a breaking change, I'd suggest documenting how to switch from old version to the new and if different circumstances could affect how that switch should be done.

offsetResetConfig.equals("none") ? OffsetReset.NONE :
OffsetReset.LATEST;
LOG.info("Offset reset policy: " + this.offsetReset);
String autoOffsetResetConfig = properties.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest").toUpperCase().trim();

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: If strings like "latest" or "none" are used in multiple places, I'd suggest using a global constant for them.

@jeffxiang
jeffxiang merged commit 0d006da into main Nov 24, 2025
1 check passed
@jeffxiang
jeffxiang deleted the consumer_improvements branch November 24, 2025 21:08
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants