[
https://issues.apache.org/jira/browse/FLINK-40575?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Xiaobing Fang updated FLINK-40575:
----------------------------------
Description:
### 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.
was:
### 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.
> 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
> Priority: Major
>
> ### 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)
