[jira] [Created] (FLINK-39888) [Kafka] Allow configuring offset reset strategy independently of startup offsets

["Jigar Bhati (Jira)" <ji...@apache.org>]

Jigar Bhati created FLINK-39888:
-----------------------------------

             Summary: [Kafka] Allow configuring offset reset strategy independently of startup offsets
                 Key: FLINK-39888
                 URL: https://issues.apache.org/jira/browse/FLINK-39888
             Project: Flink
          Issue Type: New Feature
          Components: Connectors / Kafka
            Reporter: Jigar Bhati


KafkaSource currently uses OffsetsInitializer for two separate concerns:
1. resolving the initial starting offsets
2. selecting Kafka's auto.offset.reset strategy when an initialized starting offset is unavailable

These choices are independent, but the public API does not let users compose them for all startup modes.

One concrete failure mode is broker recovery after a log corruption or truncation event. A source may have previously checkpointed or otherwise resolved a concrete numeric offset while Kafka was healthy. After recovery, Kafka can report that this offset is no longer available on the partition (which could be due to corruption or retention or anything similar). At that point, the Kafka consumer applies auto.offset.reset. With the current coupling, a source that originally started with earliest-offset or latest-offset also inherits earliest/latest as the fallback for that later out-of-range concrete offset. The job can therefore silently reset to earliest or latest instead of failing with an out-of-range error that lets an operator decide what to do.

The desired behavior is to configure these independently. For example, users may want a fresh source to start from earliest or latest, but use OffsetResetStrategy.NONE once a concrete initialized offset has been chosen, so a later unavailable offset fails explicitly instead of silently resetting.

The SQL Kafka connector has the same coupling today: properties.auto.offset.reset is only applied for group-offsets startup, while explicit values are overridden by the startup initializer for earliest-offset, latest-offset, specific-offsets, and timestamp startup modes.

Proposed change:
- Add OffsetsInitializer.withOffsetResetStrategy(OffsetsInitializer, OffsetResetStrategy).
- The wrapper delegates partition offset initialization and validation, but overrides getAutoOffsetResetStrategy().
- Apply explicit properties.auto.offset.reset to the chosen Kafka SQL startup initializer for all startup modes.
- Preserve existing defaults when properties.auto.offset.reset is not explicitly configured.
- Update docs to clarify that scan.startup.mode chooses the initial position, while auto.offset.reset controls fallback when an initialized starting offset is unavailable.

Related:
FLINK-39382 documents current auto.offset.reset support for group-offsets.

Testing:
- KafkaSourceBuilder reset strategy tests
- DynamicKafkaSourceBuilder reset strategy tests
- initializer wrapper unit test
- static Kafka SQL source tests
- dynamic Kafka SQL source tests



--
This message was sent by Atlassian Jira
(v8.20.10#820010)