[jira] [Created] (FLINK-40575) Support dynamic table unsubscription and re-subscription in Fluss pipeline source

["Xiaobing Fang (Jira)" <ji...@apache.org>]

Xiaobing Fang created FLINK-40575:
-------------------------------------

             Summary: Support dynamic table unsubscription and re-subscription in Fluss pipeline source
                 Key: FLINK-40575
                 URL: https://issues.apache.org/jira/browse/FLINK-40575
             Project: Flink
          Issue Type: Improvement
            Reporter: Xiaobing Fang


### Background

 

`TableDiscoverer` periodically returns the complete current set of subscribed tables. However, the Fluss pipeline source currently handles only newly discovered tables.

 

When a table disappears from the discovery result, `FlussSourceEnumerator` retains its assigned splits. The source therefore continues reading the table, and the retained split state may restore it after a checkpoint or savepoint recovery.

 

### Expected behavior

 

The Fluss pipeline source should treat every successful discovery result as the authoritative current subscription:

 

- Removing a table from the result stops its records without restarting the job.

- Other subscribed tables continue running normally.

- Restoring from a checkpoint must not resume a table that is still unsubscribed.

- Re-adding a table starts a new source lifecycle according to `scan.startup.mode` and emits a new `CreateTableEvent`.

- Removal affects source-side splits and cached metadata only. It must not emit a `DropTableEvent` or delete the sink table.

- A successful empty result unsubscribes all tables. Discovery failures fail the job without applying an incomplete result.

 

### Proposed change

 

- Detect removals by logical `TablePath`, independently of partition changes.

- Persist pending-removal tombstones in enumerator checkpoint state.

- Block restored splits until a fresh authoritative subscription snapshot is available.

- Coordinate split cleanup between the enumerator and every source-reader subtask.

- Unsubscribe active Fluss buckets and remove finished splits through the normal source-reader cleanup path.

- Clear table-level schema and deserialization caches so re-added tables are initialized as new tables.

- Prevent pending assignments, failed-reader split returns, and delayed asynchronous initialization from resurrecting removed tables.

- Document the discovery, removal, recovery, and re-subscription semantics.

 

The enumerator state serializer will be upgraded to version 2, while retaining support for restoring version 1 state.

 

### Acceptance criteria

 

1. With tables A and B subscribed, removing B stops new B records while A continues, without restarting the job.

2. Restoring an old checkpoint while B remains unsubscribed produces no new B records.

3. Re-adding B creates fresh splits using the configured `scan.startup.mode` rather than continuing its removed split.

4. Removing one partition from a still-subscribed partitioned table is not treated as logical-table unsubscription.

5. Pending assignments, delayed initialization results, and splits returned during failover cannot revive an unsubscribed table.

6. The behavior is covered for both Flink 1.20 and Flink 2.2.

### Background

`TableDiscoverer` periodically returns the complete current set of subscribed tables. However, the Fluss pipeline source currently handles only newly discovered tables.

When a table disappears from the discovery result, `FlussSourceEnumerator` retains its assigned splits. The source therefore continues reading the table, and the retained split state may restore it after a checkpoint or savepoint recovery.

### Expected behavior

The Fluss pipeline source should treat every successful discovery result as the authoritative current subscription:

- Removing a table from the result stops its records without restarting the job.
- Other subscribed tables continue running normally.
- Restoring from a checkpoint must not resume a table that is still unsubscribed.
- Re-adding a table starts a new source lifecycle according to `scan.startup.mode` and emits a new `CreateTableEvent`.
- Removal affects source-side splits and cached metadata only. It must not emit a `DropTableEvent` or delete the sink table.
- A successful empty result unsubscribes all tables. Discovery failures fail the job without applying an incomplete result.

### Proposed change

- Detect removals by logical `TablePath`, independently of partition changes.
- Persist pending-removal tombstones in enumerator checkpoint state.
- Block restored splits until a fresh authoritative subscription snapshot is available.
- Coordinate split cleanup between the enumerator and every source-reader subtask.
- Unsubscribe active Fluss buckets and remove finished splits through the normal source-reader cleanup path.
- Clear table-level schema and deserialization caches so re-added tables are initialized as new tables.
- Prevent pending assignments, failed-reader split returns, and delayed asynchronous initialization from resurrecting removed tables.
- Document the discovery, removal, recovery, and re-subscription semantics.

The enumerator state serializer will be upgraded to version 2, while retaining support for restoring version 1 state.

### Acceptance criteria

1. With tables A and B subscribed, removing B stops new B records while A continues, without restarting the job.
2. Restoring an old checkpoint while B remains unsubscribed produces no new B records.
3. Re-adding B creates fresh splits using the configured `scan.startup.mode` rather than continuing its removed split.
4. Removing one partition from a still-subscribed partitioned table is not treated as logical-table unsubscription.
5. Pending assignments, delayed initialization results, and splits returned during failover cannot revive an unsubscribed table.
6. The behavior is covered for both Flink 1.20 and Flink 2.2.



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