This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 5c61d4ca3 [FLINK-40732][mongodb] Release finished snapshot split
metadata after entering the stream phase (#4553)
5c61d4ca3 is described below
commit 5c61d4ca327c05ed5964cfd8fc24727724123039
Author: Spoorthi Basu <[email protected]>
AuthorDate: Tue Sep 22 01:30:41 2026 -0400
[FLINK-40732][mongodb] Release finished snapshot split metadata after
entering the stream phase (#4553)
Expose scan.incremental.snapshot.metadata.release.enabled in the MongoDB
connector, through both the DataStream builder and Flink SQL. MongoDB
implements SourceConfig directly, so the config field, the factory setter
and the fail-fast against scan.newly-added-table.enabled are implemented
here instead of inherited from BaseSourceConfig. Cover the option with
table factory and config factory tests, and add group-fetch, JobManager
failover and TaskManager failover ITCases against a real MongoDB
container.
---
.../docs/connectors/flink-sources/mongodb-cdc.md | 7 ++
.../docs/connectors/flink-sources/mongodb-cdc.md | 7 ++
.../mongodb/source/MongoDBSourceBuilder.java | 10 +++
.../mongodb/source/config/MongoDBSourceConfig.java | 23 ++++++-
.../source/config/MongoDBSourceConfigFactory.java | 16 ++++-
.../mongodb/table/MongoDBTableSource.java | 18 +++--
.../mongodb/table/MongoDBTableSourceFactory.java | 7 +-
.../source/MongoDBParallelSourceITCase.java | 80 +++++++++++++++++++++-
.../config/MongoDBSourceConfigFactoryTest.java | 60 ++++++++++++++++
.../mongodb/table/MongoDBTableFactoryTest.java | 61 ++++++++++++++++-
10 files changed, 275 insertions(+), 14 deletions(-)
diff --git a/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
b/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
index df1e8cdad..83dda0ecf 100644
--- a/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
@@ -325,6 +325,13 @@ MongoDB 的更改事件记录在消息之前没有更新。因此，我们只能
<td>Boolean</td>
<td>是否在快照结束后关闭空闲的 Reader。 此特性需要 flink 版本大于等于 1.14 并且
'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 需要设置为
true。</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.metadata.release.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>是否在 source 进入增量阶段后，释放 source coordinator
持有的快照分片元数据（已分配的分片、已完成分片的位点以及表结构），以降低快照分片数量非常大的作业的 JobManager 内存占用。默认关闭。与
scan.newly-added-table.enabled
不兼容：同时开启两者会导致作业启动失败，且已释放元数据的作业无法再开启动态加表功能。仅在成功完成一次 checkpoint 后才会释放；若未开启
checkpoint 或没有 checkpoint 完成，则会保留该元数据，因此该配置项在未开启 checkpoint 时不生效。开启该配置项后生成的
checkpoint 或 savepoint，无法在降级到 Flink CDC 3.6.0 及更早版本后用于恢复作业。</td>
+ </tr>
<tr>
<td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
<td>optional</td>
diff --git a/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
b/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
index 83d2a23f6..5eed10d75 100644
--- a/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
@@ -343,6 +343,13 @@ Connector Options
so it does not need to be explicitly configured
'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.metadata.release.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to release the snapshot split metadata (assigned splits,
finished offsets and table schemas) held by the source coordinator once the
source has entered the stream phase, to reduce JobManager memory on jobs with a
very large number of snapshot splits. Disabled by default. Incompatible with
scan.newly-added-table.enabled: enabling both fails at startup, and a job that
has released the metadata cannot later enable newly-added-table scanning.
Release happens only after a su [...]
+ </tr>
<tr>
<td>scan.cursor.no-timeout</td>
<td>optional</td>
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
index c61a5d0ce..553a533fd 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
@@ -277,6 +277,16 @@ public class MongoDBSourceBuilder<T> {
return this;
}
+ /**
+ * Whether to release the finished snapshot split metadata from the
coordinator after entering
+ * the stream phase, to reduce JobManager memory pressure (FLINK-40697).
+ */
+ public MongoDBSourceBuilder<T> releaseSnapshotMetadataEnabled(
+ boolean releaseSnapshotMetadataEnabled) {
+
this.configFactory.releaseSnapshotMetadataEnabled(releaseSnapshotMetadataEnabled);
+ return this;
+ }
+
/**
* Build the {@link MongoDBSource}.
*
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
index 035626188..b32dba452 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
@@ -28,6 +28,7 @@ import java.util.Objects;
import static
org.apache.flink.cdc.connectors.mongodb.source.utils.MongoUtils.buildConnectionString;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
/** A MongoDB Source configuration which is used by {@link MongoDBSource}. */
public class MongoDBSourceConfig implements SourceConfig {
@@ -56,6 +57,7 @@ public class MongoDBSourceConfig implements SourceConfig {
private final boolean skipSnapshotBackfill;
private final boolean isScanNewlyAddedTableEnabled;
private final boolean assignUnboundedChunkFirst;
+ private final boolean releaseSnapshotMetadataEnabled;
MongoDBSourceConfig(
String scheme,
@@ -79,7 +81,8 @@ public class MongoDBSourceConfig implements SourceConfig {
boolean disableCursorTimeout,
boolean skipSnapshotBackfill,
boolean isScanNewlyAddedTableEnabled,
- boolean assignUnboundedChunkFirst) {
+ boolean assignUnboundedChunkFirst,
+ boolean releaseSnapshotMetadataEnabled) {
this.scheme = checkNotNull(scheme);
this.hosts = checkNotNull(hosts);
this.username = username;
@@ -103,6 +106,13 @@ public class MongoDBSourceConfig implements SourceConfig {
this.skipSnapshotBackfill = skipSnapshotBackfill;
this.isScanNewlyAddedTableEnabled = isScanNewlyAddedTableEnabled;
this.assignUnboundedChunkFirst = assignUnboundedChunkFirst;
+ this.releaseSnapshotMetadataEnabled = releaseSnapshotMetadataEnabled;
+ checkState(
+ !(isScanNewlyAddedTableEnabled &&
releaseSnapshotMetadataEnabled),
+ "scan.incremental.snapshot.metadata.release.enabled and "
+ + "scan.newly-added-table.enabled cannot both be
enabled: releasing the "
+ + "snapshot split metadata would drop the assigned
splits, finished offsets "
+ + "and table schemas that newly-added-table scanning
needs.");
}
public String getScheme() {
@@ -207,6 +217,11 @@ public class MongoDBSourceConfig implements SourceConfig {
return assignUnboundedChunkFirst;
}
+ @Override
+ public boolean isReleaseSnapshotMetadataEnabled() {
+ return releaseSnapshotMetadataEnabled;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -234,7 +249,8 @@ public class MongoDBSourceConfig implements SourceConfig {
&& Objects.equals(collectionList, that.collectionList)
&& Objects.equals(connectionString, that.connectionString)
&& Objects.equals(skipSnapshotBackfill,
that.skipSnapshotBackfill)
- && Objects.equals(isScanNewlyAddedTableEnabled,
that.isScanNewlyAddedTableEnabled);
+ && Objects.equals(isScanNewlyAddedTableEnabled,
that.isScanNewlyAddedTableEnabled)
+ && releaseSnapshotMetadataEnabled ==
that.releaseSnapshotMetadataEnabled;
}
@Override
@@ -258,6 +274,7 @@ public class MongoDBSourceConfig implements SourceConfig {
samplesPerChunk,
closeIdleReaders,
skipSnapshotBackfill,
- isScanNewlyAddedTableEnabled);
+ isScanNewlyAddedTableEnabled,
+ releaseSnapshotMetadataEnabled);
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
index 691f3712a..76782e822 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
@@ -25,6 +25,7 @@ import java.util.Arrays;
import java.util.List;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
+import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED;
import static
org.apache.flink.cdc.connectors.base.utils.EnvironmentUtils.checkSupportCheckpointsAfterTasksFinished;
import static
org.apache.flink.cdc.connectors.mongodb.internal.MongoDBEnvelope.MONGODB_SCHEME;
import static
org.apache.flink.cdc.connectors.mongodb.internal.MongoDBEnvelope.MONGODB_SRV_SCHEME;
@@ -63,6 +64,8 @@ public class MongoDBSourceConfigFactory implements
Factory<MongoDBSourceConfig>
protected boolean skipSnapshotBackfill = false;
protected boolean scanNewlyAddedTableEnabled = false;
protected boolean assignUnboundedChunkFirst = false;
+ protected boolean releaseSnapshotMetadataEnabled =
+ SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED.defaultValue();
/** The protocol connected to MongoDB. For example mongodb or mongodb+srv.
*/
public MongoDBSourceConfigFactory scheme(String scheme) {
@@ -280,6 +283,16 @@ public class MongoDBSourceConfigFactory implements
Factory<MongoDBSourceConfig>
return this;
}
+ /**
+ * Whether to release the snapshot split metadata held by the source
coordinator once the stream
+ * split is assembled and checkpoint-covered. Defaults to false.
+ */
+ public MongoDBSourceConfigFactory releaseSnapshotMetadataEnabled(
+ boolean releaseSnapshotMetadataEnabled) {
+ this.releaseSnapshotMetadataEnabled = releaseSnapshotMetadataEnabled;
+ return this;
+ }
+
/** Creates a new {@link MongoDBSourceConfig} for the given subtask {@code
subtaskId}. */
@Override
public MongoDBSourceConfig create(int subtaskId) {
@@ -306,6 +319,7 @@ public class MongoDBSourceConfigFactory implements
Factory<MongoDBSourceConfig>
disableCursorTimeout,
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
- assignUnboundedChunkFirst);
+ assignUnboundedChunkFirst,
+ releaseSnapshotMetadataEnabled);
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
index 03d006264..d8aef1b5c 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
@@ -89,6 +89,7 @@ public class MongoDBTableSource implements ScanTableSource,
SupportsReadingMetad
private final boolean skipSnapshotBackfill;
private final boolean scanNewlyAddedTableEnabled;
private final boolean assignUnboundedChunkFirst;
+ private final boolean releaseSnapshotMetadataEnabled;
//
--------------------------------------------------------------------------------------------
// Mutable attributes
@@ -127,7 +128,8 @@ public class MongoDBTableSource implements ScanTableSource,
SupportsReadingMetad
boolean noCursorTimeout,
boolean skipSnapshotBackfill,
boolean scanNewlyAddedTableEnabled,
- boolean assignUnboundedChunkFirst) {
+ boolean assignUnboundedChunkFirst,
+ boolean releaseSnapshotMetadataEnabled) {
this.physicalSchema = physicalSchema;
this.scheme = checkNotNull(scheme);
this.hosts = checkNotNull(hosts);
@@ -157,6 +159,7 @@ public class MongoDBTableSource implements ScanTableSource,
SupportsReadingMetad
this.skipSnapshotBackfill = skipSnapshotBackfill;
this.scanNewlyAddedTableEnabled = scanNewlyAddedTableEnabled;
this.assignUnboundedChunkFirst = assignUnboundedChunkFirst;
+ this.releaseSnapshotMetadataEnabled = releaseSnapshotMetadataEnabled;
}
@Override
@@ -217,7 +220,8 @@ public class MongoDBTableSource implements ScanTableSource,
SupportsReadingMetad
.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
.deserializer(deserializer)
.disableCursorTimeout(noCursorTimeout)
-
.assignUnboundedChunkFirst(assignUnboundedChunkFirst);
+
.assignUnboundedChunkFirst(assignUnboundedChunkFirst)
+
.releaseSnapshotMetadataEnabled(releaseSnapshotMetadataEnabled);
Optional.ofNullable(databaseList).ifPresent(builder::databaseList);
Optional.ofNullable(collectionList).ifPresent(builder::collectionList);
@@ -324,7 +328,8 @@ public class MongoDBTableSource implements ScanTableSource,
SupportsReadingMetad
noCursorTimeout,
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
- assignUnboundedChunkFirst);
+ assignUnboundedChunkFirst,
+ releaseSnapshotMetadataEnabled);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
@@ -367,7 +372,9 @@ public class MongoDBTableSource implements ScanTableSource,
SupportsReadingMetad
&& Objects.equals(noCursorTimeout, that.noCursorTimeout)
&& Objects.equals(skipSnapshotBackfill,
that.skipSnapshotBackfill)
&& Objects.equals(scanNewlyAddedTableEnabled,
that.scanNewlyAddedTableEnabled)
- && Objects.equals(assignUnboundedChunkFirst,
that.assignUnboundedChunkFirst);
+ && Objects.equals(assignUnboundedChunkFirst,
that.assignUnboundedChunkFirst)
+ && Objects.equals(
+ releaseSnapshotMetadataEnabled,
that.releaseSnapshotMetadataEnabled);
}
@Override
@@ -401,7 +408,8 @@ public class MongoDBTableSource implements ScanTableSource,
SupportsReadingMetad
noCursorTimeout,
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
- assignUnboundedChunkFirst);
+ assignUnboundedChunkFirst,
+ releaseSnapshotMetadataEnabled);
}
@Override
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
index 71770cf1e..c85bf9111 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
@@ -37,6 +37,7 @@ import java.util.Set;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
+import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_STARTUP_MODE;
@@ -130,6 +131,8 @@ public class MongoDBTableSourceFactory implements
DynamicTableSourceFactory {
boolean scanNewlyAddedTableEnabled =
config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
boolean assignUnboundedChunkFirst =
config.get(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
+ boolean releaseSnapshotMetadataEnabled =
+ config.get(SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
int splitSizeMB = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB);
int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE);
@@ -173,7 +176,8 @@ public class MongoDBTableSourceFactory implements
DynamicTableSourceFactory {
noCursorTimeout,
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
- assignUnboundedChunkFirst);
+ assignUnboundedChunkFirst,
+ releaseSnapshotMetadataEnabled);
}
private void checkPrimaryKey(UniqueConstraint pk, String message) {
@@ -258,6 +262,7 @@ public class MongoDBTableSourceFactory implements
DynamicTableSourceFactory {
options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED);
options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
return options;
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBParallelSourceITCase.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBParallelSourceITCase.java
index 906f0a20f..a3f64ab20 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBParallelSourceITCase.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBParallelSourceITCase.java
@@ -49,7 +49,9 @@ import org.junit.jupiter.api.Timeout;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -129,6 +131,54 @@ class MongoDBParallelSourceITCase extends
MongoDBSourceTestBase {
FailoverType.JM, FailoverPhase.STREAM, new String[]
{"customers", "customers_1"});
}
+ @Test
+ void testCoordinatorReleasesSnapshotMetadataInGroupFetchPath() throws
Exception {
+ // A small meta-group size forces the reader to fetch metadata over
several groups, which
+ // is the path where the coordinator holds (and, with the option on,
releases) the bulk
+ // snapshot-split metadata. No failover: this asserts the release does
not drop any
+ // snapshot or change stream data.
+ testMongoDBParallelSource(
+ DEFAULT_PARALLELISM,
+ FailoverType.NONE,
+ FailoverPhase.NEVER,
+ new String[] {"customers", "customers_1"},
+ false,
+ releaseEnabledGroupFetchOptions());
+ }
+
+ @Test
+ void testCoordinatorReleaseSurvivesStreamPhaseFailover() throws Exception {
+ // JobManager (coordinator) failover in the stream phase after the
release: the light,
+ // released state must restore correctly and lose no data.
+ testMongoDBParallelSource(
+ DEFAULT_PARALLELISM,
+ FailoverType.JM,
+ FailoverPhase.STREAM,
+ new String[] {"customers", "customers_1"},
+ false,
+ releaseEnabledGroupFetchOptions());
+ }
+
+ @Test
+ void testCoordinatorReleaseSurvivesStreamPhaseReaderFailover() throws
Exception {
+ // TaskManager (reader) failover in the stream phase with the release
enabled: the
+ // generation guard must reject any stale assembled report and no data
is lost.
+ testMongoDBParallelSource(
+ DEFAULT_PARALLELISM,
+ FailoverType.TM,
+ FailoverPhase.STREAM,
+ new String[] {"customers", "customers_1"},
+ false,
+ releaseEnabledGroupFetchOptions());
+ }
+
+ private static Map<String, String> releaseEnabledGroupFetchOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put("chunk-meta.group.size", "1");
+ options.put("scan.incremental.snapshot.metadata.release.enabled",
"true");
+ return options;
+ }
+
@Test
void testTaskManagerFailoverSingleParallelism() throws Exception {
testMongoDBParallelSource(
@@ -516,6 +566,23 @@ class MongoDBParallelSourceITCase extends
MongoDBSourceTestBase {
String[] captureCustomerCollections,
boolean skipSnapshotBackfill)
throws Exception {
+ testMongoDBParallelSource(
+ parallelism,
+ failoverType,
+ failoverPhase,
+ captureCustomerCollections,
+ skipSnapshotBackfill,
+ Collections.emptyMap());
+ }
+
+ private void testMongoDBParallelSource(
+ int parallelism,
+ FailoverType failoverType,
+ FailoverPhase failoverPhase,
+ String[] captureCustomerCollections,
+ boolean skipSnapshotBackfill,
+ Map<String, String> extraOptions)
+ throws Exception {
String customerDatabase =
MONGO_CONTAINER.executeCommandFileInSeparateDatabase("customer");
@@ -543,13 +610,24 @@ class MongoDBParallelSourceITCase extends
MongoDBSourceTestBase {
+ " 'collection' = '%s',"
+ " 'heartbeat.interval.ms' = '500',"
+ " 'scan.incremental.snapshot.backfill.skip'
= '%s'"
+ + " %s"
+ ")",
MONGO_CONTAINER.getHostAndPort(),
FLINK_USER,
FLINK_USER_PASSWORD,
customerDatabase,
getCollectionNameRegex(customerDatabase,
captureCustomerCollections),
- skipSnapshotBackfill);
+ skipSnapshotBackfill,
+ extraOptions.isEmpty()
+ ? ""
+ : ","
+ + extraOptions.entrySet().stream()
+ .map(
+ e ->
+ String.format(
+
"'%s'='%s'",
+
e.getKey(), e.getValue()))
+
.collect(Collectors.joining(",")));
// first step: check the snapshot data
String[] snapshotForSingleTable =
new String[] {
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactoryTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactoryTest.java
new file mode 100644
index 000000000..803dd22e6
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactoryTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.mongodb.source.config;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link MongoDBSourceConfigFactory}. */
+class MongoDBSourceConfigFactoryTest {
+
+ @Test
+ void testFailFastWhenReleaseAndNewlyAddedTableBothEnabled() {
+ // Enabling metadata release together with scan.newly-added-table is
contradictory: the
+ // release would drop the metadata the newly-added-table flow needs.
The config must reject
+ // it at build time rather than silently disabling the release.
+ MongoDBSourceConfigFactory factory = new MongoDBSourceConfigFactory();
+ factory.hosts("localhost:27017")
+ .scanNewlyAddedTableEnabled(true)
+ .releaseSnapshotMetadataEnabled(true);
+
+ Assertions.assertThatThrownBy(() -> factory.create(0))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("cannot both be enabled");
+ }
+
+ @Test
+ void testReleaseSnapshotMetadataIsDisabledByDefault() {
+ MongoDBSourceConfig config =
+ new
MongoDBSourceConfigFactory().hosts("localhost:27017").create(0);
+
+
Assertions.assertThat(config.isReleaseSnapshotMetadataEnabled()).isFalse();
+ }
+
+ @Test
+ void testReleaseSnapshotMetadataReachesTheConfig() {
+ // the enumerator reads this getter to decide whether to arm the
release
+ MongoDBSourceConfig config =
+ new MongoDBSourceConfigFactory()
+ .hosts("localhost:27017")
+ .releaseSnapshotMetadataEnabled(true)
+ .create(0);
+
+
Assertions.assertThat(config.isReleaseSnapshotMetadataEnabled()).isTrue();
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
index 41f53c8e8..12430d3a8 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
@@ -49,6 +49,7 @@ import java.util.Map;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
+import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED;
import static
org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
import static
org.apache.flink.cdc.connectors.mongodb.internal.MongoDBEnvelope.MONGODB_SRV_SCHEME;
@@ -125,6 +126,9 @@ class MongoDBTableFactoryTest {
private static final boolean SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT =
SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue();
+ private static final boolean
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT =
+ SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED.defaultValue();
+
@Test
void testCommonProperties() {
Map<String, String> properties = getAllOptions();
@@ -159,7 +163,8 @@ class MongoDBTableFactoryTest {
SCAN_NO_CURSOR_TIMEOUT_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
-
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue());
+
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue(),
+
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT);
Assertions.assertThat(actualSource).isEqualTo(expectedSource);
}
@@ -215,8 +220,56 @@ class MongoDBTableFactoryTest {
false,
true,
true,
+ true,
+
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT);
+ Assertions.assertThat(actualSource).isEqualTo(expectedSource);
+ }
+
+ @Test
+ void testReleaseSnapshotMetadataOption() {
+ // testOptionalProperties cannot cover this option because it enables
+ // scan.newly-added-table.enabled, which the config rejects in
combination with the release.
+ Map<String, String> properties = getAllOptions();
+ properties.put("scan.incremental.snapshot.enabled", "true");
+ properties.put("scan.incremental.snapshot.metadata.release.enabled",
"true");
+
+ DynamicTableSource actualSource = createTableSource(SCHEMA,
properties);
+ MongoDBTableSource expectedSource =
+ new MongoDBTableSource(
+ SCHEMA,
+ SCHEME.defaultValue(),
+ MY_HOSTS,
+ USER,
+ PASSWORD,
+ MY_DATABASE,
+ MY_TABLE,
+ null,
+ StartupOptions.initial(),
+ null,
+ null,
+ null,
+ BATCH_SIZE_DEFAULT,
+ POLL_MAX_BATCH_SIZE_DEFAULT,
+ POLL_AWAIT_TIME_MILLIS_DEFAULT,
+ HEARTBEAT_INTERVAL_MILLIS_DEFAULT,
+ LOCAL_TIME_ZONE,
+ true,
+ CHUNK_META_GROUP_SIZE_DEFAULT,
+ SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB_DEFAULT,
+ SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SAMPLES_DEFAULT,
+ SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT,
+ FULL_DOCUMENT_PRE_POST_IMAGE_ENABLED_DEFAULT,
+ SCAN_NO_CURSOR_TIMEOUT_DEFAULT,
+ SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
+ SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
+
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue(),
true);
Assertions.assertThat(actualSource).isEqualTo(expectedSource);
+
+ // equals has to carry the field, or a source built without the option
would compare equal
+ Map<String, String> withoutRelease = getAllOptions();
+ withoutRelease.put("scan.incremental.snapshot.enabled", "true");
+
Assertions.assertThat(actualSource).isNotEqualTo(createTableSource(SCHEMA,
withoutRelease));
}
@Test
@@ -259,7 +312,8 @@ class MongoDBTableFactoryTest {
SCAN_NO_CURSOR_TIMEOUT_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
-
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue());
+
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue(),
+
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT);
expectedSource.producedDataType =
SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys = Arrays.asList("op_ts", "database_name",
"row_kind");
@@ -370,7 +424,8 @@ class MongoDBTableFactoryTest {
SCAN_NO_CURSOR_TIMEOUT_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
-
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue());
+
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue(),
+
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT);
Assertions.assertThat(actualSource).isEqualTo(expectedSource);
}
