apache · fxbing · Sep 7, 2026
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md b/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
@@ -25,12 +25,32 @@ under the License.
 -->
 
 # Fluss Pipeline 连接器
-Fluss Pipeline 连接器可用作 Pipeline 的 *Data Sink*，将数据写入 [Fluss](https://fluss.apache.org)。本文档介绍如何配置 Fluss Pipeline 连接器。
+Fluss Pipeline 连接器可用作 Pipeline 的 *Data Source* 或 *Data Sink*，从 [Fluss](https://fluss.apache.org)
+读取或向其写入数据。本文档介绍这两种用法的配置。
 
 ## What can the connector do?
 * 自动创建不存在的表 
 * 数据同步
 * Schema 变更同步（lenient 模式）
+* 动态 Source 表订阅
+
+## Fluss Source
+
+以下是动态发现 Fluss 表并读取的最小配置：
+
+```yaml
+source:
+  type: fluss
+  bootstrap.servers: localhost:9123
+  table.discoverer.type: fluss-default
+  table.discoverer.pattern: 'inventory\..*'
+  scan.discovery.interval: 10 s
+  scan.startup.mode: earliest
+```
+
+`table.discoverer.type` 用于选择 Source 的表发现器。`fluss-default` 通过
+`table.discoverer.pattern` 匹配全限定表名；选择其他发现器时，需配置其必需的
+`table.discoverer.*` 参数。
 
 How to create Pipeline
 ----------------
@@ -137,6 +157,20 @@ Pipeline Connector Options
 
 * 支持 Fluss 主键表和日志表。
 
+### 动态 Source 订阅
+
+当 Fluss 作为带表发现器的 Source 使用时，每次成功发现的结果都是当前订阅表的完整权威集合。配置正数
+`scan.discovery.interval` 才会周期性更新订阅。空结果会退订全部已发现的表；发现失败不会修改当前订阅，并会使作业失败。
+
+退订表只会停止并清理 Source 侧 reader，不会删除 Fluss 表，也不会改变 Sink 行为。恢复后的 reader 会先等待新的订阅快照，
+再打开 checkpoint 中恢复的 split，因此当前仍处于退订状态的表不会通过恢复的 split 输出记录。表再次被订阅时会作为新表处理，
+并使用配置的 `scan.startup.mode`。
+
+移除与 checkpoint 状态协同：故障恢复时会从最近一次成功 checkpoint 恢复 Source split 和待移除 tombstone；订阅由发现流程刷新。如果退订和重新订阅都发生在
+相邻两次已完成的 checkpoint 之间，故障回滚时可以表现为从未退订过；若人为恢复到移除 tombstone 之前的 checkpoint，
+而该表当前已重新订阅，则不承诺重新开始一个全新的表生命周期。对于主键表，移除时不会提前释放快照 lease，仍由现有的
+过期和关闭逻辑处理。
+
 * 关于自动建表
   * 没有分区键
   * 桶数量由 `bucket.num` 选项控制

diff --git a/docs/content/docs/connectors/pipeline-connectors/fluss.md b/docs/content/docs/connectors/pipeline-connectors/fluss.md
@@ -26,12 +26,33 @@ under the License.
 
 # Fluss Pipeline Connector
 
-The Fluss Pipeline connector can be used as the *Data Sink* of the pipeline, and write data to [Fluss](https://fluss.apache.org). This document describes how to set up the Fluss Pipeline connector.
+The Fluss Pipeline connector can be used as a *Data Source* or *Data Sink* of the pipeline. It
+reads from or writes data to [Fluss](https://fluss.apache.org). This document describes how to set
+up both roles.
 
 ## What can the connector do?
 * Create table automatically if not exist
 * Data synchronization
 * Schema change synchronization (lenient mode)
+* Dynamic source table subscriptions
+
+## Fluss Source
+
+The following is the minimal configuration for reading dynamically discovered Fluss tables:
+
+```yaml
+source:
+  type: fluss
+  bootstrap.servers: localhost:9123
+  table.discoverer.type: fluss-default
+  table.discoverer.pattern: 'inventory\..*'
+  scan.discovery.interval: 10 s
+  scan.startup.mode: earliest
+```
+
+`table.discoverer.type` selects the source table discoverer. `fluss-default` matches fully
+qualified table names with `table.discoverer.pattern`; configure another discoverer's required
+`table.discoverer.*` options when selecting it instead.
 
 How to create Pipeline
 ----------------
@@ -139,6 +160,27 @@ Pipeline Connector Options
 
 * Support Fluss primary key table and log table.
 
+### Dynamic source subscriptions
+
+When Fluss is used as a source with a table discoverer, each successful discovery result is the
+authoritative complete subscription set. Set a positive `scan.discovery.interval` to enable
+periodic updates. An empty result unsubscribes every discovered table; a discovery failure does not
+change the current subscription and fails the job.
+
+Unsubscribing a table only stops and cleans up its source-side readers. It does not delete the Fluss
+table or change sink behavior. A restored reader waits for a fresh subscription snapshot before it
+opens restored splits, so a table that remains unsubscribed cannot emit from restored splits after recovery.
+If a table is subscribed again, it is treated as a new table and uses the configured
+`scan.startup.mode`.
+
+Removal is coordinated with checkpoint state: a failure restores source splits and pending removal
+tombstones from the latest completed checkpoint; subscription is refreshed by discovery. A removal
+and re-addition that both occur between the same two completed checkpoints may be rolled back as
+though the removal had not occurred. Restoring a checkpoint from before a removal tombstone while
+the table is currently re-subscribed does not promise a fresh table lifecycle. For primary key
+tables, snapshot leases are not released early during removal;
+their existing expiry and close handling remain in effect.
+
 * For creating table automatically
   * There is no partition key
   * The number of buckets is controlled by `bucket.num`

diff --git a/...-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java b/...-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java
@@ -30,7 +30,13 @@ public interface ObjectIdDiscoverer<T> extends Serializable, AutoCloseable {
     /** Opens this discoverer and initializes any resources needed for discovery. */
     void open(Context context) throws Exception;
 
-    /** Discovers and returns the set of object identifiers selected by the caller configuration. */
+    /**
+     * Discovers and returns the complete current set of object identifiers selected by the caller
+     * configuration.
+     *
+     * <p>A successful result is authoritative: identifiers absent from it are no longer selected,
+     * and an empty result selects no identifiers. A failed discovery does not describe a new set.
+     */
     Set<T> discover() throws Exception;
 
     /** Closes this discoverer and releases any resources. */

diff --git a/.../src/test/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererITCase.java b/.../src/test/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererITCase.java
@@ -198,8 +198,21 @@ void testDefaultModeReflectsDynamicChanges() throws Exception {
                     "INSERT INTO cdc_subscriptions VALUES "
                             + "('analytics-subscription', 'analytics_db.sessions')");
             try {
-                Set<TableId> updated = discoverer.discover();
-                assertThat(updated)
+                assertThat(discoverer.discover())
+                        .containsExactlyInAnyOrder(
+                                TableId.tableId("analytics_db", "user_events"),
+                                TableId.tableId("analytics_db", "sessions"));
+
+                executeSql(
+                        "DELETE FROM cdc_subscriptions WHERE subscribe_table_name "
+                                + "= 'analytics_db.sessions'");
+                assertThat(discoverer.discover())
+                        .containsExactly(TableId.tableId("analytics_db", "user_events"));
+
+                executeSql(
+                        "INSERT INTO cdc_subscriptions VALUES "
+                                + "('analytics-subscription', 'analytics_db.sessions')");
+                assertThat(discoverer.discover())
                         .containsExactlyInAnyOrder(
                                 TableId.tableId("analytics_db", "user_events"),
                                 TableId.tableId("analytics_db", "sessions"));

diff --git a/...ain/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussDeserializer.java b/...ain/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussDeserializer.java
@@ -58,4 +58,7 @@ public interface FlussDeserializer<T> extends Serializable {
     default List<T> restoreState(TablePath tablePath, int schemaId, RowType rowType) {
         return Collections.emptyList();
     }
+
+    /** Removes all state retained for an unsubscribed table. */
+    default void removeState(TablePath tablePath) {}
 }
diff --git a/...va/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java b/...va/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java
@@ -244,6 +244,16 @@ public List<Event> restoreState(TablePath tablePath, int schemaId, RowType rowTy
         return Collections.emptyList();
     }
 
+    @Override
+    public void removeState(TablePath tablePath) {
+        ensureCacheInitialized();
+        latestSchemaIdCache.remove(tablePath);
+        latestRowTypeCache.remove(tablePath);
+        latestRecordDataGeneratorCache.remove(tablePath);
+        latestFieldConverterCache.remove(tablePath);
+        restoredCreateTableRowTypeCache.remove(tablePath);
+    }
+
     // -------------------------------------------------------------------------
     //  Schema change inference
     // -------------------------------------------------------------------------

diff --git a/...in/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumState.java b/...in/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumState.java
@@ -20,7 +20,9 @@
 import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase;
 
 import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.TablePath;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 
@@ -30,14 +32,24 @@ public class FlussSourceEnumState {
     private final Set<PhysicalTablePath> assignedPhysicalTablePaths;
     private final List<FlussSplitBase> remainingSplits;
     private final String leaseId;
+    private final Set<TablePath> pendingRemovalTablePaths;
 
     public FlussSourceEnumState(
             Set<PhysicalTablePath> assignedPhysicalTablePaths,
             List<FlussSplitBase> remainingSplits,
             String leaseId) {
+        this(assignedPhysicalTablePaths, remainingSplits, leaseId, Collections.emptySet());
+    }
+
+    public FlussSourceEnumState(
+            Set<PhysicalTablePath> assignedPhysicalTablePaths,
+            List<FlussSplitBase> remainingSplits,
+            String leaseId,
+            Set<TablePath> pendingRemovalTablePaths) {
         this.assignedPhysicalTablePaths = assignedPhysicalTablePaths;
         this.remainingSplits = remainingSplits;
         this.leaseId = leaseId;
+        this.pendingRemovalTablePaths = pendingRemovalTablePaths;
     }
 
     public Set<PhysicalTablePath> getAssignedPhysicalTablePaths() {
@@ -51,4 +63,8 @@ public List<FlussSplitBase> getRemainingSplits() {
     public String getLeaseId() {
         return leaseId;
     }
+
+    public Set<TablePath> getPendingRemovalTablePaths() {
+        return pendingRemovalTablePaths;
+    }
 }
diff --git a/...g/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializer.java b/...g/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializer.java
@@ -38,7 +38,7 @@
 public class FlussSourceEnumStateSerializer
         implements SimpleVersionedSerializer<FlussSourceEnumState> {
 
-    private static final int VERSION = 1;
+    private static final int VERSION = 2;
     private final FlussSplitSerializer splitSerializer = new FlussSplitSerializer();
 
     @Override
@@ -71,12 +71,20 @@ public byte[] serialize(FlussSourceEnumState state) throws IOException {
             }
             // Serialize KV snapshot lease ID
             out.writeUTF(state.getLeaseId());
+            out.writeInt(state.getPendingRemovalTablePaths().size());
+            for (TablePath tablePath : state.getPendingRemovalTablePaths()) {
+                out.writeUTF(tablePath.getDatabaseName());
+                out.writeUTF(tablePath.getTableName());
+            }
             return baos.toByteArray();
         }
     }
 
     @Override
     public FlussSourceEnumState deserialize(int version, byte[] serialized) throws IOException {
+        if (version != 1 && version != VERSION) {
+            throw new IOException("Unknown Fluss source enumerator state version: " + version);
+        }
         try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
                 DataInputViewStreamWrapper in = new DataInputViewStreamWrapper(bais)) {
             int pathCount = in.readInt();
@@ -97,7 +105,16 @@ public FlussSourceEnumState deserialize(int version, byte[] serialized) throws I
                 remaining.add(
                         splitSerializer.deserialize(splitSerializer.getVersion(), splitBytes));
             }
-            return new FlussSourceEnumState(assignedPaths, remaining, in.readUTF());
+            String leaseId = in.readUTF();
+            Set<TablePath> pendingRemovalTablePaths = new LinkedHashSet<>();
+            if (version == VERSION) {
+                int pendingRemovalCount = in.readInt();
+                for (int i = 0; i < pendingRemovalCount; i++) {
+                    pendingRemovalTablePaths.add(TablePath.of(in.readUTF(), in.readUTF()));
+                }
+            }
+            return new FlussSourceEnumState(
+                    assignedPaths, remaining, leaseId, pendingRemovalTablePaths);
         }
     }
 }