[jira] [Updated] (FLINK-39435) [flink-connector-kafka] Switching Kafka topics via Savepoint restore causes dual-topic consumption AND new topic consumed from earliest instead of configured offset

["sherlock-lin (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-39435?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

sherlock-lin updated FLINK-39435:
---------------------------------
    Description: 
In `KafkaSourceEnumerator.handlePartitionSplitChanges()`, there is a long-standing `TODO` comment:
{quote}```
// TODO: Handle removed partitions.
addPartitionSplitChangeToPendingAssignments(partitionSplitChange.newPartitionSplits);
assignPendingPartitionSplits(context.registeredReaders().keySet());
```
 
{quote}
Two separate bugs are triggered simultaneously:

*{*}Bug 1 — Dual-topic consumption (old topic is never stopped){*}*

The job consumes from *{*}both `topic-A` and `topic-B` simultaneously{*}*:
 - `topic-A` splits come from the *{*}Reader State{*}* restored from Savepoint (`SourceOperator.open()` → `sourceReader.addSplits(restoredSplits)`).
 - `topic-B` splits come from the *{*}Enumerator{*}* discovering new partitions (`initializePartitionSplits()` → `assignSplits()`).
 - In `KafkaPartitionSplitReader.handleSplitsChanges()`, line:
```java
newPartitionAssignments.addAll(consumer.assignment()); // merges, does NOT replace
consumer.assign(newPartitionAssignments);
```
The old topic's partitions remain in the consumer assignment permanently. The Enumerator detects the removed partitions in `getPartitionChange()` but does nothing with them (`// TODO: Handle removed partitions.`).

*{*}Bug 2 — New topic consumed from earliest instead of configured `latest()`{*}*

Even though the job is configured with `.setStartingOffsets(OffsetsInitializer.latest())`, the new topic (`topic-B`) is consumed from the *{*}earliest{*}* available offset.

Log evidence:
```
INFO KafkaSourceEnumerator - Assigning splits to readers

{57=[[Partition: topic-B-0, StartingOffset: -2, StoppingOffset: -9223372036854775808], ...]}

```
Sentinel `-2` = `EARLIEST_OFFSET`, not `-1` (`LATEST_OFFSET`) as configured.

Root cause: `setStartingOffsets()` only takes effect for *{*}truly new partitions{*}* discovered by the Enumerator. During Savepoint restore, the Reader State injects old splits directly into the Reader via `SourceOperator.open() → sourceReader.addSplits(restoredSplits)`, bypassing `setStartingOffsets()` entirely. The default `startingOffsetsInitializer` in `KafkaSourceBuilder` (line 106) is `OffsetsInitializer.earliest()`, and for partitions not found in Reader State this default is used instead of the user-configured value.

*{*}Expected behavior{*}*

When `getPartitionChange()` detects `removedPartitions`, the Enumerator should send a signal to the affected Readers to *{*}unassign{*}* those partitions from their `KafkaConsumer`, so the job only consumes from the newly subscribed topic.

  was:
In `KafkaSourceEnumerator.handlePartitionSplitChanges()`, there is a long-standing `TODO` comment:

```java
// TODO: Handle removed partitions.
addPartitionSplitChangeToPendingAssignments(partitionSplitChange.newPartitionSplits);
assignPendingPartitionSplits(context.registeredReaders().keySet());
```
 

Two separate bugs are triggered simultaneously:

**Bug 1 — Dual-topic consumption (old topic is never stopped)**

The job consumes from **both `topic-A` and `topic-B` simultaneously**:

- `topic-A` splits come from the **Reader State** restored from Savepoint (`SourceOperator.open()` → `sourceReader.addSplits(restoredSplits)`). 
- `topic-B` splits come from the **Enumerator** discovering new partitions (`initializePartitionSplits()` → `assignSplits()`). 
- In `KafkaPartitionSplitReader.handleSplitsChanges()`, line:
```java
newPartitionAssignments.addAll(consumer.assignment()); // merges, does NOT replace
consumer.assign(newPartitionAssignments);
```
The old topic's partitions remain in the consumer assignment permanently. The Enumerator detects the removed partitions in `getPartitionChange()` but does nothing with them (`// TODO: Handle removed partitions.`).

**Bug 2 — New topic consumed from earliest instead of configured `latest()`**

Even though the job is configured with `.setStartingOffsets(OffsetsInitializer.latest())`, the new topic (`topic-B`) is consumed from the **earliest** available offset.

Log evidence:
```
INFO KafkaSourceEnumerator - Assigning splits to readers
{57=[[Partition: topic-B-0, StartingOffset: -2, StoppingOffset: -9223372036854775808], ...]}
```
Sentinel `-2` = `EARLIEST_OFFSET`, not `-1` (`LATEST_OFFSET`) as configured.

Root cause: `setStartingOffsets()` only takes effect for **truly new partitions** discovered by the Enumerator. During Savepoint restore, the Reader State injects old splits directly into the Reader via `SourceOperator.open() → sourceReader.addSplits(restoredSplits)`, bypassing `setStartingOffsets()` entirely. The default `startingOffsetsInitializer` in `KafkaSourceBuilder` (line 106) is `OffsetsInitializer.earliest()`, and for partitions not found in Reader State this default is used instead of the user-configured value.

**Expected behavior**

When `getPartitionChange()` detects `removedPartitions`, the Enumerator should send a signal to the affected Readers to **unassign** those partitions from their `KafkaConsumer`, so the job only consumes from the newly subscribed topic.


> [flink-connector-kafka] Switching Kafka topics via Savepoint restore causes dual-topic consumption AND new topic consumed from earliest instead of configured offset
> --------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-39435
>                 URL: https://issues.apache.org/jira/browse/FLINK-39435
>             Project: Flink
>          Issue Type: Bug
>         Environment: - Flink version: 1.17.0
>            Reporter: sherlock-lin
>            Priority: Major
>              Labels: bug, flink-connector-kafka
>
> In `KafkaSourceEnumerator.handlePartitionSplitChanges()`, there is a long-standing `TODO` comment:
> {quote}```
> // TODO: Handle removed partitions.
> addPartitionSplitChangeToPendingAssignments(partitionSplitChange.newPartitionSplits);
> assignPendingPartitionSplits(context.registeredReaders().keySet());
> ```
>  
> {quote}
> Two separate bugs are triggered simultaneously:
> *{*}Bug 1 — Dual-topic consumption (old topic is never stopped){*}*
> The job consumes from *{*}both `topic-A` and `topic-B` simultaneously{*}*:
>  - `topic-A` splits come from the *{*}Reader State{*}* restored from Savepoint (`SourceOperator.open()` → `sourceReader.addSplits(restoredSplits)`).
>  - `topic-B` splits come from the *{*}Enumerator{*}* discovering new partitions (`initializePartitionSplits()` → `assignSplits()`).
>  - In `KafkaPartitionSplitReader.handleSplitsChanges()`, line:
> ```java
> newPartitionAssignments.addAll(consumer.assignment()); // merges, does NOT replace
> consumer.assign(newPartitionAssignments);
> ```
> The old topic's partitions remain in the consumer assignment permanently. The Enumerator detects the removed partitions in `getPartitionChange()` but does nothing with them (`// TODO: Handle removed partitions.`).
> *{*}Bug 2 — New topic consumed from earliest instead of configured `latest()`{*}*
> Even though the job is configured with `.setStartingOffsets(OffsetsInitializer.latest())`, the new topic (`topic-B`) is consumed from the *{*}earliest{*}* available offset.
> Log evidence:
> ```
> INFO KafkaSourceEnumerator - Assigning splits to readers
> {57=[[Partition: topic-B-0, StartingOffset: -2, StoppingOffset: -9223372036854775808], ...]}
> ```
> Sentinel `-2` = `EARLIEST_OFFSET`, not `-1` (`LATEST_OFFSET`) as configured.
> Root cause: `setStartingOffsets()` only takes effect for *{*}truly new partitions{*}* discovered by the Enumerator. During Savepoint restore, the Reader State injects old splits directly into the Reader via `SourceOperator.open() → sourceReader.addSplits(restoredSplits)`, bypassing `setStartingOffsets()` entirely. The default `startingOffsetsInitializer` in `KafkaSourceBuilder` (line 106) is `OffsetsInitializer.earliest()`, and for partitions not found in Reader State this default is used instead of the user-configured value.
> *{*}Expected behavior{*}*
> When `getPartitionChange()` detects `removedPartitions`, the Enumerator should send a signal to the affected Readers to *{*}unassign{*}* those partitions from their `KafkaConsumer`, so the job only consumes from the newly subscribed topic.



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