diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md b/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md index 1ac06c19c11..db09b0e3745 100644 --- 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 index a2d7c532499..bcd22325f12 100644 --- 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/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java index b92fc1393f9..6af18c2da7a 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java @@ -30,7 +30,13 @@ public interface ObjectIdDiscoverer 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. + * + *

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 discover() throws Exception; /** Closes this discoverer and releases any resources. */ diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererITCase.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererITCase.java index b42aa3e7743..55d7c176337 100644 --- a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererITCase.java +++ b/flink-cdc-common/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 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/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussDeserializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussDeserializer.java index f68ae409d18..c16f70c2798 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussDeserializer.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussDeserializer.java @@ -58,4 +58,7 @@ public interface FlussDeserializer extends Serializable { default List 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/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java index b2be7695bce..ecbfbd69370 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java @@ -244,6 +244,16 @@ public List 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/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumState.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumState.java index 00ba1797235..b4e76f16339 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumState.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/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 assignedPhysicalTablePaths; private final List remainingSplits; private final String leaseId; + private final Set pendingRemovalTablePaths; public FlussSourceEnumState( Set assignedPhysicalTablePaths, List remainingSplits, String leaseId) { + this(assignedPhysicalTablePaths, remainingSplits, leaseId, Collections.emptySet()); + } + + public FlussSourceEnumState( + Set assignedPhysicalTablePaths, + List remainingSplits, + String leaseId, + Set pendingRemovalTablePaths) { this.assignedPhysicalTablePaths = assignedPhysicalTablePaths; this.remainingSplits = remainingSplits; this.leaseId = leaseId; + this.pendingRemovalTablePaths = pendingRemovalTablePaths; } public Set getAssignedPhysicalTablePaths() { @@ -51,4 +63,8 @@ public List getRemainingSplits() { public String getLeaseId() { return leaseId; } + + public Set getPendingRemovalTablePaths() { + return pendingRemovalTablePaths; + } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializer.java index bf42fdb67df..35fec9dd006 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializer.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializer.java @@ -38,7 +38,7 @@ public class FlussSourceEnumStateSerializer implements SimpleVersionedSerializer { - 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 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); } } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java index 5ec15b8c066..0d29dd676e1 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java @@ -27,6 +27,8 @@ import org.apache.flink.cdc.common.source.discover.TableDiscovererFactory; import org.apache.flink.cdc.connectors.fluss.source.discover.FlussDefaultDiscoverer; import org.apache.flink.cdc.connectors.fluss.source.event.FinishedKvSnapshotConsumeEvent; +import org.apache.flink.cdc.connectors.fluss.source.event.TableRemovalAckEvent; +import org.apache.flink.cdc.connectors.fluss.source.event.TableSubscriptionEvent; import org.apache.flink.cdc.connectors.fluss.source.reader.LeaseContext; import org.apache.flink.cdc.connectors.fluss.source.split.FlussHybridSnapshotLogSplit; import org.apache.flink.cdc.connectors.fluss.source.split.FlussLogSplit; @@ -60,12 +62,14 @@ import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.NavigableMap; import java.util.OptionalLong; import java.util.Set; import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; /** @@ -96,6 +100,9 @@ public class FlussSourceEnumerator implements SplitEnumerator { private static final Logger LOG = LoggerFactory.getLogger(FlussSourceEnumerator.class); + // Keep request IDs unique across restored enumerator instances so stale acknowledgements cannot + // complete a later removal. + private static final AtomicLong NEXT_REMOVAL_REQUEST_ID = new AtomicLong(); private final SplitEnumeratorContext context; private final TableDiscoverer discoverer; @@ -108,6 +115,17 @@ public class FlussSourceEnumerator private final Set assignedPhysicalTablePaths; private final Map> pendingPartitionSplitAssignment; private final TreeMap> consumedKvSnapshotMap; + private final Set subscribedTablePaths; + private final Set pendingRemovalTablePaths; + private final Map pendingRemovalRequests; + private final Map> removalAcknowledgements; + private final Map pendingRemovalBuckets; + private final Set initializingTablePaths; + private final Map removalFences; + private final Map fencedFreshSplits; + private List lastDiscoveredTableBuckets; + private boolean sentSubscriptionSnapshot; + private final Set seenReaderSubtasks; private volatile boolean checkpointCompletedBefore; @@ -135,6 +153,23 @@ public FlussSourceEnumerator( this.assignedPhysicalTablePaths = assignedPhysicalTablePaths; this.pendingPartitionSplitAssignment = new HashMap<>(); this.consumedKvSnapshotMap = new TreeMap<>(); + this.subscribedTablePaths = new HashSet<>(); + this.pendingRemovalTablePaths = new HashSet<>(); + this.pendingRemovalRequests = new HashMap<>(); + this.removalAcknowledgements = new HashMap<>(); + this.pendingRemovalBuckets = new HashMap<>(); + this.initializingTablePaths = new HashSet<>(); + this.removalFences = new HashMap<>(); + this.fencedFreshSplits = new HashMap<>(); + this.lastDiscoveredTableBuckets = Collections.emptyList(); + this.sentSubscriptionSnapshot = false; + this.seenReaderSubtasks = new HashSet<>(); + assignedPhysicalTablePaths.stream() + .map(PhysicalTablePath::getTablePath) + .forEach(subscribedTablePaths::add); + remainingSplits.stream() + .map(FlussSplitBase::getTablePath) + .forEach(subscribedTablePaths::add); this.checkpointCompletedBefore = checkpointCompletedBefore; addPartitionSplitChangeToPendingAssignments(remainingSplits); } @@ -160,6 +195,11 @@ public FlussSourceEnumerator( new LeaseContext( restoredState.getLeaseId(), leaseContext.getKvSnapshotLeaseDurationMs()), true); + pendingRemovalTablePaths.addAll(restoredState.getPendingRemovalTablePaths()); + pendingRemovalTablePaths.forEach( + tablePath -> + pendingRemovalRequests.put( + tablePath, NEXT_REMOVAL_REQUEST_ID.incrementAndGet())); } @Override @@ -202,13 +242,13 @@ public void start() { * * @return the full list of discovered table-bucket entries. */ - private List getSubscribedTableBuckets() throws Exception { + private DiscoveryResult getSubscribedTableBuckets() throws Exception { List allBuckets = new ArrayList<>(); Set discoveredTableIds = discoverer.discover(); Set subscribedPaths = discoveredTableIds.stream() .map(FlussDefaultDiscoverer::toTablePath) - .collect(Collectors.toCollection(java.util.LinkedHashSet::new)); + .collect(Collectors.toCollection(LinkedHashSet::new)); for (TablePath tablePath : subscribedPaths) { TableInfo tableInfo = admin.getTableInfo(tablePath).get(); @@ -239,7 +279,7 @@ private List getSubscribedTableBuckets() throws Exception { } } } - return allBuckets; + return new DiscoveryResult(subscribedPaths, allBuckets); } // ------------------------------------------------------------------------- @@ -250,11 +290,23 @@ private List getSubscribedTableBuckets() throws Exception { * Compares the discovered table-buckets against assigned and pending {@link PhysicalTablePath}s * and triggers split creation for newly discovered table-buckets. */ - private void checkTableBucketChanges(List allBuckets, Throwable error) { + private void checkTableBucketChanges(DiscoveryResult discoveryResult, Throwable error) { if (error != null) { throw new FlinkRuntimeException("Failed to discover subscribed table-buckets.", error); } + boolean subscriptionChanged = updateSubscriptions(discoveryResult.subscribedTablePaths); + lastDiscoveredTableBuckets = discoveryResult.tableBuckets; + if (!sentSubscriptionSnapshot || subscriptionChanged) { + sendSubscriptionSnapshot(); + sentSubscriptionSnapshot = true; + } + + initializeNewTableBuckets(discoveryResult.tableBuckets); + } + + private void initializeNewTableBuckets(List allBuckets) { + Set assignedOrPendingPhysicalTablePaths = new HashSet<>(assignedPhysicalTablePaths); pendingPartitionSplitAssignment.values().stream() @@ -264,7 +316,9 @@ private void checkTableBucketChanges(List allBuckets, Throwable List newBuckets = new ArrayList<>(); for (TableBucketInfo info : allBuckets) { - if (!assignedOrPendingPhysicalTablePaths.contains(info.physicalTablePath)) { + if (!initializingTablePaths.contains(info.physicalTablePath.getTablePath()) + && !pendingRemovalTablePaths.contains(info.physicalTablePath.getTablePath()) + && !assignedOrPendingPhysicalTablePaths.contains(info.physicalTablePath)) { newBuckets.add(info); } } @@ -275,8 +329,86 @@ private void checkTableBucketChanges(List allBuckets, Throwable } LOG.info("Discovered {} new table-bucket(s) to initialize.", newBuckets.size()); + Set initializingPaths = + newBuckets.stream() + .map(info -> info.physicalTablePath.getTablePath()) + .collect(Collectors.toSet()); + initializingTablePaths.addAll(initializingPaths); context.callAsync( - () -> initPendingBucketSplits(newBuckets), this::handleTableBucketChanges); + () -> initPendingBucketSplits(newBuckets), + (splits, error) -> handleTableBucketChanges(splits, error, initializingPaths)); + } + + private boolean updateSubscriptions(Set currentSubscribedTablePaths) { + Set removedTablePaths = new HashSet<>(subscribedTablePaths); + removedTablePaths.removeAll(currentSubscribedTablePaths); + boolean changed = !subscribedTablePaths.equals(currentSubscribedTablePaths); + for (TablePath tablePath : removedTablePaths) { + removeConsumedKvSnapshotBuckets(tablePath); + pendingRemovalTablePaths.add(tablePath); + pendingRemovalRequests.put(tablePath, NEXT_REMOVAL_REQUEST_ID.incrementAndGet()); + removalAcknowledgements.remove(tablePath); + assignedPhysicalTablePaths.removeIf( + physicalTablePath -> physicalTablePath.getTablePath().equals(tablePath)); + pendingPartitionSplitAssignment + .values() + .forEach( + splits -> + splits.removeIf( + split -> split.getTablePath().equals(tablePath))); + fencedFreshSplits + .entrySet() + .removeIf(entry -> entry.getValue().getTablePath().equals(tablePath)); + LOG.warn( + "Removing table {} from the source. KV snapshot leases are retained until their existing expiry or close handling.", + tablePath); + } + subscribedTablePaths.clear(); + subscribedTablePaths.addAll(currentSubscribedTablePaths); + return changed; + } + + private void sendSubscriptionSnapshot() { + TableSubscriptionEvent event = + new TableSubscriptionEvent( + subscribedTablePaths, pendingRemovalRequests, removalFences.keySet()); + for (int subtaskId : context.registeredReaders().keySet()) { + context.sendEventToSourceReader(subtaskId, event); + } + } + + private void maybeClearRemovalTombstones() { + Set expectedReaders = new HashSet<>(); + for (int subtaskId = 0; subtaskId < context.currentParallelism(); subtaskId++) { + expectedReaders.add(subtaskId); + } + if (!context.registeredReaders().keySet().containsAll(expectedReaders)) { + return; + } + Set clearedTablePaths = new HashSet<>(); + for (TablePath tablePath : pendingRemovalTablePaths) { + if (!initializingTablePaths.contains(tablePath) + && removalAcknowledgements + .getOrDefault(tablePath, Collections.emptySet()) + .containsAll(expectedReaders)) { + clearedTablePaths.add(tablePath); + } + } + if (clearedTablePaths.isEmpty()) { + return; + } + pendingRemovalTablePaths.removeAll(clearedTablePaths); + clearedTablePaths.forEach( + tablePath -> { + removalFences.put(tablePath, -1L); + pendingRemovalRequests.remove(tablePath); + removalAcknowledgements.remove(tablePath); + pendingRemovalBuckets + .entrySet() + .removeIf(entry -> entry.getValue().equals(tablePath)); + }); + sendSubscriptionSnapshot(); + initializeNewTableBuckets(lastDiscoveredTableBuckets); } // ------------------------------------------------------------------------- @@ -500,8 +632,21 @@ private static void validateBucketOffsets( * Receives newly created splits, records their {@link PhysicalTablePath}s as assigned, and * distributes the splits to registered readers. */ - private void handleTableBucketChanges(List newSplits, Throwable error) { + private void handleTableBucketChanges( + List newSplits, Throwable error, Set initializingPaths) { + initializingTablePaths.removeAll(initializingPaths); if (error != null) { + boolean hasActiveInitializingPath = + initializingPaths.stream() + .anyMatch( + tablePath -> + subscribedTablePaths.contains(tablePath) + && !pendingRemovalTablePaths.contains( + tablePath)); + if (!hasActiveInitializingPath) { + maybeClearRemovalTombstones(); + return; + } throw new FlinkRuntimeException( "Failed to initialize splits for new table-buckets.", error); } @@ -509,9 +654,19 @@ private void handleTableBucketChanges(List newSplits, Throwable if (newSplits.isEmpty()) { throw new FlinkRuntimeException("No splits were created for discovered table-buckets."); } - - addPartitionSplitChangeToPendingAssignments(newSplits); - assignPendingPartitionSplits(context.registeredReaders().keySet()); + List effectiveNewSplits = + newSplits.stream() + .filter( + split -> + subscribedTablePaths.contains(split.getTablePath()) + && !pendingRemovalTablePaths.contains( + split.getTablePath())) + .collect(Collectors.toList()); + if (!effectiveNewSplits.isEmpty()) { + addPartitionSplitChangeToPendingAssignments(effectiveNewSplits); + assignPendingPartitionSplits(context.registeredReaders().keySet()); + } + maybeClearRemovalTombstones(); } // ------------------------------------------------------------------------- @@ -549,6 +704,9 @@ private void assignPendingPartitionSplits(Set pendingReaders) { // Mark pending partitions as already assigned pendingAssignmentForReader.forEach( split -> { + if (removalFences.containsKey(split.getTablePath())) { + fencedFreshSplits.put(split.splitId(), split); + } assignedPhysicalTablePaths.add(split.getPhysicalTablePath()); }); } @@ -594,16 +752,46 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) { event.getCheckpointId()); event.getTableBuckets() .forEach( - tableBucket -> + tableBucket -> { + TablePath removedTablePath = pendingRemovalBuckets.get(tableBucket); + if (removedTablePath == null + || !pendingRemovalTablePaths.contains(removedTablePath)) { addConsumedKvSnapshotBucket( - event.getCheckpointId(), tableBucket)); + event.getCheckpointId(), tableBucket); + } + }); + } else if (sourceEvent instanceof TableRemovalAckEvent) { + TableRemovalAckEvent event = (TableRemovalAckEvent) sourceEvent; + event.getCompletedRemovalRequests() + .forEach( + (tablePath, requestId) -> { + Long expectedRequestId = pendingRemovalRequests.get(tablePath); + if (expectedRequestId != null + && expectedRequestId.equals(requestId)) { + removalAcknowledgements + .computeIfAbsent(tablePath, ignored -> new HashSet<>()) + .add(subtaskId); + } + }); + maybeClearRemovalTombstones(); } } @Override public void addSplitsBack(List splits, int subtaskId) { LOG.info("Adding {} splits back from subtask {}", splits.size(), subtaskId); - addPartitionSplitChangeToPendingAssignments(splits); + addPartitionSplitChangeToPendingAssignments( + splits.stream() + .filter( + split -> + subscribedTablePaths.contains(split.getTablePath()) + && !pendingRemovalTablePaths.contains( + split.getTablePath()) + && (!removalFences.containsKey(split.getTablePath()) + || split.equals( + fencedFreshSplits.get( + split.splitId())))) + .collect(Collectors.toList())); // If the failed subtask has already restarted, we need to assign pending splits to it if (context.registeredReaders().containsKey(subtaskId)) { assignPendingPartitionSplits(Collections.singleton(subtaskId)); @@ -613,20 +801,53 @@ public void addSplitsBack(List splits, int subtaskId) { @Override public void addReader(int subtaskId) { LOG.info("Reader {} added, assigning pending splits.", subtaskId); + boolean restartedWithPendingRemoval = + !seenReaderSubtasks.add(subtaskId) + && !pendingRemovalTablePaths.isEmpty() + && sentSubscriptionSnapshot; + if (restartedWithPendingRemoval) { + pendingRemovalTablePaths.forEach( + tablePath -> + pendingRemovalRequests.put( + tablePath, NEXT_REMOVAL_REQUEST_ID.incrementAndGet())); + removalAcknowledgements.clear(); + sendSubscriptionSnapshot(); + } + removalAcknowledgements + .values() + .forEach(acknowledgements -> acknowledgements.remove(subtaskId)); + if (sentSubscriptionSnapshot && !restartedWithPendingRemoval) { + context.sendEventToSourceReader( + subtaskId, + new TableSubscriptionEvent( + subscribedTablePaths, pendingRemovalRequests, removalFences.keySet())); + } assignPendingPartitionSplits(Collections.singleton(subtaskId)); } @Override public FlussSourceEnumState snapshotState(long checkpointId) throws Exception { + removalFences.replaceAll( + (tablePath, fenceCheckpointId) -> + fenceCheckpointId < 0 ? checkpointId : fenceCheckpointId); List remainingSplits = new ArrayList<>(); pendingPartitionSplitAssignment.forEach((reader, splits) -> remainingSplits.addAll(splits)); return new FlussSourceEnumState( - assignedPhysicalTablePaths, remainingSplits, leaseContext.getKvSnapshotLeaseId()); + assignedPhysicalTablePaths, + remainingSplits, + leaseContext.getKvSnapshotLeaseId(), + pendingRemovalTablePaths); } @Override public void notifyCheckpointComplete(long checkpointId) throws Exception { checkpointCompletedBefore = true; + removalFences + .entrySet() + .removeIf(entry -> entry.getValue() >= 0 && entry.getValue() <= checkpointId); + fencedFreshSplits + .entrySet() + .removeIf(entry -> !removalFences.containsKey(entry.getValue().getTablePath())); Set consumedKvSnapshots = getAndRemoveConsumedKvSnapshotBucketsBefore(checkpointId); @@ -709,6 +930,26 @@ private Set getAndRemoveConsumedKvSnapshotBucketsBefore(long checkp return tableBuckets; } + private void removeConsumedKvSnapshotBuckets(TablePath tablePath) { + Set removedBuckets = + lastDiscoveredTableBuckets.stream() + .filter(info -> info.physicalTablePath.getTablePath().equals(tablePath)) + .map(info -> info.tableBucket) + .collect(Collectors.toSet()); + if (removedBuckets.isEmpty()) { + return; + } + removedBuckets.forEach(bucket -> pendingRemovalBuckets.put(bucket, tablePath)); + consumedKvSnapshotMap.values().forEach(buckets -> buckets.removeAll(removedBuckets)); + consumedKvSnapshotMap.entrySet().removeIf(entry -> entry.getValue().isEmpty()); + } + + Set pendingKvSnapshotBucketsForTesting() { + return consumedKvSnapshotMap.values().stream() + .flatMap(Set::stream) + .collect(Collectors.toSet()); + } + private void maybeDropKvSnapshotLease() throws Exception { if (admin != null && offsetsInitializer instanceof SnapshotOffsetsInitializer @@ -770,6 +1011,17 @@ static Configuration toSourceConfig(org.apache.fluss.config.Configuration flussC return Configuration.fromMap(map); } + /** Complete successful subscription snapshot and its resolved table buckets. */ + private static class DiscoveryResult { + final Set subscribedTablePaths; + final List tableBuckets; + + DiscoveryResult(Set subscribedTablePaths, List tableBuckets) { + this.subscribedTablePaths = subscribedTablePaths; + this.tableBuckets = tableBuckets; + } + } + /** Container for a discovered table-bucket with its {@link PhysicalTablePath}. */ private static class TableBucketInfo { final PhysicalTablePath physicalTablePath; diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/event/TableRemovalAckEvent.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/event/TableRemovalAckEvent.java new file mode 100644 index 00000000000..51f9f0f4a0b --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/event/TableRemovalAckEvent.java @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.fluss.source.event; + +import org.apache.flink.api.connector.source.SourceEvent; + +import org.apache.fluss.metadata.TablePath; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +/** Acknowledges completed table-removal requests from a source reader. */ +public class TableRemovalAckEvent implements SourceEvent { + + private static final long serialVersionUID = 1L; + + private final Map completedRemovalRequests; + + public TableRemovalAckEvent(Map completedRemovalRequests) { + this.completedRemovalRequests = + Collections.unmodifiableMap(new LinkedHashMap<>(completedRemovalRequests)); + } + + public Map getCompletedRemovalRequests() { + return completedRemovalRequests; + } +} diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/event/TableSubscriptionEvent.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/event/TableSubscriptionEvent.java new file mode 100644 index 00000000000..e0589889667 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/event/TableSubscriptionEvent.java @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.fluss.source.event; + +import org.apache.flink.api.connector.source.SourceEvent; + +import org.apache.fluss.metadata.TablePath; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; + +/** Authoritative source-table subscription snapshot sent from the enumerator to a reader. */ +public class TableSubscriptionEvent implements SourceEvent { + + private static final long serialVersionUID = 1L; + + private final Set subscribedTablePaths; + private final Map pendingRemovalRequests; + private final Set fencedTablePaths; + + public TableSubscriptionEvent( + Set subscribedTablePaths, Map pendingRemovalRequests) { + this(subscribedTablePaths, pendingRemovalRequests, Collections.emptySet()); + } + + public TableSubscriptionEvent( + Set subscribedTablePaths, + Map pendingRemovalRequests, + Set fencedTablePaths) { + this.subscribedTablePaths = + Collections.unmodifiableSet(new LinkedHashSet<>(subscribedTablePaths)); + this.pendingRemovalRequests = + Collections.unmodifiableMap(new LinkedHashMap<>(pendingRemovalRequests)); + this.fencedTablePaths = Collections.unmodifiableSet(new LinkedHashSet<>(fencedTablePaths)); + } + + public Set getSubscribedTablePaths() { + return subscribedTablePaths; + } + + public Map getPendingRemovalRequests() { + return pendingRemovalRequests; + } + + public Set getFencedTablePaths() { + return fencedTablePaths; + } +} diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussRecordEmitter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussRecordEmitter.java index 448e74777ed..8bf4e8f2c2f 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussRecordEmitter.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussRecordEmitter.java @@ -31,8 +31,10 @@ import java.util.ArrayList; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; /** * A {@link RecordEmitter} that uses a {@link FlussDeserializer} to convert {@link @@ -62,6 +64,8 @@ public class FlussRecordEmitter implements RecordEmitter> pendingTableEvents = new HashMap<>(); + private final Set inactiveTablePaths = new HashSet<>(); + public FlussRecordEmitter(FlussDeserializer deserializer) { this.deserializer = deserializer; } @@ -72,6 +76,9 @@ public void emitRecord( throws Exception { // Emit pending events for this table before processing the actual record. TablePath tablePath = element.getTablePath(); + if (inactiveTablePaths.contains(tablePath)) { + return; + } List pendingEvents = pendingTableEvents.remove(tablePath); if (pendingEvents != null) { for (T event : pendingEvents) { @@ -141,6 +148,7 @@ private void updateSchemaTracking(FlussSplitState splitState, FlussSourceRecord * return pending events for emission on the first record. */ public void applySplit(FlussSplitBase split) { + inactiveTablePaths.remove(split.getTablePath()); if (split.getSchemaId() != null && split.getRowType() != null) { TablePath tablePath = split.getTablePath(); List pendingEvents = @@ -156,4 +164,11 @@ public void applySplit(FlussSplitBase split) { } } } + + /** Stops emitting a table and discards its restored schema state. */ + public void removeTable(TablePath tablePath) { + inactiveTablePaths.add(tablePath); + pendingTableEvents.remove(tablePath); + deserializer.removeState(tablePath); + } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceFetcherManager.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceFetcherManager.java new file mode 100644 index 00000000000..6cac7d0238d --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceFetcherManager.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.fluss.source.reader; + +import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase; +import org.apache.flink.cdc.source.SingleThreadFetcherManagerAdapter; +import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; +import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher; +import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherTask; +import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue; + +import org.apache.fluss.metadata.TablePath; + +import java.io.IOException; +import java.util.Set; +import java.util.function.Supplier; + +/** Executes table removal on the connector's single fetcher thread. */ +class FlussSourceFetcherManager + extends SingleThreadFetcherManagerAdapter { + + FlussSourceFetcherManager( + FutureCompletingBlockingQueue> elementsQueue, + Supplier splitReaderSupplier) { + super(elementsQueue, splitReaderSupplier::get); + } + + void removeTables(Set tablePaths) { + if (tablePaths.isEmpty()) { + return; + } + SplitFetcher fetcher = getRunningFetcher(); + if (fetcher == null) { + fetcher = createSplitFetcher(); + enqueueRemovalTask(fetcher, tablePaths); + startFetcher(fetcher); + } else { + enqueueRemovalTask(fetcher, tablePaths); + ((FlussSplitReader) fetcher.getSplitReader()).wakeUp(); + } + } + + private void enqueueRemovalTask( + SplitFetcher fetcher, Set tablePaths) { + FlussSplitReader splitReader = (FlussSplitReader) fetcher.getSplitReader(); + fetcher.enqueueTask( + new SplitFetcherTask() { + @Override + public boolean run() throws IOException { + splitReader.removeTables(tablePaths); + return true; + } + + @Override + public void wakeUp() {} + }); + } +} diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReader.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReader.java index 35e1641d786..78a5a58586f 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReader.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReader.java @@ -17,15 +17,17 @@ package org.apache.flink.cdc.connectors.fluss.source.reader; +import org.apache.flink.api.connector.source.SourceEvent; import org.apache.flink.api.connector.source.SourceReaderContext; import org.apache.flink.cdc.connectors.fluss.sink.v2.metrics.WrapperFlussMetricRegistry; import org.apache.flink.cdc.connectors.fluss.source.event.FinishedKvSnapshotConsumeEvent; +import org.apache.flink.cdc.connectors.fluss.source.event.TableRemovalAckEvent; +import org.apache.flink.cdc.connectors.fluss.source.event.TableSubscriptionEvent; import org.apache.flink.cdc.connectors.fluss.source.metrics.FlussSourceReaderMetrics; import org.apache.flink.cdc.connectors.fluss.source.split.FlussHybridSnapshotLogSplitState; import org.apache.flink.cdc.connectors.fluss.source.split.FlussLogSplitState; import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase; import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitState; -import org.apache.flink.cdc.source.SingleThreadFetcherManagerAdapter; import org.apache.flink.cdc.source.SingleThreadMultiplexSourceReaderBaseAdapter; import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; @@ -33,13 +35,17 @@ import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue; import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TablePath; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.stream.Collectors; /** * A generic {@link org.apache.flink.api.connector.source.SourceReader} for Fluss, built on top of @@ -62,6 +68,12 @@ public class FlussSourceReader private final WrapperFlussMetricRegistry metricRegistry; private final SourceReaderContext readerContext; private final Set reportedFinishedSnapshotBuckets; + private final FlussSourceFetcherManager fetcherManager; + private final List stagedSplits; + private final Map activeSplits; + private final Map pendingRemovalRequests; + private final Set subscribedTablePaths; + private boolean receivedSubscriptionSnapshot; public FlussSourceReader( FutureCompletingBlockingQueue> elementsQueue, @@ -70,20 +82,34 @@ public FlussSourceReader( WrapperFlussMetricRegistry metricRegistry, FlussSourceReaderMetrics sourceReaderMetrics, FlussRecordEmitter recordEmitter) { - super( + this( elementsQueue, - new SingleThreadFetcherManagerAdapter( + readerContext, + metricRegistry, + recordEmitter, + new FlussSourceFetcherManager( elementsQueue, () -> new FlussSplitReader( - flussConfig, metricRegistry, sourceReaderMetrics)), - recordEmitter, - new Configuration(), - readerContext); + flussConfig, metricRegistry, sourceReaderMetrics))); + } + + FlussSourceReader( + FutureCompletingBlockingQueue> elementsQueue, + SourceReaderContext readerContext, + WrapperFlussMetricRegistry metricRegistry, + FlussRecordEmitter recordEmitter, + FlussSourceFetcherManager fetcherManager) { + super(elementsQueue, fetcherManager, recordEmitter, new Configuration(), readerContext); this.recordEmitter = recordEmitter; this.metricRegistry = metricRegistry; this.readerContext = readerContext; this.reportedFinishedSnapshotBuckets = new HashSet<>(); + this.fetcherManager = fetcherManager; + this.stagedSplits = new ArrayList<>(); + this.activeSplits = new HashMap<>(); + this.pendingRemovalRequests = new HashMap<>(); + this.subscribedTablePaths = new HashSet<>(); } @Override @@ -99,6 +125,7 @@ public void close() throws Exception { protected FlussSplitState initializedState(FlussSplitBase split) { // Restore deserializer schema caches from the recovered split (like MySQL's applySplit) recordEmitter.applySplit(split); + activeSplits.put(split.splitId(), split); if (split.isHybridSnapshotLogSplit()) { return new FlussHybridSnapshotLogSplitState(split.asHybridSnapshotLogSplit()); } else if (split.isLogSplit()) { @@ -135,12 +162,103 @@ public List snapshotState(long checkpointId) { new FinishedKvSnapshotConsumeEvent(checkpointId, finishedSnapshotBuckets)); reportedFinishedSnapshotBuckets.addAll(finishedSnapshotBuckets); } + splits.addAll(stagedSplits); return splits; } + @Override + public void addSplits(List splits) { + if (!receivedSubscriptionSnapshot) { + stagedSplits.addAll(splits); + return; + } + activateSplits(splits); + } + + @Override + public void handleSourceEvents(SourceEvent sourceEvent) { + if (!(sourceEvent instanceof TableSubscriptionEvent)) { + super.handleSourceEvents(sourceEvent); + return; + } + + TableSubscriptionEvent event = (TableSubscriptionEvent) sourceEvent; + boolean firstSubscriptionSnapshot = !receivedSubscriptionSnapshot; + receivedSubscriptionSnapshot = true; + subscribedTablePaths.clear(); + subscribedTablePaths.addAll(event.getSubscribedTablePaths()); + pendingRemovalRequests.clear(); + pendingRemovalRequests.putAll(event.getPendingRemovalRequests()); + + Set activeRemovalTablePaths = + activeSplits.values().stream() + .filter(split -> !isActive(split.getTablePath())) + .map(FlussSplitBase::getTablePath) + .collect(Collectors.toSet()); + activeSplits.values().stream() + .filter(split -> activeRemovalTablePaths.contains(split.getTablePath())) + .map(FlussSplitBase::getTableBucket) + .forEach(reportedFinishedSnapshotBuckets::remove); + + for (TablePath tablePath : activeRemovalTablePaths) { + recordEmitter.removeTable(tablePath); + } + if (!activeRemovalTablePaths.isEmpty()) { + fetcherManager.removeTables(activeRemovalTablePaths); + } + + List effectiveStagedSplits = new ArrayList<>(); + for (FlussSplitBase split : stagedSplits) { + if (isActive(split.getTablePath()) + && (!firstSubscriptionSnapshot + || !event.getFencedTablePaths().contains(split.getTablePath()))) { + effectiveStagedSplits.add(split); + } else { + reportedFinishedSnapshotBuckets.remove(split.getTableBucket()); + } + } + stagedSplits.clear(); + activateSplits(effectiveStagedSplits); + acknowledgeCompletedRemovals(); + } + @Override protected void onSplitFinished(Map finishedSplitIds) { // Fluss source is continuous and unbounded; splits should not normally finish. LOG.info("Splits finished: {}", finishedSplitIds.keySet()); + finishedSplitIds.keySet().forEach(activeSplits::remove); + acknowledgeCompletedRemovals(); + } + + private void activateSplits(List splits) { + List activeSplits = new ArrayList<>(); + for (FlussSplitBase split : splits) { + if (isActive(split.getTablePath())) { + activeSplits.add(split); + } + } + if (!activeSplits.isEmpty()) { + super.addSplits(activeSplits); + } + } + + private boolean isActive(TablePath tablePath) { + return subscribedTablePaths.contains(tablePath) + && !pendingRemovalRequests.containsKey(tablePath); + } + + private void acknowledgeCompletedRemovals() { + Map completed = new HashMap<>(); + for (Map.Entry entry : pendingRemovalRequests.entrySet()) { + if (activeSplits.values().stream() + .noneMatch(split -> split.getTablePath().equals(entry.getKey())) + && stagedSplits.stream() + .noneMatch(split -> split.getTablePath().equals(entry.getKey()))) { + completed.put(entry.getKey(), entry.getValue()); + } + } + if (!completed.isEmpty()) { + readerContext.sendSourceEventToCoordinator(new TableRemovalAckEvent(completed)); + } } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReader.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReader.java index 3bbcdc77ac1..e52c42f0163 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReader.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReader.java @@ -57,9 +57,13 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Queue; +import java.util.Set; /** * A {@link SplitReader} implementation for Fluss. It reads change log records from Fluss log @@ -86,10 +90,11 @@ public class FlussSplitReader implements SplitReader boundedSplits; + private final Set removedSplitIds; @Nullable private FlussSplitBase currentBoundedSplit; @Nullable private BatchScanner currentBatchScanner; @Nullable private Integer currentBatchSchemaId; - @Nullable private MultiTableLogScanner currentLogScanner; + @Nullable private volatile MultiTableLogScanner currentLogScanner; private long snapshotRecordsToSkip; private long currentReadRecordsCount; @@ -106,12 +111,19 @@ public FlussSplitReader( this.tablePartitionKeyNames = new HashMap<>(); this.bucketToSplit = new HashMap<>(); this.boundedSplits = new ArrayDeque<>(); + this.removedSplitIds = new LinkedHashSet<>(); } @Override public RecordsWithSplitIds fetch() throws IOException { RecordsBySplits.Builder builder = new RecordsBySplits.Builder<>(); + if (!removedSplitIds.isEmpty()) { + removedSplitIds.forEach(builder::addFinishedSplit); + removedSplitIds.clear(); + return builder.build(); + } + // Priority: read bounded (snapshot) splits first, then log checkSnapshotSplitOrStartNext(); if (currentBatchScanner != null) { @@ -395,8 +407,67 @@ private static RowType schemaToRowType(org.apache.fluss.metadata.Schema schema) return new RowType(fields); } + /** Removes all reader-side resources for the specified logical tables. */ + void removeTables(Set tablePaths) throws IOException { + Set removedBuckets = new HashSet<>(); + for (Iterator> iterator = + bucketToSplit.entrySet().iterator(); + iterator.hasNext(); ) { + Map.Entry entry = iterator.next(); + FlussSplitBase split = entry.getValue(); + if (!tablePaths.contains(split.getTablePath())) { + continue; + } + removedBuckets.add(entry.getKey()); + removedSplitIds.add(split.splitId()); + iterator.remove(); + if (currentLogScanner != null) { + if (entry.getKey().getPartitionId() == null) { + currentLogScanner.unsubscribe(split.getTablePath(), entry.getKey().getBucket()); + } else { + currentLogScanner.unsubscribe( + split.getTablePath(), + entry.getKey().getPartitionId(), + entry.getKey().getBucket()); + } + } + } + for (Iterator iterator = boundedSplits.iterator(); iterator.hasNext(); ) { + FlussSplitBase split = iterator.next(); + if (tablePaths.contains(split.getTablePath())) { + removedSplitIds.add(split.splitId()); + iterator.remove(); + } + } + if (currentBoundedSplit != null + && tablePaths.contains(currentBoundedSplit.getTablePath())) { + removedSplitIds.add(currentBoundedSplit.splitId()); + closeCurrentBoundedSplit(); + } + for (TablePath tablePath : tablePaths) { + Table table = tables.remove(tablePath); + if (table != null) { + try { + table.close(); + } catch (Exception e) { + throw new IOException("Failed to close table " + tablePath, e); + } + } + tableRowTypes.remove(tablePath); + tablePrimaryKeyNames.remove(tablePath); + tablePartitionKeyNames.remove(tablePath); + } + if (!removedBuckets.isEmpty()) { + LOG.info("Removed Fluss source table buckets {}.", removedBuckets); + } + } + @Override - public void wakeUp() {} + public void wakeUp() { + if (currentLogScanner != null) { + currentLogScanner.wakeup(); + } + } @Override public void close() throws Exception { diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/FlussSourcePipelineITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/FlussSourcePipelineITCase.java index b122b570703..5170a17a65a 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/FlussSourcePipelineITCase.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/FlussSourcePipelineITCase.java @@ -73,6 +73,8 @@ import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import java.util.regex.Pattern; import java.util.stream.Collectors; import static org.apache.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS; @@ -750,6 +752,299 @@ void testNewTableDiscoveryViaSubscriptionTable() throws Exception { "+I[2, b2]"); } + @Test + void testSubscriptionRemovalStopsTableAndReaddStartsNewLifecycle() throws Exception { + String subscriptionTable = "subscription_removal_list"; + String tableA = "subscription_removal_a"; + String tableB = "subscription_removal_b"; + createSubscriptionTables(subscriptionTable, tableA, tableB); + + FlussSource source = + createFlussSourceWithTableSubscriber( + DATABASE_NAME + "." + subscriptionTable, "earliest", Duration.ofSeconds(1)); + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + CloseableIterator iter = + env.fromSource( + source, + WatermarkStrategy.noWatermarks(), + "FlussSource", + new EventTypeInfo()) + .executeAndCollect("SubscriptionRemovalTest"); + List events = Collections.synchronizedList(new ArrayList<>()); + Thread collector = startCollector(iter, events, "subscription-removal"); + try { + awaitEvents( + events, + collected -> + eventsForTable(collected, tableA).size() >= 2 + && eventsForTable(collected, tableB).size() >= 2, + COLLECT_TIMEOUT); + int initialEventCount = snapshotEvents(events).size(); + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, tableB)) + .await(); + Thread.sleep(Duration.ofSeconds(4).toMillis()); + tBatchEnv.executeSql(String.format("INSERT INTO %s VALUES (2, 'a2')", tableA)).await(); + tBatchEnv.executeSql(String.format("INSERT INTO %s VALUES (2, 'b2')", tableB)).await(); + + awaitEvents( + events, + collected -> + convertToStringList( + eventsForTable(collected, tableA), + DataTypes.INT(), + DataTypes.STRING()) + .contains("+I[2, a2]"), + COLLECT_TIMEOUT); + Thread.sleep(Duration.ofSeconds(2).toMillis()); + List removalSnapshot = snapshotEvents(events); + List afterRemoval = + removalSnapshot.subList(initialEventCount, removalSnapshot.size()); + assertThat(convertToStringList(afterRemoval, DataTypes.INT(), DataTypes.STRING())) + .contains("+I[2, a2]") + .doesNotContain("+I[2, b2]"); + + int readdBaseline = snapshotEvents(events).size(); + tBatchEnv + .executeSql( + String.format( + "INSERT INTO %s VALUES ('%s.%s')", + subscriptionTable, DATABASE_NAME, tableB)) + .await(); + Thread.sleep(Duration.ofSeconds(4).toMillis()); + tBatchEnv.executeSql(String.format("INSERT INTO %s VALUES (3, 'b3')", tableB)).await(); + + awaitEvents( + events, + collected -> + hasCreateTableEvent(eventsForTable(collected, tableB)) + && convertToStringList( + eventsForTable(collected, tableB), + DataTypes.INT(), + DataTypes.STRING()) + .contains("+I[3, b3]"), + COLLECT_TIMEOUT); + List readdSnapshot = snapshotEvents(events); + List readdedEvents = readdSnapshot.subList(readdBaseline, readdSnapshot.size()); + assertThat(readdedEvents).filteredOn(CreateTableEvent.class::isInstance).hasSize(1); + assertThat(convertToStringList(readdedEvents, DataTypes.INT(), DataTypes.STRING())) + .contains("+I[1, subscription_removal_b1]", "+I[2, b2]", "+I[3, b3]"); + } finally { + iter.close(); + collector.join(5000); + } + } + + @Test + void testSavepointRestoreDoesNotResubscribeRemovedTable(@TempDir Path tmpDir) throws Exception { + String subscriptionTable = "subscription_savepoint_remove_list"; + String tableA = "subscription_savepoint_remove_a"; + String tableB = "subscription_savepoint_remove_b"; + createSubscriptionTables(subscriptionTable, tableA, tableB); + + JobClient jobClient = + startSubscriptionSource( + DATABASE_NAME + "." + subscriptionTable, + "SubscriptionSavepointRemovalPhase1"); + try { + // A1 and B1 must be consumed and checkpointed before the subscription changes. + Thread.sleep(Duration.ofSeconds(10).toMillis()); + String savepointPath = + jobClient + .stopWithSavepoint( + false, + tmpDir.toAbsolutePath().toString(), + SavepointFormatType.CANONICAL) + .get(); + jobClient = null; + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, tableB)) + .await(); + tBatchEnv.executeSql(String.format("INSERT INTO %s VALUES (2, 'a2')", tableA)).await(); + tBatchEnv.executeSql(String.format("INSERT INTO %s VALUES (2, 'b2')", tableB)).await(); + + try (RestoredSubscription restored = + startRestoredSubscription( + savepointPath, + DATABASE_NAME + "." + subscriptionTable, + "SubscriptionSavepointRemovalPhase2")) { + List events = restored.events; + awaitEvents( + events, + collected -> + convertToStringList( + eventsForTable(collected, tableA), + DataTypes.INT(), + DataTypes.STRING()) + .contains("+I[2, a2]"), + COLLECT_TIMEOUT); + Thread.sleep(Duration.ofSeconds(4).toMillis()); + List restoredEvents = snapshotEvents(events); + assertThat(eventsForTable(restoredEvents, tableB)).isEmpty(); + } + } finally { + if (jobClient != null) { + jobClient.cancel().get(); + } + } + } + + @Test + void testSavepointRestoreReaddedTableStartsNewEarliestLifecycle(@TempDir Path tmpDir) + throws Exception { + String subscriptionTable = "subscription_savepoint_readd_list"; + String tableA = "subscription_savepoint_readd_a"; + String tableB = "subscription_savepoint_readd_b"; + createSubscriptionTables(subscriptionTable, tableA, tableB); + + JobClient jobClient = + startSubscriptionSource( + DATABASE_NAME + "." + subscriptionTable, + "SubscriptionSavepointReaddPhase1"); + try { + // B1 must be consumed and checkpointed before removal distinguishes a new lifecycle. + Thread.sleep(Duration.ofSeconds(10).toMillis()); + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, tableB)) + .await(); + // Wait through multiple discovery cycles before capturing the removal state. + Thread.sleep(Duration.ofSeconds(4).toMillis()); + String savepointPath = + jobClient + .stopWithSavepoint( + false, + tmpDir.toAbsolutePath().toString(), + SavepointFormatType.CANONICAL) + .get(); + jobClient = null; + tBatchEnv + .executeSql( + String.format( + "INSERT INTO %s VALUES ('%s.%s')", + subscriptionTable, DATABASE_NAME, tableB)) + .await(); + + try (RestoredSubscription restored = + startRestoredSubscription( + savepointPath, + DATABASE_NAME + "." + subscriptionTable, + "SubscriptionSavepointReaddPhase2")) { + List events = restored.events; + awaitEvents( + events, + collected -> + hasCreateTableEvent(eventsForTable(collected, tableB)) + && convertToStringList( + eventsForTable(collected, tableB), + DataTypes.INT(), + DataTypes.STRING()) + .contains("+I[1, " + tableB + "1]"), + COLLECT_TIMEOUT); + tBatchEnv + .executeSql(String.format("INSERT INTO %s VALUES (2, 'b2')", tableB)) + .await(); + awaitEvents( + events, + collected -> + convertToStringList( + eventsForTable(collected, tableB), + DataTypes.INT(), + DataTypes.STRING()) + .contains("+I[2, b2]"), + COLLECT_TIMEOUT); + List restoredEvents = snapshotEvents(events); + List readdedEvents = eventsForTable(restoredEvents, tableB); + assertThat(readdedEvents).filteredOn(CreateTableEvent.class::isInstance).hasSize(1); + assertThat(convertToStringList(readdedEvents, DataTypes.INT(), DataTypes.STRING())) + .contains("+I[1, " + tableB + "1]", "+I[2, b2]"); + } + } finally { + if (jobClient != null) { + jobClient.cancel().get(); + } + } + } + + @Test + void testSameCheckpointWindowRemovalAndReaddRollsBackAsContinuation(@TempDir Path tmpDir) + throws Exception { + String subscriptionTable = "subscription_savepoint_rollback_list"; + String tableB = "subscription_savepoint_rollback_b"; + createSubscriptionTables(subscriptionTable, tableB); + + JobClient jobClient = + startSubscriptionSource( + DATABASE_NAME + "." + subscriptionTable, + "SubscriptionSavepointRollbackPhase1"); + try { + // B1 must be consumed and checkpointed in N before the same-window changes. + Thread.sleep(Duration.ofSeconds(10).toMillis()); + String savepointPath = + jobClient + .stopWithSavepoint( + false, + tmpDir.toAbsolutePath().toString(), + SavepointFormatType.CANONICAL) + .get(); + jobClient = null; + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, tableB)) + .await(); + tBatchEnv + .executeSql( + String.format( + "INSERT INTO %s VALUES ('%s.%s')", + subscriptionTable, DATABASE_NAME, tableB)) + .await(); + tBatchEnv.executeSql(String.format("INSERT INTO %s VALUES (2, 'b2')", tableB)).await(); + + try (RestoredSubscription restored = + startRestoredSubscription( + savepointPath, + DATABASE_NAME + "." + subscriptionTable, + "SubscriptionSavepointRollbackPhase2")) { + List events = restored.events; + awaitEvents( + events, + collected -> + convertToStringList( + eventsForTable(collected, tableB), + DataTypes.INT(), + DataTypes.STRING()) + .contains("+I[2, b2]"), + COLLECT_TIMEOUT); + List restoredData = + convertToStringList( + eventsForTable(snapshotEvents(events), tableB), + DataTypes.INT(), + DataTypes.STRING()); + assertThat(restoredData) + .contains("+I[2, b2]") + .doesNotContain("+I[1, " + tableB + "1]"); + } + } finally { + if (jobClient != null) { + jobClient.cancel().get(); + } + } + } + @Test void testNewPartitionDiscovery() throws Exception { String tableName = "part_discover_table"; @@ -883,6 +1178,113 @@ void testAddColumnSchemaEvolution() throws Exception { // ======================== Helper methods ======================== + private JobClient startSubscriptionSource(String subscriptionTableFqn, String jobName) + throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + env.enableCheckpointing(200); + env.fromSource( + createFlussSourceWithTableSubscriber( + subscriptionTableFqn, "earliest", Duration.ofSeconds(1)), + WatermarkStrategy.noWatermarks(), + "FlussSource", + new EventTypeInfo()) + .uid("fluss-source") + .sinkTo(new DiscardingSink<>()) + .uid("discard-sink"); + return env.executeAsync(jobName); + } + + private RestoredSubscription startRestoredSubscription( + String savepointPath, String subscriptionTableFqn, String jobName) throws Exception { + org.apache.flink.configuration.Configuration restoreConf = + new org.apache.flink.configuration.Configuration(); + restoreConf.setString("execution.savepoint.path", savepointPath); + StreamExecutionEnvironment restoreEnv = + StreamExecutionEnvironment.getExecutionEnvironment(restoreConf); + restoreEnv.setParallelism(1); + restoreEnv.enableCheckpointing(200); + CloseableIterator iter = + restoreEnv + .fromSource( + createFlussSourceWithTableSubscriber( + subscriptionTableFqn, "earliest", Duration.ofSeconds(1)), + WatermarkStrategy.noWatermarks(), + "FlussSource", + new EventTypeInfo()) + .uid("fluss-source") + .executeAndCollect(jobName); + List events = Collections.synchronizedList(new ArrayList<>()); + return new RestoredSubscription(iter, events, startCollector(iter, events, jobName)); + } + + private static final class RestoredSubscription implements AutoCloseable { + + private final CloseableIterator iter; + private final List events; + private final Thread collector; + + private RestoredSubscription( + CloseableIterator iter, List events, Thread collector) { + this.iter = iter; + this.events = events; + this.collector = collector; + } + + @Override + public void close() throws Exception { + iter.close(); + collector.join(5000); + } + } + + private void createSubscriptionTables(String subscriptionTable, String... tableNames) + throws Exception { + tBatchEnv + .executeSql( + String.format( + "CREATE TABLE %s (table_name STRING, PRIMARY KEY (table_name) NOT ENFORCED)", + subscriptionTable)) + .await(); + for (String tableName : tableNames) { + tBatchEnv + .executeSql( + String.format( + "CREATE TABLE %s (id INT, val STRING, PRIMARY KEY (id) NOT ENFORCED)", + tableName)) + .await(); + tBatchEnv + .executeSql( + String.format("INSERT INTO %s VALUES (1, '%s1')", tableName, tableName)) + .await(); + tBatchEnv + .executeSql( + String.format( + "INSERT INTO %s VALUES ('%s.%s')", + subscriptionTable, DATABASE_NAME, tableName)) + .await(); + } + } + + private static Thread startCollector( + CloseableIterator iter, List events, String collectorName) { + Thread collector = + new Thread( + () -> { + try { + while (iter.hasNext()) { + events.add(iter.next()); + } + } catch (Exception ignored) { + // Iterator close terminates the collector. + } + }, + collectorName + "-collector"); + collector.setDaemon(true); + collector.start(); + return collector; + } + private FlussSource createFlussSource( String database, String tablePattern, String startupMode) { return createFlussSourceWithDiscoveryInterval( @@ -968,7 +1370,7 @@ private FlussSource createFlussSourceWithDiscoveryInterval( * tablePattern is translated to regex {@code .*}. */ private static String toFqnRegex(String database, String tablePattern) { - return java.util.regex.Pattern.quote(database) + "\\." + tablePattern.replace("*", ".*"); + return Pattern.quote(database) + "\\." + tablePattern.replace("*", ".*"); } /** @@ -1010,6 +1412,33 @@ private List collectAllEvents( return collectAllEvents(source, expectedCount, timeout, 2); } + private static void awaitEvents( + List events, Predicate> condition, Duration timeout) + throws InterruptedException { + long deadline = System.nanoTime() + timeout.toNanos(); + while (!condition.test(snapshotEvents(events)) && System.nanoTime() < deadline) { + Thread.sleep(50L); + } + assertThat(condition.test(snapshotEvents(events))).isTrue(); + } + + private static List snapshotEvents(List events) { + synchronized (events) { + return new ArrayList<>(events); + } + } + + private static List eventsForTable(List events, String tableName) { + return events.stream() + .filter(ChangeEvent.class::isInstance) + .filter(event -> ((ChangeEvent) event).tableId().getTableName().equals(tableName)) + .collect(Collectors.toList()); + } + + private static boolean hasCreateTableEvent(List events) { + return events.stream().anyMatch(CreateTableEvent.class::isInstance); + } + private List collectAllEvents( FlussSource source, int expectedCount, Duration timeout, int parallelism) throws Exception { diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializerTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializerTest.java index 0cdd52c0192..749d7230ac2 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializerTest.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumStateSerializerTest.java @@ -19,21 +19,25 @@ import org.apache.flink.cdc.connectors.fluss.source.split.FlussLogSplit; import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase; +import org.apache.flink.core.memory.DataOutputViewStreamWrapper; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; import org.junit.jupiter.api.Test; +import java.io.ByteArrayOutputStream; +import java.io.IOException; import java.util.Collections; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link FlussSourceEnumStateSerializer}. */ class FlussSourceEnumStateSerializerTest { @Test - void testSerializeLeaseIdInVersionOneState() throws Exception { + void testSerializeStateInVersionTwo() throws Exception { PhysicalTablePath physicalTablePath = PhysicalTablePath.of(TablePath.of("database", "table")); FlussSplitBase split = new FlussLogSplit(physicalTablePath, new TableBucket(1L, 0), 42L); @@ -47,9 +51,58 @@ void testSerializeLeaseIdInVersionOneState() throws Exception { FlussSourceEnumState restored = serializer.deserialize(serializer.getVersion(), serializer.serialize(state)); - assertThat(serializer.getVersion()).isEqualTo(1); + assertThat(serializer.getVersion()).isEqualTo(2); assertThat(restored.getAssignedPhysicalTablePaths()).containsExactly(physicalTablePath); assertThat(restored.getRemainingSplits()).containsExactly(split); assertThat(restored.getLeaseId()).isEqualTo("lease-id"); + assertThat(restored.getPendingRemovalTablePaths()).isEmpty(); + } + + @Test + void testSerializePendingRemovalTablePathsInVersionTwoState() throws Exception { + PhysicalTablePath physicalTablePath = + PhysicalTablePath.of(TablePath.of("database", "table")); + TablePath pendingRemovalTablePath = TablePath.of("database", "removed_table"); + FlussSplitBase split = new FlussLogSplit(physicalTablePath, new TableBucket(1L, 0), 42L); + FlussSourceEnumState state = + new FlussSourceEnumState( + Collections.singleton(physicalTablePath), + Collections.singletonList(split), + "lease-id", + Collections.singleton(pendingRemovalTablePath)); + FlussSourceEnumStateSerializer serializer = new FlussSourceEnumStateSerializer(); + + FlussSourceEnumState restored = + serializer.deserialize(serializer.getVersion(), serializer.serialize(state)); + + assertThat(serializer.getVersion()).isEqualTo(2); + assertThat(restored.getPendingRemovalTablePaths()).containsExactly(pendingRemovalTablePath); + } + + @Test + void testDeserializeVersionOneStateWithoutPendingRemovalTablePaths() throws Exception { + FlussSourceEnumStateSerializer serializer = new FlussSourceEnumStateSerializer(); + byte[] versionOneState; + try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(bytes)) { + out.writeInt(0); + out.writeInt(0); + out.writeUTF("lease-id"); + versionOneState = bytes.toByteArray(); + } + + FlussSourceEnumState restored = serializer.deserialize(1, versionOneState); + + assertThat(restored.getLeaseId()).isEqualTo("lease-id"); + assertThat(restored.getPendingRemovalTablePaths()).isEmpty(); + } + + @Test + void testRejectUnknownStateVersion() { + FlussSourceEnumStateSerializer serializer = new FlussSourceEnumStateSerializer(); + + assertThatThrownBy(() -> serializer.deserialize(3, new byte[0])) + .isInstanceOf(IOException.class) + .hasMessageContaining("Unknown Fluss source enumerator state version"); } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumeratorTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumeratorTest.java index aa447657b52..3fe9653a414 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumeratorTest.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumeratorTest.java @@ -25,8 +25,12 @@ import org.apache.flink.cdc.common.source.discover.TableDiscoverer; import org.apache.flink.cdc.connectors.fluss.source.discover.FlussDefaultDiscoverer; import org.apache.flink.cdc.connectors.fluss.source.discover.FlussSubscriberTableDiscoverer; +import org.apache.flink.cdc.connectors.fluss.source.event.FinishedKvSnapshotConsumeEvent; +import org.apache.flink.cdc.connectors.fluss.source.event.TableRemovalAckEvent; +import org.apache.flink.cdc.connectors.fluss.source.event.TableSubscriptionEvent; import org.apache.flink.cdc.connectors.fluss.source.reader.LeaseContext; import org.apache.flink.cdc.connectors.fluss.source.split.FlussHybridSnapshotLogSplit; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussLogSplit; import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableEnvironment; @@ -39,9 +43,12 @@ import org.apache.fluss.client.table.scanner.batch.BatchScanUtils; import org.apache.fluss.client.table.scanner.batch.BatchScanner; import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.row.InternalRow; import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.types.DataTypes; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -55,11 +62,13 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; +import java.util.regex.Pattern; import java.util.stream.Collectors; import static org.apache.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS; import static org.apache.fluss.server.testutils.FlussClusterExtension.BUILTIN_DATABASE; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** @@ -192,13 +201,8 @@ void testPatternSubscriberDiscoversNewTableDynamically() throws Throwable { } } - /** - * Tests that when a subscribed table is dropped, the enumerator does not emit any new - * assignments on the following discovery cycle (current enumerator intentionally does not - * revoke already-assigned tables). - */ @Test - void testPatternSubscriberIgnoresTableRemoval() throws Throwable { + void testPatternSubscriberRemovesDroppedTableAfterReaderAcknowledgements() throws Throwable { String tableA = "rm_a"; String tableB = "rm_b"; createPkTable(tableA); @@ -214,21 +218,257 @@ void testPatternSubscriberIgnoresTableRemoval() throws Throwable { enumerator.start(); registerAllReaders(context, enumerator); - // First cycle: both tables assigned. runDiscoveryCycle(context); assertThat(assignedTableNames(context)).containsExactlyInAnyOrder(tableA, tableB); int assignmentsAfterFirst = context.getSplitsAssignmentSequence().size(); - // Drop tableB — pattern no longer matches it. tBatchEnv.executeSql(String.format("DROP TABLE %s", tableB)).await(); + runDiscoveryCycle(context); + + assertThat(context.getSplitsAssignmentSequence()).hasSize(assignmentsAfterFirst); + TablePath removedTablePath = TablePath.of(DATABASE_NAME, tableB); + TableSubscriptionEvent removal = latestSubscriptionEvent(context, 0); + assertThat(removal.getSubscribedTablePaths()) + .containsExactly(TablePath.of(DATABASE_NAME, tableA)); + assertThat(removal.getPendingRemovalRequests()).containsOnlyKeys(removedTablePath); + long requestId = removal.getPendingRemovalRequests().get(removedTablePath); + assertThat(enumerator.snapshotState(1L).getPendingRemovalTablePaths()) + .containsExactly(removedTablePath); + + acknowledgeRemoval(enumerator, removedTablePath, requestId); + assertThat(enumerator.snapshotState(2L).getPendingRemovalTablePaths()).isEmpty(); + assertThat(latestSubscriptionEvent(context, 0).getSubscribedTablePaths()) + .containsExactly(TablePath.of(DATABASE_NAME, tableA)); + } finally { + enumerator.close(); + } + } + } + + @Test + void testRemovalFenceSurvivesAbortedCheckpointAndRearm() throws Throwable { + String tableA = "fence_a"; + String tableB = "fence_b"; + createPkTable(tableA); + createPkTable(tableB); + + FlussDefaultDiscoverer discoverer = new FlussDefaultDiscoverer(); + String pattern = fqnRegex(DATABASE_NAME, "fence_.*"); - // Second cycle: no new assignments should be emitted. The enumerator keeps its - // previously-assigned state (no revocation support yet). + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = newEnumerator(context, discoverer, pattern); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + runDiscoveryCycle(context); + + TablePath tablePath = TablePath.of(DATABASE_NAME, tableB); + tBatchEnv.executeSql(String.format("DROP TABLE %s", tableB)).await(); + runDiscoveryCycle(context); + long firstRequestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + acknowledgeRemoval(enumerator, tablePath, firstRequestId); + assertThat(latestSubscriptionEvent(context, 0).getFencedTablePaths()) + .contains(tablePath); + enumerator.snapshotState(1L); + enumerator.addReader(0); + assertThat(latestSubscriptionEvent(context, 0).getFencedTablePaths()) + .contains(tablePath); + enumerator.snapshotState(2L); + enumerator.notifyCheckpointComplete(2L); + enumerator.addReader(0); + assertThat(latestSubscriptionEvent(context, 0).getFencedTablePaths()) + .doesNotContain(tablePath); + + createPkTable(tableB); runDiscoveryCycle(context); + tBatchEnv.executeSql(String.format("DROP TABLE %s", tableB)).await(); + runDiscoveryCycle(context); + long secondRequestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + acknowledgeRemoval(enumerator, tablePath, secondRequestId); + assertThat(latestSubscriptionEvent(context, 0).getFencedTablePaths()) + .contains(tablePath); + enumerator.snapshotState(3L); + + createPkTable(tableB); + runDiscoveryCycle(context); + tBatchEnv.executeSql(String.format("DROP TABLE %s", tableB)).await(); + runDiscoveryCycle(context); + long thirdRequestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + acknowledgeRemoval(enumerator, tablePath, thirdRequestId); + + enumerator.notifyCheckpointComplete(3L); + enumerator.addReader(0); + assertThat(latestSubscriptionEvent(context, 0).getFencedTablePaths()) + .contains(tablePath); + + enumerator.snapshotState(4L); + enumerator.notifyCheckpointComplete(4L); + enumerator.addReader(0); + assertThat(latestSubscriptionEvent(context, 0).getFencedTablePaths()) + .doesNotContain(tablePath); + } finally { + enumerator.close(); + } + } + } + @Test + void testFencedAddSplitsBackDropsStaleSplitWithoutReassigningOtherBuckets() throws Throwable { + String subscriptionTable = "sub_fenced_failed_split"; + String targetTable = "tgt_fenced_failed_split"; + createSubscriptionTable(subscriptionTable); + createPkTable(targetTable, 2); + insertSubscription(subscriptionTable, targetTable); + + FlussSubscriberTableDiscoverer subscriber = + new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = newEnumerator(context, subscriber, null); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + runDiscoveryCycle(context); + + TablePath tablePath = TablePath.of(DATABASE_NAME, targetTable); + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, targetTable)) + .await(); + runDiscoveryCycle(context); + long requestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + acknowledgeRemoval(enumerator, tablePath, requestId); + + insertSubscription(subscriptionTable, targetTable); + int assignmentsBeforeReadd = context.getSplitsAssignmentSequence().size(); + runDiscoveryCycle(context); + List freshSplits = + context + .getSplitsAssignmentSequence() + .subList( + assignmentsBeforeReadd, + context.getSplitsAssignmentSequence().size()) + .stream() + .flatMap(assignment -> assignment.assignment().values().stream()) + .flatMap(List::stream) + .collect(Collectors.toList()); + assertThat(freshSplits).hasSize(2); + FlussLogSplit freshB0 = + freshSplits.stream() + .filter( + split -> + FlussSourceEnumerator.getSplitOwner( + split.getTableBucket(), NUM_READERS) + == 0) + .map(FlussSplitBase::asLogSplit) + .findFirst() + .orElseThrow(AssertionError::new); + assertThat( + freshSplits.stream() + .filter( + split -> + FlussSourceEnumerator.getSplitOwner( + split.getTableBucket(), + NUM_READERS) + == 1)) + .singleElement(); + + FlussLogSplit staleB0 = + new FlussLogSplit( + freshB0.getPhysicalTablePath(), + freshB0.getTableBucket(), + freshB0.getStartingOffset() + 1); + int assignmentsBeforeFailedReader = context.getSplitsAssignmentSequence().size(); + enumerator.addSplitsBack(List.of(staleB0, freshB0), 0); + + assertThat(context.getSplitsAssignmentSequence()) + .hasSize(assignmentsBeforeFailedReader + 1); + SplitsAssignment failedReaderAssignment = + context.getSplitsAssignmentSequence() + .get(context.getSplitsAssignmentSequence().size() - 1); + assertThat(failedReaderAssignment.assignment()).containsOnlyKeys(0); + assertThat(failedReaderAssignment.assignment().get(0)).containsExactly(freshB0); + + int assignmentsBeforeNextDiscovery = context.getSplitsAssignmentSequence().size(); + runDiscoveryCycle(context); assertThat(context.getSplitsAssignmentSequence()) - .as("Shrinking subscription should not emit new assignments") - .hasSize(assignmentsAfterFirst); + .hasSize(assignmentsBeforeNextDiscovery); + } finally { + enumerator.close(); + } + } + } + + @Test + void testDroppedPartitionDoesNotCreateTableRemovalTombstone() throws Throwable { + String tableName = "partition_still_subscribed"; + tBatchEnv + .executeSql( + String.format( + "CREATE TABLE %s (id INT, ds STRING, val STRING, " + + "PRIMARY KEY (id, ds) NOT ENFORCED) PARTITIONED BY (ds)", + tableName)) + .await(); + tBatchEnv + .executeSql( + String.format( + "INSERT INTO %s VALUES (1, '20260904', 'first'), (2, '20260905', 'second')", + tableName)) + .await(); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS); + Connection connection = + ConnectionFactory.createConnection( + FLUSS_CLUSTER_EXTENSION.getClientConfig())) { + FlussSourceEnumerator enumerator = + newEnumerator( + context, + new FlussDefaultDiscoverer(), + fqnRegex(DATABASE_NAME, tableName)); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + runDiscoveryCycle(context); + + TablePath tablePath = TablePath.of(DATABASE_NAME, tableName); + connection + .getAdmin() + .dropPartition( + tablePath, + new PartitionSpec(Collections.singletonMap("ds", "20260904")), + false) + .get(); + runDiscoveryCycle(context); + + assertThat(enumerator.snapshotState(1L).getPendingRemovalTablePaths()).isEmpty(); + TableSubscriptionEvent subscription = latestSubscriptionEvent(context, 0); + assertThat(subscription.getSubscribedTablePaths()).containsExactly(tablePath); + assertThat(subscription.getPendingRemovalRequests()).isEmpty(); + + tBatchEnv.executeSql(String.format("DROP TABLE %s", tableName)).await(); + runDiscoveryCycle(context); + + TableSubscriptionEvent removal = latestSubscriptionEvent(context, 0); + assertThat(enumerator.snapshotState(2L).getPendingRemovalTablePaths()) + .containsExactly(tablePath); + assertThat(removal.getPendingRemovalRequests()).containsKey(tablePath); } finally { enumerator.close(); } @@ -645,21 +885,13 @@ void testFlussTableSubscriberDynamicallyAddsTable() throws Throwable { } } - /** - * Tests that removing a row from the subscription table does NOT cause the enumerator to emit - * new assignments or revoke any splits on the next discovery cycle — the current enumerator - * intentionally does not revoke already-assigned tables. - */ @Test - void testFlussTableSubscriberIgnoresSubscriptionShrinkage() throws Throwable { - String subscriptionTable = "sub_shrink"; - String targetA = "tgt_shrink_a"; - String targetB = "tgt_shrink_b"; + void testSubscriptionDeletionPersistsTombstoneUntilEveryReaderAcknowledges() throws Throwable { + String subscriptionTable = "sub_delete"; + String targetTable = "tgt_delete"; createSubscriptionTable(subscriptionTable); - createPkTable(targetA); - createPkTable(targetB); - insertSubscription(subscriptionTable, targetA); - insertSubscription(subscriptionTable, targetB); + createPkTable(targetTable); + insertSubscription(subscriptionTable, targetTable); FlussSubscriberTableDiscoverer subscriber = new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); @@ -671,26 +903,474 @@ void testFlussTableSubscriberIgnoresSubscriptionShrinkage() throws Throwable { enumerator.start(); registerAllReaders(context, enumerator); - // First cycle: both tables assigned. runDiscoveryCycle(context); - assertThat(assignedTableNames(context)).containsExactlyInAnyOrder(targetA, targetB); + assertThat(assignedTableNames(context)).containsExactly(targetTable); int assignmentsAfterFirst = context.getSplitsAssignmentSequence().size(); + FlussSplitBase assignedSplit = + context.getSplitsAssignmentSequence().get(0).assignment().values().stream() + .flatMap(List::stream) + .findFirst() + .orElseThrow(AssertionError::new); + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, targetTable)) + .await(); - // Shrink the subscription by dropping & recreating the subscription table with - // only targetA. (Using DROP+CREATE avoids relying on SQL DELETE support and still - // reflects a valid subscription-shrinkage scenario.) - tBatchEnv.executeSql(String.format("DROP TABLE %s", subscriptionTable)).await(); - createSubscriptionTable(subscriptionTable); - insertSubscription(subscriptionTable, targetA); + runDiscoveryCycle(context); - // Second cycle: no new assignments should be emitted; previously-assigned state - // remains stable (the enumerator does not revoke already-assigned tables). + assertThat(context.getSplitsAssignmentSequence()).hasSize(assignmentsAfterFirst); + TablePath tablePath = TablePath.of(DATABASE_NAME, targetTable); + TableSubscriptionEvent removal = latestSubscriptionEvent(context, 0); + assertThat(removal.getSubscribedTablePaths()).isEmpty(); + assertThat(removal.getPendingRemovalRequests()).containsOnlyKeys(tablePath); + long firstRequestId = removal.getPendingRemovalRequests().get(tablePath); + assertThat(enumerator.snapshotState(1L).getPendingRemovalTablePaths()) + .containsExactly(tablePath); + + enumerator.addSplitsBack( + Collections.singletonList(assignedSplit), + FlussSourceEnumerator.getSplitOwner( + assignedSplit.getTableBucket(), NUM_READERS)); + assertThat(context.getSplitsAssignmentSequence()).hasSize(assignmentsAfterFirst); + + enumerator.handleSourceEvent( + 0, + new TableRemovalAckEvent( + Collections.singletonMap(tablePath, firstRequestId - 1))); + assertThat(enumerator.snapshotState(2L).getPendingRemovalTablePaths()) + .containsExactly(tablePath); + + enumerator.addReader(0); + long restartedRequestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + assertThat(restartedRequestId).isNotEqualTo(firstRequestId); + + insertSubscription(subscriptionTable, targetTable); runDiscoveryCycle(context); + assertThat(context.getSplitsAssignmentSequence()).hasSize(assignmentsAfterFirst); + assertThat(latestSubscriptionEvent(context, 0).getSubscribedTablePaths()) + .containsExactly(tablePath); + assertThat(latestSubscriptionEvent(context, 0).getPendingRemovalRequests()) + .containsOnlyKeys(tablePath); + + acknowledgeRemoval(enumerator, tablePath, firstRequestId); + assertThat(enumerator.snapshotState(3L).getPendingRemovalTablePaths()) + .containsExactly(tablePath); + + acknowledgeRemoval(enumerator, tablePath, restartedRequestId); + assertThat(enumerator.snapshotState(4L).getPendingRemovalTablePaths()).isEmpty(); + context.runNextOneTimeCallable(); + assertThat(context.getSplitsAssignmentSequence()) + .hasSize(assignmentsAfterFirst + 1); + assertThat(latestAssignmentTableNames(context)).containsExactly(targetTable); + } finally { + enumerator.close(); + } + } + } + @Test + void testRemovalWaitsForRegisteredReadersAndDropsLateFailedSplit() throws Throwable { + String subscriptionTable = "sub_reader_restart"; + String targetTable = "tgt_reader_restart"; + createSubscriptionTable(subscriptionTable); + createPkTable(targetTable); + insertSubscription(subscriptionTable, targetTable); + + FlussSubscriberTableDiscoverer subscriber = + new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = newEnumerator(context, subscriber, null); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + runDiscoveryCycle(context); + FlussSplitBase failedSplit = + context.getSplitsAssignmentSequence().get(0).assignment().values().stream() + .flatMap(List::stream) + .findFirst() + .orElseThrow(AssertionError::new); + TablePath tablePath = TablePath.of(DATABASE_NAME, targetTable); + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, targetTable)) + .await(); + runDiscoveryCycle(context); + long oldRequestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + + enumerator.handleSourceEvent( + 0, + new TableRemovalAckEvent( + Collections.singletonMap(tablePath, oldRequestId))); + context.unregisterReader(0); + enumerator.handleSourceEvent( + 1, + new TableRemovalAckEvent( + Collections.singletonMap(tablePath, oldRequestId))); + assertThat(enumerator.snapshotState(1L).getPendingRemovalTablePaths()) + .containsExactly(tablePath); + + context.registerReader(new ReaderInfo(0, "restarted_0")); + enumerator.addReader(0); + long restartedRequestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + assertThat(restartedRequestId).isNotEqualTo(oldRequestId); + + acknowledgeRemoval(enumerator, tablePath, oldRequestId); + assertThat(enumerator.snapshotState(2L).getPendingRemovalTablePaths()) + .containsExactly(tablePath); + + acknowledgeRemoval(enumerator, tablePath, restartedRequestId); + assertThat(enumerator.snapshotState(3L).getPendingRemovalTablePaths()).isEmpty(); + + int assignmentsBeforeLateSplit = context.getSplitsAssignmentSequence().size(); + enumerator.addSplitsBack( + Collections.singletonList(failedSplit), + FlussSourceEnumerator.getSplitOwner( + failedSplit.getTableBucket(), NUM_READERS)); assertThat(context.getSplitsAssignmentSequence()) - .as("Shrinking subscription should not emit new assignments") - .hasSize(assignmentsAfterFirst); - assertThat(assignedTableNames(context)).containsExactlyInAnyOrder(targetA, targetB); + .hasSize(assignmentsBeforeLateSplit); + FlussSourceEnumState state = enumerator.snapshotState(4L); + assertThat(state.getAssignedPhysicalTablePaths()) + .noneMatch(path -> path.getTablePath().equals(tablePath)); + assertThat(state.getRemainingSplits()) + .noneMatch(split -> split.getTablePath().equals(tablePath)); + } finally { + enumerator.close(); + } + } + } + + @Test + void testLateInitializationDoesNotReviveDeletedSubscription() throws Throwable { + String subscriptionTable = "sub_late_init"; + String targetTable = "tgt_late_init"; + createSubscriptionTable(subscriptionTable); + createPkTable(targetTable); + insertSubscription(subscriptionTable, targetTable); + + FlussSubscriberTableDiscoverer subscriber = + new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = newEnumerator(context, subscriber, null); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + + context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX); + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, targetTable)) + .await(); + context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX); + context.runNextOneTimeCallable(); + + assertThat(context.getSplitsAssignmentSequence()).isEmpty(); + assertThat(enumerator.snapshotState(1L).getPendingRemovalTablePaths()) + .containsExactly(TablePath.of(DATABASE_NAME, targetTable)); + } finally { + enumerator.close(); + } + } + } + + @Test + void testLateFailedInitializationAfterRemovalAcknowledgementIsIgnored() throws Throwable { + String subscriptionTable = "sub_late_failed_init"; + String targetTable = "tgt_late_failed_init"; + createSubscriptionTable(subscriptionTable); + createPkTable(targetTable); + insertSubscription(subscriptionTable, targetTable); + + RuntimeException offsetFailure = new RuntimeException("Injected late offset failure"); + OffsetsInitializer failingOffsetsInitializer = + (partitionName, bucketIds, retriever) -> { + throw offsetFailure; + }; + FlussSubscriberTableDiscoverer subscriber = + new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = + newEnumerator( + context, + subscriber, + null, + failingOffsetsInitializer, + DISCOVERY_INTERVAL_MS); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX); + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, targetTable)) + .await(); + context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX); + + TablePath tablePath = TablePath.of(DATABASE_NAME, targetTable); + long requestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + acknowledgeRemoval(enumerator, tablePath, requestId); + + assertThatCode(context::runNextOneTimeCallable).doesNotThrowAnyException(); + assertThat(context.getSplitsAssignmentSequence()).isEmpty(); + assertThat(enumerator.snapshotState(1L).getPendingRemovalTablePaths()).isEmpty(); + } finally { + enumerator.close(); + } + } + } + + @Test + void testRestoreRegeneratesRemovalRequestIdAndRejectsOldAcknowledgement() throws Throwable { + String subscriptionTable = "sub_restore"; + String targetTable = "tgt_restore"; + TablePath tablePath = TablePath.of(DATABASE_NAME, targetTable); + createSubscriptionTable(subscriptionTable); + createPkTable(targetTable); + insertSubscription(subscriptionTable, targetTable); + + FlussSubscriberTableDiscoverer subscriber = + new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); + FlussSourceEnumState restoredState; + long oldRequestId; + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = newEnumerator(context, subscriber, null); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + runDiscoveryCycle(context); + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, targetTable)) + .await(); + runDiscoveryCycle(context); + oldRequestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + restoredState = enumerator.snapshotState(1L); + } finally { + enumerator.close(); + } + } + + insertSubscription(subscriptionTable, targetTable); + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = + newRestoredEnumerator( + context, + new FlussSubscriberTableDiscoverer( + DATABASE_NAME + "." + subscriptionTable, 100), + restoredState); + try { + enumerator.start(); + runDiscoveryCycle(context); + registerAllReaders(context, enumerator); + + long restoredRequestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(tablePath); + assertThat(restoredRequestId).isNotEqualTo(oldRequestId); + assertThat(enumerator.snapshotState(2L).getPendingRemovalTablePaths()) + .containsExactly(tablePath); + + acknowledgeRemoval(enumerator, tablePath, oldRequestId); + assertThat(enumerator.snapshotState(3L).getPendingRemovalTablePaths()) + .containsExactly(tablePath); + } finally { + enumerator.close(); + } + } + } + + @Test + void testRestartedReaderWaitsForFreshDiscoveryBeforeReceivingRemovalSnapshot() + throws Throwable { + String subscriptionTable = "sub_restore_fresh_discovery"; + String targetTable = "tgt_restore_fresh_discovery"; + createSubscriptionTable(subscriptionTable); + createPkTable(targetTable); + insertSubscription(subscriptionTable, targetTable); + + FlussSubscriberTableDiscoverer subscriber = + new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); + FlussSourceEnumState restoredState; + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = newEnumerator(context, subscriber, null); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + runDiscoveryCycle(context); + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, targetTable)) + .await(); + runDiscoveryCycle(context); + restoredState = enumerator.snapshotState(1L); + } finally { + enumerator.close(); + } + } + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = + newRestoredEnumerator( + context, + new FlussSubscriberTableDiscoverer( + DATABASE_NAME + "." + subscriptionTable, 100), + restoredState); + try { + enumerator.start(); + context.registerReader(new ReaderInfo(0, "loc_0")); + enumerator.addReader(0); + context.unregisterReader(0); + context.registerReader(new ReaderInfo(0, "loc_0_restarted")); + enumerator.addReader(0); + + assertThat(context.getSentSourceEvent()).doesNotContainKey(0); + + context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX); + + assertThat(latestSubscriptionEvent(context, 0).getPendingRemovalRequests()) + .containsKey(TablePath.of(DATABASE_NAME, targetTable)); + } finally { + enumerator.close(); + } + } + } + + @Test + void testMixedInitializationCallbackClearsAcknowledgedRemovalTombstone() throws Throwable { + String subscriptionTable = "sub_mixed_init"; + String removedTable = "tgt_mixed_removed"; + String retainedTable = "tgt_mixed_retained"; + createSubscriptionTable(subscriptionTable); + createPkTable(removedTable); + createPkTable(retainedTable); + insertSubscription(subscriptionTable, removedTable); + insertSubscription(subscriptionTable, retainedTable); + + FlussSubscriberTableDiscoverer subscriber = + new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = newEnumerator(context, subscriber, null); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + + context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX); + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, removedTable)) + .await(); + context.runPeriodicCallable(DISCOVERY_CALLABLE_INDEX); + + TablePath removedTablePath = TablePath.of(DATABASE_NAME, removedTable); + long requestId = + latestSubscriptionEvent(context, 0) + .getPendingRemovalRequests() + .get(removedTablePath); + acknowledgeRemoval(enumerator, removedTablePath, requestId); + assertThat(enumerator.snapshotState(1L).getPendingRemovalTablePaths()) + .containsExactly(removedTablePath); + + context.runNextOneTimeCallable(); + + assertThat(latestAssignmentTableNames(context)).containsExactly(retainedTable); + assertThat(enumerator.snapshotState(2L).getPendingRemovalTablePaths()).isEmpty(); + } finally { + enumerator.close(); + } + } + } + + @Test + void testSubscriptionDeletionRemovesPendingKvSnapshotRelease() throws Throwable { + String subscriptionTable = "sub_snapshot_release"; + String targetTable = "tgt_snapshot_release"; + createSubscriptionTable(subscriptionTable); + createPkTable(targetTable); + insertSubscription(subscriptionTable, targetTable); + + FlussSubscriberTableDiscoverer subscriber = + new FlussSubscriberTableDiscoverer(DATABASE_NAME + "." + subscriptionTable, 100); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(NUM_READERS)) { + FlussSourceEnumerator enumerator = newEnumerator(context, subscriber, null); + try { + enumerator.start(); + registerAllReaders(context, enumerator); + runDiscoveryCycle(context); + + FlussSplitBase split = + context.getSplitsAssignmentSequence().get(0).assignment().values().stream() + .flatMap(List::stream) + .findFirst() + .orElseThrow(AssertionError::new); + enumerator.handleSourceEvent( + 0, + new FinishedKvSnapshotConsumeEvent( + 1L, Collections.singleton(split.getTableBucket()))); + assertThat(enumerator.pendingKvSnapshotBucketsForTesting()) + .containsExactly(split.getTableBucket()); + + tBatchEnv + .executeSql( + String.format( + "DELETE FROM %s WHERE table_name = '%s.%s'", + subscriptionTable, DATABASE_NAME, targetTable)) + .await(); + runDiscoveryCycle(context); + + assertThat(enumerator.pendingKvSnapshotBucketsForTesting()).isEmpty(); + enumerator.handleSourceEvent( + 0, + new FinishedKvSnapshotConsumeEvent( + 1L, Collections.singleton(split.getTableBucket()))); + assertThat(enumerator.pendingKvSnapshotBucketsForTesting()).isEmpty(); } finally { enumerator.close(); } @@ -731,6 +1411,23 @@ private FlussSourceEnumerator newEnumerator( false); } + private FlussSourceEnumerator newRestoredEnumerator( + MockSplitEnumeratorContext context, + TableDiscoverer discoverer, + FlussSourceEnumState restoredState) { + org.apache.fluss.config.Configuration flussConfig = + FLUSS_CLUSTER_EXTENSION.getClientConfig(); + return new FlussSourceEnumerator( + context, + discoverer, + flussConfig, + buildSourceConfig(flussConfig, null), + OffsetsInitializer.earliest(), + DISCOVERY_INTERVAL_MS, + restoredState, + LeaseContext.fromConf(new org.apache.flink.configuration.Configuration())); + } + private static Configuration buildSourceConfig( org.apache.fluss.config.Configuration flussConfig, String pattern) { Map map = new HashMap<>(); @@ -763,6 +1460,15 @@ private static void registerAllReaders( } } + private static void acknowledgeRemoval( + FlussSourceEnumerator enumerator, TablePath tablePath, long requestId) { + for (int readerId = 0; readerId < NUM_READERS; readerId++) { + enumerator.handleSourceEvent( + readerId, + new TableRemovalAckEvent(Collections.singletonMap(tablePath, requestId))); + } + } + private static int findParallelismWithDifferentOwner( org.apache.fluss.metadata.TableBucket tableBucket, int currentParallelism, @@ -814,6 +1520,15 @@ private static Set latestAssignmentTableNames( .collect(Collectors.toSet()); } + private static TableSubscriptionEvent latestSubscriptionEvent( + MockSplitEnumeratorContext context, int readerId) throws Exception { + return context.getSentSourceEvent().get(readerId).stream() + .filter(TableSubscriptionEvent.class::isInstance) + .map(TableSubscriptionEvent.class::cast) + .reduce((first, second) -> second) + .orElseThrow(AssertionError::new); + } + private void createPkTable(String tableName) throws Exception { tBatchEnv .executeSql( @@ -823,6 +1538,28 @@ private void createPkTable(String tableName) throws Exception { .await(); } + private void createPkTable(String tableName, int bucketCount) throws Exception { + TablePath tablePath = TablePath.of(DATABASE_NAME, tableName); + try (Connection connection = + ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig())) { + connection + .getAdmin() + .createTable( + tablePath, + TableDescriptor.builder() + .schema( + org.apache.fluss.metadata.Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("val", DataTypes.STRING()) + .primaryKey("id") + .build()) + .distributedBy(bucketCount, "id") + .build(), + false) + .get(); + } + } + private void createSubscriptionTable(String tableName) throws Exception { tBatchEnv .executeSql( @@ -842,7 +1579,7 @@ private void insertSubscription(String subscriptionTable, String targetTable) th } private static String fqnRegex(String database, String tablePattern) { - return java.util.regex.Pattern.quote(database) + "\\." + tablePattern; + return Pattern.quote(database) + "\\." + tablePattern; } private void waitForFlussClusterReady() throws Exception { diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/event/TableSubscriptionEventTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/event/TableSubscriptionEventTest.java new file mode 100644 index 00000000000..3187852c9ec --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/event/TableSubscriptionEventTest.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.fluss.source.event; + +import org.apache.fluss.metadata.TablePath; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link TableSubscriptionEvent}. */ +class TableSubscriptionEventTest { + + @Test + void testFencedTablePathsAreDefensivelyCopied() { + TablePath tablePath = TablePath.of("test_db", "test_table"); + Set fencedTablePaths = new HashSet<>(Collections.singleton(tablePath)); + + TableSubscriptionEvent event = + new TableSubscriptionEvent( + Collections.emptySet(), Collections.emptyMap(), fencedTablePaths); + fencedTablePaths.clear(); + + assertThat(event.getFencedTablePaths()).containsExactly(tablePath); + assertThatThrownBy(() -> event.getFencedTablePaths().clear()) + .isInstanceOf(UnsupportedOperationException.class); + } +} diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussRecordEmitterTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussRecordEmitterTest.java index eb9662863b3..074db9c192a 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussRecordEmitterTest.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussRecordEmitterTest.java @@ -199,6 +199,34 @@ void testSecondRecordDoesNotReEmitCreateTableEvent() throws Exception { assertThat(events.get(0)).isInstanceOf(DataChangeEvent.class); } + @Test + void testRemovedTableDoesNotEmitOrAdvanceStateUntilItIsReadded() throws Exception { + RowType rowType = + rowType(field("id", new IntType(false), 1), field("name", new StringType(true), 2)); + FlussLogSplit split = logSplitWithSchema(1, rowType); + FlussLogSplitState splitState = new FlussLogSplitState(split); + emitter.applySplit(split); + emitter.removeTable(TABLE_PATH); + + emitter.emitRecord( + logRecord(1, rowType, GenericRow.of(1, BinaryString.fromString("late")), 100L), + output, + splitState); + + assertThat(output.getCollectedEvents()).isEmpty(); + assertThat(splitState.toFlussSplit().getStartingOffset()).isEqualTo(100L); + + emitter.applySplit(split); + emitter.emitRecord( + logRecord(1, rowType, GenericRow.of(2, BinaryString.fromString("readded")), 100L), + output, + splitState); + assertThat(output.getCollectedEvents()) + .extracting(Event::getClass) + .containsExactly(CreateTableEvent.class, DataChangeEvent.class); + assertThat(splitState.toFlussSplit().getStartingOffset()).isEqualTo(101L); + } + @Test void testCreateTableEventContainsPrimaryKey() throws Exception { RowType rt = diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReaderTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReaderTest.java new file mode 100644 index 00000000000..f7098cbbf32 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReaderTest.java @@ -0,0 +1,356 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.fluss.source.reader; + +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.connectors.fluss.sink.v2.metrics.WrapperFlussMetricRegistry; +import org.apache.flink.cdc.connectors.fluss.source.deserializer.FlussRecordDeserializer; +import org.apache.flink.cdc.connectors.fluss.source.event.FinishedKvSnapshotConsumeEvent; +import org.apache.flink.cdc.connectors.fluss.source.event.TableRemovalAckEvent; +import org.apache.flink.cdc.connectors.fluss.source.event.TableSubscriptionEvent; +import org.apache.flink.cdc.connectors.fluss.source.metrics.FlussSourceReaderMetrics; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussHybridSnapshotLogSplit; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussHybridSnapshotLogSplitState; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussLogSplit; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitState; +import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; +import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue; +import org.apache.flink.metrics.groups.SourceReaderMetricGroup; +import org.apache.flink.metrics.testutils.MetricListener; +import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup; +import org.apache.flink.util.UserCodeClassLoader; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TablePath; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for subscription-aware split handling in {@link FlussSourceReader}. */ +class FlussSourceReaderTest { + + @Test + void testStagesSplitUntilFirstSubscriptionSnapshotAndIncludesItInCheckpoint() throws Exception { + TestingReaderContext readerContext = new TestingReaderContext(); + FlussSourceReader reader = newReader(readerContext); + FlussSplitBase split = + new FlussLogSplit( + PhysicalTablePath.of(TablePath.of("test_db", "test_table")), + new TableBucket(1L, 0), + 0L); + + try { + reader.addSplits(Collections.singletonList(split)); + + List checkpoint = reader.snapshotState(1L); + assertThat(checkpoint).containsExactly(split); + assertThat(readerContext.getSentEvents()).isEmpty(); + } finally { + reader.close(); + } + } + + @Test + void testFirstSubscriptionFenceDropsRestoredSplitButAllowsFreshSplit() throws Exception { + TestingReaderContext readerContext = new TestingReaderContext(); + FutureCompletingBlockingQueue> queue = + new FutureCompletingBlockingQueue<>(); + TrackingFetcherManager fetcherManager = new TrackingFetcherManager(queue); + FlussSourceReader reader = + new FlussSourceReader<>( + queue, + readerContext, + new WrapperFlussMetricRegistry( + readerContext.metricGroup(), Collections.emptySet()), + new FlussRecordEmitter<>(new FlussRecordDeserializer()), + fetcherManager); + FlussLogSplit restoredSplit = testSplit(); + FlussLogSplit freshSplit = + new FlussLogSplit( + restoredSplit.getPhysicalTablePath(), restoredSplit.getTableBucket(), 10L); + + try { + reader.addSplits(Collections.singletonList(restoredSplit)); + reader.handleSourceEvents( + new TableSubscriptionEvent( + Collections.singleton(restoredSplit.getTablePath()), + Collections.emptyMap(), + Collections.singleton(restoredSplit.getTablePath()))); + + assertThat(fetcherManager.addedSplits).isEmpty(); + assertThat(fetcherManager.removedTablePaths).isEmpty(); + + reader.addSplits(Collections.singletonList(freshSplit)); + assertThat(fetcherManager.addedSplits).containsExactly(freshSplit); + } finally { + reader.close(); + } + } + + @Test + void testAcknowledgesRemovalAfterFinishedSplitCallback() throws Exception { + TestingReaderContext readerContext = new TestingReaderContext(); + FutureCompletingBlockingQueue> queue = + new FutureCompletingBlockingQueue<>(); + TrackingFetcherManager fetcherManager = new TrackingFetcherManager(queue); + FlussSourceReader reader = + new FlussSourceReader<>( + queue, + readerContext, + new WrapperFlussMetricRegistry( + readerContext.metricGroup(), Collections.emptySet()), + new FlussRecordEmitter<>(new FlussRecordDeserializer()), + fetcherManager); + FlussSplitBase split = testSplit(); + FlussSplitState splitState = reader.initializedState(split); + long requestId = 7L; + + try { + reader.handleSourceEvents( + new TableSubscriptionEvent( + Collections.emptySet(), + Collections.singletonMap(split.getTablePath(), requestId))); + assertThat(fetcherManager.removedTablePaths).containsExactly(split.getTablePath()); + assertThat(readerContext.getSentEvents()).isEmpty(); + + reader.onSplitFinished(Collections.singletonMap(split.splitId(), splitState)); + + assertThat(readerContext.getSentEvents()) + .singleElement() + .isInstanceOfSatisfying( + TableRemovalAckEvent.class, + ack -> + assertThat(ack.getCompletedRemovalRequests()) + .containsExactlyEntriesOf( + Collections.singletonMap( + split.getTablePath(), requestId))); + } finally { + reader.close(); + } + } + + @Test + void testAcknowledgesRemovalWithoutLocalSplitsWithoutFetcherRemoval() throws Exception { + TestingReaderContext readerContext = new TestingReaderContext(); + FutureCompletingBlockingQueue> queue = + new FutureCompletingBlockingQueue<>(); + TrackingFetcherManager fetcherManager = new TrackingFetcherManager(queue); + FlussSourceReader reader = + new FlussSourceReader<>( + queue, + readerContext, + new WrapperFlussMetricRegistry( + readerContext.metricGroup(), Collections.emptySet()), + new FlussRecordEmitter<>(new FlussRecordDeserializer()), + fetcherManager); + TablePath tablePath = testSplit().getTablePath(); + + try { + reader.handleSourceEvents( + new TableSubscriptionEvent( + Collections.emptySet(), Collections.singletonMap(tablePath, 1L))); + + assertThat(fetcherManager.removedTablePaths).isEmpty(); + assertThat(readerContext.getSentEvents()) + .singleElement() + .isInstanceOfSatisfying( + TableRemovalAckEvent.class, + ack -> + assertThat(ack.getCompletedRemovalRequests()) + .containsExactlyEntriesOf( + Collections.singletonMap(tablePath, 1L))); + } finally { + reader.close(); + } + } + + @Test + void testReaddedSnapshotBucketReportsFinishedAgainAfterRemoval() throws Exception { + TestingReaderContext readerContext = new TestingReaderContext(); + FutureCompletingBlockingQueue> queue = + new FutureCompletingBlockingQueue<>(); + TrackingFetcherManager fetcherManager = new TrackingFetcherManager(queue); + FlussSourceReader reader = + new FlussSourceReader<>( + queue, + readerContext, + new WrapperFlussMetricRegistry( + readerContext.metricGroup(), Collections.emptySet()), + new FlussRecordEmitter<>(new FlussRecordDeserializer()), + fetcherManager); + FlussHybridSnapshotLogSplit split = finishedSnapshotSplit(); + TablePath tablePath = split.getTablePath(); + + try { + reader.handleSourceEvents( + new TableSubscriptionEvent( + Collections.singleton(tablePath), Collections.emptyMap())); + reader.addSplits(Collections.singletonList(split)); + reader.snapshotState(1L); + assertThat(readerContext.getSentEvents()) + .singleElement() + .isInstanceOf(FinishedKvSnapshotConsumeEvent.class); + + readerContext.clearSentEvents(); + reader.handleSourceEvents( + new TableSubscriptionEvent( + Collections.emptySet(), Collections.singletonMap(tablePath, 1L))); + reader.onSplitFinished( + Collections.singletonMap( + split.splitId(), new FlussHybridSnapshotLogSplitState(split))); + readerContext.clearSentEvents(); + + reader.handleSourceEvents( + new TableSubscriptionEvent( + Collections.singleton(tablePath), Collections.emptyMap())); + reader.addSplits(Collections.singletonList(split)); + reader.snapshotState(2L); + + assertThat(readerContext.getSentEvents()) + .singleElement() + .isInstanceOfSatisfying( + FinishedKvSnapshotConsumeEvent.class, + event -> + assertThat(event.getTableBuckets()) + .containsExactly(split.getTableBucket())); + } finally { + reader.close(); + } + } + + private static FlussSourceReader newReader(TestingReaderContext readerContext) { + return new FlussSourceReader<>( + new FutureCompletingBlockingQueue>(), + readerContext, + new Configuration(), + new WrapperFlussMetricRegistry(readerContext.metricGroup(), Collections.emptySet()), + new FlussSourceReaderMetrics(readerContext.metricGroup()), + new FlussRecordEmitter<>(new FlussRecordDeserializer())); + } + + private static FlussLogSplit testSplit() { + return new FlussLogSplit( + PhysicalTablePath.of(TablePath.of("test_db", "test_table")), + new TableBucket(1L, 0), + 0L); + } + + private static FlussHybridSnapshotLogSplit finishedSnapshotSplit() { + return new FlussHybridSnapshotLogSplit( + PhysicalTablePath.of(TablePath.of("test_db", "test_table")), + new TableBucket(1L, 0), + 1L, + 0L, + 0L, + true, + null, + null); + } + + private static class TrackingFetcherManager extends FlussSourceFetcherManager { + + private Set removedTablePaths = Collections.emptySet(); + private final List addedSplits = new ArrayList<>(); + + private TrackingFetcherManager( + FutureCompletingBlockingQueue> queue) { + super(queue, () -> null); + } + + @Override + void removeTables(Set tablePaths) { + removedTablePaths = tablePaths; + } + + @Override + public void addSplits(List splits) { + addedSplits.addAll(splits); + } + } + + private static class TestingReaderContext implements SourceReaderContext { + + private final SourceReaderMetricGroup metricGroup; + private final List sentEvents = new ArrayList<>(); + + private TestingReaderContext() { + MetricListener metricListener = new MetricListener(); + metricGroup = InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup()); + } + + @Override + public SourceReaderMetricGroup metricGroup() { + return metricGroup; + } + + @Override + public org.apache.flink.configuration.Configuration getConfiguration() { + return new org.apache.flink.configuration.Configuration(); + } + + @Override + public String getLocalHostName() { + return "localhost"; + } + + @Override + public int getIndexOfSubtask() { + return 0; + } + + @Override + public void sendSplitRequest() {} + + @Override + public void sendSourceEventToCoordinator(SourceEvent sourceEvent) { + sentEvents.add(sourceEvent); + } + + @Override + public UserCodeClassLoader getUserCodeClassLoader() { + return new UserCodeClassLoader() { + @Override + public ClassLoader asClassLoader() { + return FlussSourceReaderTest.class.getClassLoader(); + } + + @Override + public void registerReleaseHookIfAbsent( + String releaseHookName, Runnable releaseHook) {} + }; + } + + private List getSentEvents() { + return sentEvents; + } + + private void clearSentEvents() { + sentEvents.clear(); + } + } +} diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReaderTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReaderTest.java index c9130ebdf80..eb5241a0d25 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReaderTest.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReaderTest.java @@ -20,12 +20,36 @@ import org.apache.flink.cdc.connectors.fluss.source.split.FlussHybridSnapshotLogSplit; import org.apache.flink.cdc.connectors.fluss.source.split.FlussLogSplit; import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase; +import org.apache.flink.connector.base.source.reader.RecordsBySplits; +import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; +import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange; +import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue; +import org.apache.fluss.client.table.Table; +import org.apache.fluss.client.table.scanner.batch.BatchScanner; +import org.apache.fluss.client.table.scanner.log.MultiTableLogScanner; +import org.apache.fluss.config.Configuration; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.types.RowType; import org.junit.jupiter.api.Test; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Proxy; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -49,6 +73,108 @@ void testValidateHybridSnapshotLogSplitTableId() { new FlussHybridSnapshotLogSplit(PHYSICAL_TABLE_PATH, TABLE_BUCKET, 10L, 100L)); } + @Test + void testRemoveTablesFinishesLogAndSnapshotSplitsAndClearsTableResources() throws Exception { + TableBucket logBucket = new TableBucket(TABLE_ID, 0); + TableBucket currentSnapshotBucket = new TableBucket(TABLE_ID, 1); + TableBucket queuedSnapshotBucket = new TableBucket(TABLE_ID, 2); + FlussLogSplit logSplit = new FlussLogSplit(PHYSICAL_TABLE_PATH, logBucket, 10L); + FlussHybridSnapshotLogSplit currentSnapshot = + new FlussHybridSnapshotLogSplit( + PHYSICAL_TABLE_PATH, currentSnapshotBucket, 1L, 10L); + FlussHybridSnapshotLogSplit queuedSnapshot = + new FlussHybridSnapshotLogSplit(PHYSICAL_TABLE_PATH, queuedSnapshotBucket, 2L, 10L); + FlussSplitReader reader = new FlussSplitReader(new Configuration(), null, null); + AtomicBoolean tableClosed = new AtomicBoolean(); + AtomicBoolean batchScannerClosed = new AtomicBoolean(); + AtomicBoolean logScannerWokenUp = new AtomicBoolean(); + Set unsubscribedBuckets = new HashSet<>(); + Table table = + proxy( + Table.class, + (proxy, method, arguments) -> { + if (method.getName().equals("close")) { + tableClosed.set(true); + } + return null; + }); + MultiTableLogScanner logScanner = + proxy( + MultiTableLogScanner.class, + (proxy, method, arguments) -> { + if (method.getName().equals("unsubscribe")) { + unsubscribedBuckets.add((Integer) arguments[1]); + } else if (method.getName().equals("wakeup")) { + logScannerWokenUp.set(true); + } + return null; + }); + BatchScanner batchScanner = + proxy( + BatchScanner.class, + (proxy, method, arguments) -> { + if (method.getName().equals("close")) { + batchScannerClosed.set(true); + } + return null; + }); + + tableResources(reader).put(TABLE_PATH, table); + tableRowTypes(reader).put(TABLE_PATH, new RowType(Collections.emptyList())); + tablePrimaryKeyNames(reader).put(TABLE_PATH, Collections.singletonList("id")); + tablePartitionKeyNames(reader).put(TABLE_PATH, Collections.singletonList("part")); + bucketToSplit(reader).put(logBucket, logSplit); + bucketToSplit(reader).put(currentSnapshotBucket, currentSnapshot); + bucketToSplit(reader).put(queuedSnapshotBucket, queuedSnapshot); + boundedSplits(reader).add(queuedSnapshot); + setField(reader, "currentBoundedSplit", currentSnapshot); + setField(reader, "currentBatchScanner", batchScanner); + setField(reader, "currentLogScanner", logScanner); + + reader.removeTables(Collections.singleton(TABLE_PATH)); + + RecordsWithSplitIds records = reader.fetch(); + assertThat(records.finishedSplits()) + .containsExactlyInAnyOrder( + logSplit.splitId(), currentSnapshot.splitId(), queuedSnapshot.splitId()); + assertThat(bucketToSplit(reader)).isEmpty(); + assertThat(boundedSplits(reader)).isEmpty(); + assertThat(tableResources(reader)).isEmpty(); + assertThat(tableRowTypes(reader)).isEmpty(); + assertThat(tablePrimaryKeyNames(reader)).isEmpty(); + assertThat(tablePartitionKeyNames(reader)).isEmpty(); + assertThat(unsubscribedBuckets) + .containsExactlyInAnyOrder( + logBucket.getBucket(), + currentSnapshotBucket.getBucket(), + queuedSnapshotBucket.getBucket()); + assertThat(logScannerWokenUp).isFalse(); + assertThatCode(reader::fetch).doesNotThrowAnyException(); + assertThat(batchScannerClosed).isTrue(); + assertThat(tableClosed).isTrue(); + } + + @Test + void testFetcherManagerFinishesLastAssignedSplitAfterTableRemoval() throws Exception { + TestingSplitReader reader = new TestingSplitReader(); + FlussSourceFetcherManager manager = + new FlussSourceFetcherManager(new FutureCompletingBlockingQueue<>(), () -> reader); + FlussLogSplit lastSplit = new FlussLogSplit(PHYSICAL_TABLE_PATH, TABLE_BUCKET, 10L); + try { + manager.addSplits(Collections.singletonList(lastSplit)); + assertThat(reader.splitAdded.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(reader.fetchStarted.await(10, TimeUnit.SECONDS)).isTrue(); + manager.removeTables(Collections.singleton(TABLE_PATH)); + assertThat(reader.removalRequested.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(reader.removedTables).containsExactly(TABLE_PATH); + assertThat(reader.wakeUps).hasValue(1); + assertThat(awaitRecords(manager.getQueue(), lastSplit.splitId()).finishedSplits()) + .containsExactly(lastSplit.splitId()); + } finally { + manager.close(1000L); + } + } + private static void assertTableIdValidation(FlussSplitBase split) { assertThatCode(() -> FlussSplitReader.validateTableId(split, TABLE_ID)) .doesNotThrowAnyException(); @@ -57,4 +183,124 @@ private static void assertTableIdValidation(FlussSplitBase split) { .hasMessage( "Table ID mismatch for split test_db.test_table.0: split table ID is 1001, but table test_db.test_table has ID 2002."); } + + @SuppressWarnings("unchecked") + private static Map tableResources(FlussSplitReader reader) throws Exception { + return (Map) getField(reader, "tables"); + } + + @SuppressWarnings("unchecked") + private static Map tableRowTypes(FlussSplitReader reader) throws Exception { + return (Map) getField(reader, "tableRowTypes"); + } + + @SuppressWarnings("unchecked") + private static Map> tablePrimaryKeyNames(FlussSplitReader reader) + throws Exception { + return (Map>) getField(reader, "tablePrimaryKeyNames"); + } + + @SuppressWarnings("unchecked") + private static Map> tablePartitionKeyNames(FlussSplitReader reader) + throws Exception { + return (Map>) getField(reader, "tablePartitionKeyNames"); + } + + @SuppressWarnings("unchecked") + private static Map bucketToSplit(FlussSplitReader reader) + throws Exception { + return (Map) getField(reader, "bucketToSplit"); + } + + @SuppressWarnings("unchecked") + private static Queue boundedSplits(FlussSplitReader reader) throws Exception { + return (Queue) getField(reader, "boundedSplits"); + } + + private static Object getField(FlussSplitReader reader, String name) throws Exception { + Field field = FlussSplitReader.class.getDeclaredField(name); + field.setAccessible(true); + return field.get(reader); + } + + private static void setField(FlussSplitReader reader, String name, Object value) + throws Exception { + Field field = FlussSplitReader.class.getDeclaredField(name); + field.setAccessible(true); + field.set(reader, value); + } + + private static RecordsWithSplitIds awaitRecords( + FutureCompletingBlockingQueue> queue, + String expectedFinishedSplitId) + throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + RecordsWithSplitIds records; + while (System.nanoTime() < deadline) { + records = queue.poll(); + if (records != null && records.finishedSplits().contains(expectedFinishedSplitId)) { + return records; + } + Thread.sleep(10L); + } + throw new AssertionError("Timed out waiting for finished split " + expectedFinishedSplitId); + } + + private static T proxy(Class type, InvocationHandler handler) { + return type.cast( + Proxy.newProxyInstance(type.getClassLoader(), new Class[] {type}, handler)); + } + + private static class TestingSplitReader extends FlussSplitReader { + + private final CountDownLatch splitAdded = new CountDownLatch(1); + private final CountDownLatch fetchStarted = new CountDownLatch(1); + private final CountDownLatch removalRequested = new CountDownLatch(1); + private final CountDownLatch wakeUp = new CountDownLatch(1); + private final AtomicInteger wakeUps = new AtomicInteger(); + private Set removedTables = Collections.emptySet(); + private String assignedSplitId; + private String finishedSplitId; + + private TestingSplitReader() { + super(new Configuration(), null, null); + } + + @Override + public RecordsWithSplitIds fetch() { + if (finishedSplitId != null) { + RecordsBySplits.Builder builder = + new RecordsBySplits.Builder<>(); + builder.addFinishedSplit(finishedSplitId); + finishedSplitId = null; + return builder.build(); + } + fetchStarted.countDown(); + try { + wakeUp.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return new RecordsBySplits.Builder().build(); + } + + @Override + void removeTables(Set tablePaths) { + removedTables = tablePaths; + finishedSplitId = assignedSplitId; + removalRequested.countDown(); + } + + @Override + public void handleSplitsChanges(SplitsChange splitsChanges) { + assignedSplitId = splitsChanges.splits().get(0).splitId(); + splitAdded.countDown(); + } + + @Override + public void wakeUp() { + wakeUps.incrementAndGet(); + wakeUp.countDown(); + } + } }