This is an automated email from the ASF dual-hosted git repository.
leonardBang 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 09d467466b [FLINK-39775][mysql] Release snapshot split metadata after
entering the stream phase (#4418)
09d467466b is described below
commit 09d467466bc24fa7afd3f45ce5f13a45f4c18614
Author: Spoorthi Basu <[email protected]>
AuthorDate: Wed Sep 16 21:44:04 2026 -0400
[FLINK-39775][mysql] Release snapshot split metadata after entering the
stream phase (#4418)
This closes #4418.
---
.../docs/connectors/flink-sources/mysql-cdc.md | 7 +
.../docs/connectors/pipeline-connectors/mysql.md | 7 +
.../docs/connectors/flink-sources/mysql-cdc.md | 7 +
.../docs/connectors/pipeline-connectors/mysql.md | 7 +
.../mysql/factory/MySqlDataSourceFactory.java | 5 +
.../mysql/source/MySqlDataSourceOptions.java | 8 +
.../cdc/connectors/mysql/source/MySqlSource.java | 6 +-
.../mysql/source/MySqlSourceBuilder.java | 11 +
.../source/assigners/MySqlHybridSplitAssigner.java | 113 +++-
.../assigners/MySqlSnapshotSplitAssigner.java | 34 +-
.../state/PendingSplitsStateSerializer.java | 38 +-
.../state/SnapshotPendingSplitsState.java | 43 +-
.../mysql/source/config/MySqlSourceConfig.java | 9 +-
.../source/config/MySqlSourceConfigFactory.java | 22 +-
.../mysql/source/config/MySqlSourceOptions.java | 15 +
.../source/enumerator/MySqlSourceEnumerator.java | 71 ++-
.../events/BinlogSplitMetaAssembledEvent.java | 56 ++
.../mysql/source/events/BinlogSplitMetaEvent.java | 15 +-
.../mysql/source/reader/MySqlSourceReader.java | 19 +
.../connectors/mysql/table/MySqlTableSource.java | 7 +
.../mysql/table/MySqlTableSourceFactory.java | 4 +
.../connectors/mysql/source/MySqlSourceITCase.java | 39 ++
.../MySqlSnapshotMetadataReleaseTest.java | 625 +++++++++++++++++++++
.../state/PendingSplitsStateSerializerTest.java | 119 +++-
.../enumerator/MySqlSourceEnumeratorTest.java | 66 +++
.../mysql/source/reader/MySqlSourceReaderTest.java | 103 ++++
.../mysql/table/MySqlTableSourceFactoryTest.java | 25 +
27 files changed, 1461 insertions(+), 20 deletions(-)
diff --git a/docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md
b/docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md
index 85b14141be..cbf7ec5b03 100644
--- a/docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md
@@ -389,6 +389,13 @@ Flink SQL> SELECT * FROM orders;
For example: <code>'debezium.snapshot.mode' = 'never'</code>.
查看更多关于 <a
href="https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-connector-properties";>
Debezium 的 MySQL 连接器属性</a></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 进入 binlog 阶段后，释放 source coordinator
持有的快照分片元数据（已分配的分片、已完成分片的位点以及表结构），以降低快照分片数量非常大的作业的 JobManager 内存占用。默认关闭。与
scan.newly-added-table.enabled
不兼容：同时开启两者会导致作业启动失败，且已释放元数据的作业无法再开启动态加表功能。仅在成功完成一次 checkpoint 后才会释放；若未开启
checkpoint 或没有 checkpoint 完成，则会保留该元数据，因此该配置项在未开启 checkpoint 时不生效。</td>
+ </tr>
<tr>
<td>scan.incremental.close-idle-reader.enabled</td>
<td>optional</td>
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
index 04ec284455..3efb4c63a9 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
@@ -282,6 +282,13 @@ pipeline:
<td>Boolean</td>
<td>是否启用动态加表特性，默认关闭。 此配置项只有作业从savepoint/checkpoint启动时才生效。</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>是否在进入 binlog 阶段后，释放 source coordinator 持有的快照分片元数据，以降低 JobManager
内存占用。与 scan.newly-added-table.enabled 不兼容（同时开启两者会导致作业启动失败）；默认关闭。仅在成功完成一次
checkpoint 后才会释放；若未开启 checkpoint 或没有 checkpoint 完成，则会保留该元数据，因此该配置项在未开启
checkpoint 时不生效。</td>
+ </tr>
<tr>
<td>scan.binlog.newly-added-table.enabled</td>
<td>optional</td>
diff --git a/docs/content/docs/connectors/flink-sources/mysql-cdc.md
b/docs/content/docs/connectors/flink-sources/mysql-cdc.md
index 2d47d9b7d7..e857f7799b 100644
--- a/docs/content/docs/connectors/flink-sources/mysql-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/mysql-cdc.md
@@ -387,6 +387,13 @@ Only valid for cdc 1.x version. During a snapshot
operation, the connector will
For example: <code>'debezium.snapshot.mode' = 'never'</code>.
See more about the <a
href="https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-connector-properties";>Debezium's
MySQL Connector properties</a></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 binlog 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.incremental.close-idle-reader.enabled</td>
<td>optional</td>
diff --git a/docs/content/docs/connectors/pipeline-connectors/mysql.md
b/docs/content/docs/connectors/pipeline-connectors/mysql.md
index e6e61e2072..bd8974a7ea 100644
--- a/docs/content/docs/connectors/pipeline-connectors/mysql.md
+++ b/docs/content/docs/connectors/pipeline-connectors/mysql.md
@@ -289,6 +289,13 @@ pipeline:
<td>Boolean</td>
<td>Whether to enable scan the newly added tables feature or not, by
default is false. This option is only useful when we start the job from a
savepoint/checkpoint.</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 held by the source
coordinator after entering the binlog phase to reduce JobManager memory.
Incompatible with scan.newly-added-table.enabled (enabling both fails at
startup); disabled by default. Release happens only after a successful
checkpoint; if checkpointing is disabled or no checkpoint completes, the
metadata is retained, so this option has no effect without checkpointing.</td>
+ </tr>
<tr>
<td>scan.binlog.newly-added-table.enabled</td>
<td>optional</td>
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
index 1b9cea5618..1dfc6194dc 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
@@ -80,6 +80,7 @@ import static
org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOption
import static
org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
import static
org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
import static
org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
+import static
org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED;
import static
org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED;
import static
org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
import static
org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE;
@@ -161,6 +162,8 @@ public class MySqlDataSourceFactory implements
DataSourceFactory {
int connectMaxRetries = config.get(CONNECT_MAX_RETRIES);
int connectionPoolSize = config.get(CONNECTION_POOL_SIZE);
boolean scanNewlyAddedTableEnabled =
config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ boolean releaseSnapshotMetadataEnabled =
+ config.get(SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
boolean scanBinlogNewlyAddedTableEnabled =
config.get(SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
boolean isParsingOnLineSchemaChanges =
config.get(PARSE_ONLINE_SCHEMA_CHANGES);
@@ -216,6 +219,7 @@ public class MySqlDataSourceFactory implements
DataSourceFactory {
.debeziumProperties(getDebeziumProperties(configMap))
.jdbcProperties(getJdbcProperties(configMap))
.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
+
.releaseSnapshotMetadataEnabled(releaseSnapshotMetadataEnabled)
.parseOnLineSchemaChanges(isParsingOnLineSchemaChanges)
.treatTinyInt1AsBoolean(treatTinyInt1AsBoolean)
.useLegacyJsonFormat(useLegacyJsonFormat)
@@ -347,6 +351,7 @@ public class MySqlDataSourceFactory implements
DataSourceFactory {
options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);
options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
options.add(CHUNK_META_GROUP_SIZE);
options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
index 6aff556e7f..20f7b0efa3 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
@@ -242,6 +242,14 @@ public class MySqlDataSourceOptions {
.withDescription(
"Whether to scan the newly added tables or not, by
default is false. This option is only useful when we start the job from a
savepoint/checkpoint.");
+ @Experimental
+ public static final ConfigOption<Boolean>
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED =
+
ConfigOptions.key("scan.incremental.snapshot.metadata.release.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to release the snapshot split metadata
held by the source coordinator after entering the binlog phase to reduce
JobManager memory. Disabled by default. The release happens only after a
checkpoint completes, so it has no effect without checkpointing. Cannot be
enabled together with scan.newly-added-table.enabled (the job fails to
start).");
+
@Experimental
public static final ConfigOption<Boolean> SCHEMA_CHANGE_ENABLED =
ConfigOptions.key("schema-change.enabled")
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java
index 5fcbae1a1e..3d25182edc 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java
@@ -274,7 +274,11 @@ public class MySqlSource<T>
@Override
public SimpleVersionedSerializer<PendingSplitsState>
getEnumeratorCheckpointSerializer() {
- return new PendingSplitsStateSerializer(getSplitSerializer());
+ // Only write the v6 released flag when release is enabled. A
default-off job keeps the v5
+ // checkpoint format and stays restorable by an older connector build.
+ return new PendingSplitsStateSerializer(
+ getSplitSerializer(),
+
configFactory.createConfig(0).isReleaseSnapshotMetadataEnabled());
}
@Override
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java
index 93fa2a0d36..ea40417ec0 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java
@@ -211,6 +211,17 @@ public class MySqlSourceBuilder<T> {
return this;
}
+ /**
+ * Whether the source coordinator should release the snapshot split
metadata after entering the
+ * binlog phase to reduce JobManager memory. Incompatible with {@link
+ * #scanNewlyAddedTableEnabled(boolean)}; disabled by default.
+ */
+ public MySqlSourceBuilder<T> releaseSnapshotMetadataEnabled(
+ boolean releaseSnapshotMetadataEnabled) {
+
this.configFactory.releaseSnapshotMetadataEnabled(releaseSnapshotMetadataEnabled);
+ return this;
+ }
+
/** Specifies the startup options. */
public MySqlSourceBuilder<T> startupOptions(StartupOptions startupOptions)
{
this.configFactory.startupOptions(startupOptions);
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java
index 4a25491b23..b81327820b 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java
@@ -21,16 +21,20 @@ import
org.apache.flink.api.connector.source.SplitEnumeratorContext;
import
org.apache.flink.cdc.connectors.mysql.source.assigners.state.HybridPendingSplitsState;
import
org.apache.flink.cdc.connectors.mysql.source.assigners.state.PendingSplitsState;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaAssembledEvent;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
import
org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit;
import
org.apache.flink.cdc.connectors.mysql.source.split.MySqlSchemalessSnapshotSplit;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplit;
+import org.apache.flink.util.Preconditions;
import io.debezium.relational.TableId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -52,6 +56,20 @@ public class MySqlHybridSplitAssigner implements
MySqlSplitAssigner {
private boolean isBinlogSplitAssigned;
+ /** Whether the reader has reported holding the complete binlog split. */
+ private boolean binlogSplitMetaAssembled;
+
+ /** Checkpoint at which the metadata release was scheduled; released once
it completes. */
+ @Nullable private Long checkpointIdToReleaseMeta;
+
+ /**
+ * Generation of the current binlog split assignment, bumped on every
add-back. It rides on the
+ * meta groups served to the reader and comes back in the assembled event,
so a stale event from
+ * a failed attempt (older generation) can't arm the release for a fresh
reader. Not
+ * checkpointed: it resets to 0 on restore, which is fine because the
reader re-learns it.
+ */
+ private long binlogAssignmentGeneration;
+
private final MySqlSnapshotSplitAssigner snapshotSplitAssigner;
public MySqlHybridSplitAssigner(
@@ -97,6 +115,32 @@ public class MySqlHybridSplitAssigner implements
MySqlSplitAssigner {
this.snapshotSplitAssigner = snapshotSplitAssigner;
this.isBinlogSplitAssigned = isBinlogSplitAssigned;
this.splitMetaGroupSize = splitMetaGroupSize;
+ // Fail fast on an incompatible restore. A job that released its
snapshot metadata cannot
+ // later enable scan.newly-added-table, because the splits and schemas
that flow needs are
+ // gone. Reject it loudly instead of silently building an inconsistent
binlog split.
+ if (sourceConfig.isScanNewlyAddedTableEnabled()
+ && isBinlogSplitAssigned
+ && snapshotSplitAssigner.isSnapshotMetaReleased()) {
+ throw new IllegalStateException(
+ "scan.newly-added-table.enabled cannot be turned on for a
job that previously "
+ + "released its snapshot split metadata "
+ +
"(scan.incremental.snapshot.metadata.release.enabled=true). The "
+ + "assigned splits, finished offsets and table
schemas needed to capture "
+ + "newly added tables are no longer in state.
Start the job from a fresh "
+ + "state, or keep metadata release disabled if
newly-added-table scanning "
+ + "is required.");
+ }
+ // Fail fast if a job that already released its snapshot metadata is
restarted with the
+ // release option off. The metadata is gone from state, so falling
back to the v5 format
+ // would drop the released marker. Keep the option enabled or start
from a fresh state.
+ if (!sourceConfig.isReleaseSnapshotMetadataEnabled()
+ && snapshotSplitAssigner.isSnapshotMetaReleased()) {
+ throw new IllegalStateException(
+ "scan.incremental.snapshot.metadata.release.enabled cannot
be turned off for a "
+ + "job that already released its snapshot split
metadata. The assigned "
+ + "splits, finished offsets and table schemas are
no longer in state. "
+ + "Keep the option enabled, or start the job from
a fresh state.");
+ }
}
@Override
@@ -159,15 +203,39 @@ public class MySqlHybridSplitAssigner implements
MySqlSplitAssigner {
if (split.isSnapshotSplit()) {
snapshotSplits.add(split);
} else {
- // we don't store the split, but will re-create binlog split
later
+ // The binlog split is handed back for re-assignment; we don't
store it, but will
+ // re-create it later. Mark it unassigned and invalidate any
in-flight assembly.
isBinlogSplitAssigned = false;
+ onBinlogReaderReset();
}
}
snapshotSplitAssigner.addSplits(snapshotSplits);
}
+ /**
+ * Invalidates any in-flight binlog split assembly when the binlog reader
resets, so a stale
+ * assembled event from the failed attempt cannot arm a release. Also runs
when the split is not
+ * handed back, since a checkpoint-covered reset produces an empty
add-back.
+ */
+ public void onBinlogReaderReset() {
+ binlogSplitMetaAssembled = false;
+ checkpointIdToReleaseMeta = null;
+ binlogAssignmentGeneration++;
+ }
+
@Override
public PendingSplitsState snapshotState(long checkpointId) {
+ // Schedule the metadata release once the binlog split is assigned and
the reader has
+ // assembled it; it happens in notifyCheckpointComplete. Gated behind
the opt-in; the
+ // newly-added-table term is defensive, since createConfig already
rejects enabling both.
+ if (isBinlogSplitAssigned
+ && binlogSplitMetaAssembled
+ && checkpointIdToReleaseMeta == null
+ && !snapshotSplitAssigner.isSnapshotMetaReleased()
+ && sourceConfig.isReleaseSnapshotMetadataEnabled()
+ && !sourceConfig.isScanNewlyAddedTableEnabled()) {
+ checkpointIdToReleaseMeta = checkpointId;
+ }
return new HybridPendingSplitsState(
snapshotSplitAssigner.snapshotState(checkpointId),
isBinlogSplitAssigned);
}
@@ -175,6 +243,45 @@ public class MySqlHybridSplitAssigner implements
MySqlSplitAssigner {
@Override
public void notifyCheckpointComplete(long checkpointId) {
snapshotSplitAssigner.notifyCheckpointComplete(checkpointId);
+ // Release the snapshot metadata only after the checkpoint covering
the binlog split
+ // assignment completes. Doing it here (not in snapshotState) keeps
the assignment
+ // checkpoint-covered, so addSplitsBack can never return the split to
an emptied assigner.
+ if (checkpointIdToReleaseMeta != null
+ && checkpointId >= checkpointIdToReleaseMeta
+ && !snapshotSplitAssigner.isSnapshotMetaReleased()) {
+ snapshotSplitAssigner.releaseSnapshotMetadata();
+ }
+ }
+
+ /**
+ * Marks that the reader holds the complete binlog split, arming the
metadata release. A
+ * reported generation that is neither the current one nor the inline
sentinel is ignored, which
+ * drops a stale event from a reader attempt that an add-back has already
superseded.
+ */
+ public void onBinlogSplitMetaAssembled(long reportedGeneration) {
+ // An inline reader requests no meta groups, so it reports
COMPLETE_WITHOUT_META_GENERATION;
+ // honor that regardless, since it never needs the coordinator's
metadata. Otherwise require
+ // an exact match, which drops a stale event from a failed attempt
superseded by an
+ // add-back.
+ if (reportedGeneration !=
BinlogSplitMetaAssembledEvent.COMPLETE_WITHOUT_META_GENERATION
+ && reportedGeneration != binlogAssignmentGeneration) {
+ LOG.info(
+ "Ignoring binlog split assembled event for generation {};
current generation is {}.",
+ reportedGeneration,
+ binlogAssignmentGeneration);
+ return;
+ }
+ this.binlogSplitMetaAssembled = true;
+ }
+
+ /** Returns the current binlog split assignment generation. */
+ public long getBinlogAssignmentGeneration() {
+ return binlogAssignmentGeneration;
+ }
+
+ /** Returns whether the snapshot split metadata has been released. */
+ public boolean isSnapshotMetaReleased() {
+ return snapshotSplitAssigner.isSnapshotMetaReleased();
}
@Override
@@ -205,6 +312,10 @@ public class MySqlHybridSplitAssigner implements
MySqlSplitAssigner {
//
--------------------------------------------------------------------------------------------
private MySqlBinlogSplit createBinlogSplit() {
+ Preconditions.checkState(
+ !snapshotSplitAssigner.isSnapshotMetaReleased(),
+ "Snapshot metadata was already released; the binlog split must
not be re-created "
+ + "after that.");
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
new
ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java
index f51df018d4..8f8794b4c7 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java
@@ -108,6 +108,9 @@ public class MySqlSnapshotSplitAssigner implements
MySqlSplitAssigner {
@Nullable private Long checkpointIdToFinish;
+ /** Whether the snapshot split metadata has been released. */
+ private boolean snapshotMetaReleased;
+
public MySqlSnapshotSplitAssigner(
MySqlSourceConfig sourceConfig,
int currentParallelism,
@@ -127,6 +130,7 @@ public class MySqlSnapshotSplitAssigner implements
MySqlSplitAssigner {
isTableIdCaseSensitive,
true,
ChunkSplitterState.NO_SPLITTING_TABLE_STATE,
+ false,
enumeratorContext);
}
@@ -148,6 +152,7 @@ public class MySqlSnapshotSplitAssigner implements
MySqlSplitAssigner {
checkpoint.isTableIdCaseSensitive(),
checkpoint.isRemainingTablesCheckpointed(),
checkpoint.getChunkSplitterState(),
+ checkpoint.isSnapshotMetaReleased(),
enumeratorContext);
}
@@ -164,6 +169,7 @@ public class MySqlSnapshotSplitAssigner implements
MySqlSplitAssigner {
boolean isTableIdCaseSensitive,
boolean isRemainingTablesCheckpointed,
ChunkSplitterState chunkSplitterState,
+ boolean snapshotMetaReleased,
SplitEnumeratorContext<MySqlSplit> enumeratorContext) {
this.sourceConfig = sourceConfig;
this.currentParallelism = currentParallelism;
@@ -181,6 +187,9 @@ public class MySqlSnapshotSplitAssigner implements
MySqlSplitAssigner {
this.partition =
new
MySqlPartition(sourceConfig.getMySqlConnectorConfig().getLogicalName());
this.enumeratorContext = enumeratorContext;
+ // The released flag is persisted in the checkpoint
(PendingSplitsStateSerializer v6), so
+ // restore it directly rather than inferring it from the state shape.
+ this.snapshotMetaReleased = snapshotMetaReleased;
}
@Override
@@ -462,7 +471,8 @@ public class MySqlSnapshotSplitAssigner implements
MySqlSplitAssigner {
remainingTables,
isTableIdCaseSensitive,
true,
- chunkSplitter.snapshotState(checkpointId));
+ chunkSplitter.snapshotState(checkpointId),
+ snapshotMetaReleased);
// we need a complete checkpoint before mark this assigner to be
finished, to wait for
// all records of snapshot splits are completely processed
if (checkpointIdToFinish == null
@@ -571,6 +581,28 @@ public class MySqlSnapshotSplitAssigner implements
MySqlSplitAssigner {
return splitFinishedOffsets;
}
+ /**
+ * Releases the heavyweight snapshot split metadata (assigned splits,
finished offsets, table
+ * schemas) after the binlog phase begins, so it is no longer held in the
coordinator nor
+ * checkpointed. Keeps {@link #alreadyProcessedTables} and the assigner
status so a restore does
+ * not re-discover tables. Only called once the binlog split is assigned
and checkpoint-covered.
+ */
+ public void releaseSnapshotMetadata() {
+ if (snapshotMetaReleased) {
+ return;
+ }
+ assignedSplits.clear();
+ splitFinishedOffsets.clear();
+ tableSchemas.clear();
+ snapshotMetaReleased = true;
+ LOG.info("Released snapshot split metadata after entering the binlog
phase.");
+ }
+
+ /** Returns whether {@link #releaseSnapshotMetadata()} has already been
performed. */
+ public boolean isSnapshotMetaReleased() {
+ return snapshotMetaReleased;
+ }
+
//
-------------------------------------------------------------------------------------------
/**
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializer.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializer.java
index 0df1ee9799..212b9a8c24 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializer.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializer.java
@@ -45,8 +45,13 @@ import java.util.Map;
*/
public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<PendingSplitsState> {
- // TODO: need proper implementation of the new version
- private static final int VERSION = 5;
+ // Checkpoint format version. Version 6 (FLINK-39775) additionally
persists the
+ // snapshot-metadata released flag, and is written only when metadata
release is enabled. A job
+ // that leaves the option off keeps writing version 5, so it can still be
restored by an older
+ // connector build that does not understand the flag.
+ private static final int VERSION_WITHOUT_RELEASED_FLAG = 5;
+ private static final int VERSION_WITH_RELEASED_FLAG = 6;
+
private static final ThreadLocal<DataOutputSerializer> SERIALIZER_CACHE =
ThreadLocal.withInitial(() -> new DataOutputSerializer(64));
@@ -55,19 +60,27 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
private static final int HYBRID_PENDING_SPLITS_STATE_FLAG = 3;
private final SimpleVersionedSerializer<MySqlSplit> splitSerializer;
+ private final boolean releaseSnapshotMetadataEnabled;
- public PendingSplitsStateSerializer(SimpleVersionedSerializer<MySqlSplit>
splitSerializer) {
+ public PendingSplitsStateSerializer(
+ SimpleVersionedSerializer<MySqlSplit> splitSerializer,
+ boolean releaseSnapshotMetadataEnabled) {
this.splitSerializer = splitSerializer;
+ this.releaseSnapshotMetadataEnabled = releaseSnapshotMetadataEnabled;
}
@Override
public int getVersion() {
- return VERSION;
+ return releaseSnapshotMetadataEnabled
+ ? VERSION_WITH_RELEASED_FLAG
+ : VERSION_WITHOUT_RELEASED_FLAG;
}
@Override
public byte[] serialize(PendingSplitsState state) throws IOException {
- // optimization: the splits lazily cache their own serialized form
+ // optimization: the splits lazily cache their own serialized form. A
job serializes with a
+ // single serializer whose released-flag setting is fixed, so the
cached payload never mixes
+ // the v5 and v6 forms; the cache is transient and starts null after a
restore.
if (state.serializedFormCache != null) {
return state.serializedFormCache;
}
@@ -110,6 +123,7 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
case 3:
case 4:
case 5:
+ case 6:
return deserializePendingSplitsState(version, serialized);
default:
throw new IOException("Unknown version: " + version);
@@ -176,6 +190,11 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
new Object[]
{chunkSplitterState.getNextChunkStart().getValue()}));
out.writeInt(chunkSplitterState.getNextChunkId());
}
+ // v6 (FLINK-39775): the released flag is written only when release is
enabled. A job with
+ // the option off keeps the v5 format and stays restorable by an older
connector build.
+ if (releaseSnapshotMetadataEnabled) {
+ out.writeBoolean(state.isSnapshotMetaReleased());
+ }
}
private void serializeHybridPendingSplitsState(
@@ -299,6 +318,12 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
nextChunkId = in.readInt();
}
}
+ // v6 (FLINK-39775): the released flag is persisted; older versions
never released, so it
+ // defaults to false.
+ boolean snapshotMetaReleased = false;
+ if (version >= 6) {
+ snapshotMetaReleased = in.readBoolean();
+ }
return new SnapshotPendingSplitsState(
alreadyProcessedTables,
remainingSchemalessSplits,
@@ -314,7 +339,8 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
: new ChunkSplitterState(
splittingTableId,
ChunkSplitterState.ChunkBound.middleOf(nextChunkStart),
- nextChunkId));
+ nextChunkId),
+ snapshotMetaReleased);
}
private HybridPendingSplitsState deserializeHybridPendingSplitsState(
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/SnapshotPendingSplitsState.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/SnapshotPendingSplitsState.java
index 55df2e7daf..28ff7b73ee 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/SnapshotPendingSplitsState.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/SnapshotPendingSplitsState.java
@@ -75,6 +75,9 @@ public class SnapshotPendingSplitsState extends
PendingSplitsState {
/** The data structure to record the state of a {@link ChunkSplitter}. */
private final ChunkSplitterState chunkSplitterState;
+ /** Whether the heavyweight snapshot split metadata has been released
(FLINK-39775). */
+ private final boolean snapshotMetaReleased;
+
public SnapshotPendingSplitsState(
List<TableId> alreadyProcessedTables,
List<MySqlSchemalessSnapshotSplit> remainingSplits,
@@ -86,6 +89,32 @@ public class SnapshotPendingSplitsState extends
PendingSplitsState {
boolean isTableIdCaseSensitive,
boolean isRemainingTablesCheckpointed,
ChunkSplitterState chunkSplitterState) {
+ this(
+ alreadyProcessedTables,
+ remainingSplits,
+ assignedSplits,
+ tableSchemas,
+ splitFinishedOffsets,
+ assignerStatus,
+ remainingTables,
+ isTableIdCaseSensitive,
+ isRemainingTablesCheckpointed,
+ chunkSplitterState,
+ false);
+ }
+
+ public SnapshotPendingSplitsState(
+ List<TableId> alreadyProcessedTables,
+ List<MySqlSchemalessSnapshotSplit> remainingSplits,
+ LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits,
+ Map<TableId, TableChange> tableSchemas,
+ Map<String, BinlogOffset> splitFinishedOffsets,
+ AssignerStatus assignerStatus,
+ List<TableId> remainingTables,
+ boolean isTableIdCaseSensitive,
+ boolean isRemainingTablesCheckpointed,
+ ChunkSplitterState chunkSplitterState,
+ boolean snapshotMetaReleased) {
// FLINK-38061: make defensive copy to avoid potential concurrent
modification of the
// collections.
this.alreadyProcessedTables = new ArrayList<>(alreadyProcessedTables);
@@ -98,6 +127,7 @@ public class SnapshotPendingSplitsState extends
PendingSplitsState {
this.isRemainingTablesCheckpointed = isRemainingTablesCheckpointed;
this.tableSchemas = new HashMap<>(tableSchemas);
this.chunkSplitterState = chunkSplitterState;
+ this.snapshotMetaReleased = snapshotMetaReleased;
}
public List<TableId> getAlreadyProcessedTables() {
@@ -140,6 +170,11 @@ public class SnapshotPendingSplitsState extends
PendingSplitsState {
return chunkSplitterState;
}
+ /** Whether the heavyweight snapshot split metadata has been released
(FLINK-39775). */
+ public boolean isSnapshotMetaReleased() {
+ return snapshotMetaReleased;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -157,7 +192,8 @@ public class SnapshotPendingSplitsState extends
PendingSplitsState {
&& Objects.equals(remainingSplits, that.remainingSplits)
&& Objects.equals(assignedSplits, that.assignedSplits)
&& Objects.equals(splitFinishedOffsets,
that.splitFinishedOffsets)
- && Objects.equals(chunkSplitterState, that.chunkSplitterState);
+ && Objects.equals(chunkSplitterState, that.chunkSplitterState)
+ && snapshotMetaReleased == that.snapshotMetaReleased;
}
@Override
@@ -171,7 +207,8 @@ public class SnapshotPendingSplitsState extends
PendingSplitsState {
assignerStatus,
isTableIdCaseSensitive,
isRemainingTablesCheckpointed,
- chunkSplitterState);
+ chunkSplitterState,
+ snapshotMetaReleased);
}
@Override
@@ -195,6 +232,8 @@ public class SnapshotPendingSplitsState extends
PendingSplitsState {
+ isRemainingTablesCheckpointed
+ ", chunkSplitterState="
+ chunkSplitterState
+ + ", snapshotMetaReleased="
+ + snapshotMetaReleased
+ '}';
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
index 6b87dfef9a..85e1cb55e4 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java
@@ -73,6 +73,7 @@ public class MySqlSourceConfig implements Serializable {
private final boolean parseOnLineSchemaChanges;
public static boolean useLegacyJsonFormat = true;
private final boolean assignUnboundedChunkFirst;
+ private final boolean releaseSnapshotMetadataEnabled;
//
--------------------------------------------------------------------------------------------
// Debezium Configurations
@@ -113,7 +114,8 @@ public class MySqlSourceConfig implements Serializable {
boolean parseOnLineSchemaChanges,
boolean treatTinyInt1AsBoolean,
boolean useLegacyJsonFormat,
- boolean assignUnboundedChunkFirst) {
+ boolean assignUnboundedChunkFirst,
+ boolean releaseSnapshotMetadataEnabled) {
this.hostname = checkNotNull(hostname);
this.port = port;
this.username = checkNotNull(username);
@@ -161,6 +163,7 @@ public class MySqlSourceConfig implements Serializable {
this.treatTinyInt1AsBoolean = treatTinyInt1AsBoolean;
this.useLegacyJsonFormat = useLegacyJsonFormat;
this.assignUnboundedChunkFirst = assignUnboundedChunkFirst;
+ this.releaseSnapshotMetadataEnabled = releaseSnapshotMetadataEnabled;
}
public String getHostname() {
@@ -248,6 +251,10 @@ public class MySqlSourceConfig implements Serializable {
return scanNewlyAddedTableEnabled;
}
+ public boolean isReleaseSnapshotMetadataEnabled() {
+ return releaseSnapshotMetadataEnabled;
+ }
+
public boolean isCloseIdleReaders() {
return closeIdleReaders;
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java
index 569b62232d..495acac4d9 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java
@@ -37,6 +37,7 @@ import java.util.UUID;
import static
org.apache.flink.cdc.connectors.mysql.source.utils.EnvironmentUtils.checkSupportCheckpointsAfterTasksFinished;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
/** A factory to construct {@link MySqlSourceConfig}. */
@Internal
@@ -68,6 +69,7 @@ public class MySqlSourceConfigFactory implements Serializable
{
private boolean includeHeartbeatEvents = false;
private boolean includeTransactionMetadataEvents = false;
private boolean scanNewlyAddedTableEnabled = false;
+ private boolean releaseSnapshotMetadataEnabled = false;
private boolean closeIdleReaders = false;
private Properties jdbcProperties;
private Duration heartbeatInterval =
MySqlSourceOptions.HEARTBEAT_INTERVAL.defaultValue();
@@ -258,6 +260,17 @@ public class MySqlSourceConfigFactory implements
Serializable {
return this;
}
+ /**
+ * Whether the source coordinator should release the snapshot split
metadata after entering the
+ * binlog phase to reduce JobManager memory. Incompatible with {@link
+ * #scanNewlyAddedTableEnabled(boolean)}; disabled by default.
+ */
+ public MySqlSourceConfigFactory releaseSnapshotMetadataEnabled(
+ boolean releaseSnapshotMetadataEnabled) {
+ this.releaseSnapshotMetadataEnabled = releaseSnapshotMetadataEnabled;
+ return this;
+ }
+
/** Custom properties that will overwrite the default JDBC connection URL.
*/
public MySqlSourceConfigFactory jdbcProperties(Properties jdbcProperties) {
this.jdbcProperties = jdbcProperties;
@@ -355,6 +368,12 @@ public class MySqlSourceConfigFactory implements
Serializable {
/** Creates a new {@link MySqlSourceConfig} for the given subtask {@code
subtaskId}. */
public MySqlSourceConfig createConfig(int subtaskId, String serverName) {
checkSupportCheckpointsAfterTasksFinished(closeIdleReaders);
+ checkState(
+ !(scanNewlyAddedTableEnabled &&
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.");
Properties props = new Properties();
props.setProperty("database.server.name", serverName);
props.setProperty("database.hostname", checkNotNull(hostname));
@@ -444,6 +463,7 @@ public class MySqlSourceConfigFactory implements
Serializable {
parseOnLineSchemaChanges,
treatTinyInt1AsBoolean,
useLegacyJsonFormat,
- assignUnboundedChunkFirst);
+ assignUnboundedChunkFirst,
+ releaseSnapshotMetadataEnabled);
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java
index a8e143f5fc..8a12a0c9e7 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java
@@ -243,6 +243,21 @@ public class MySqlSourceOptions {
.withDescription(
"Whether capture the scan the newly added tables
or not, by default is false. This option is only useful when we start the job
from a savepoint/checkpoint.");
+ @Experimental
+ public static final ConfigOption<Boolean>
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED =
+
ConfigOptions.key("scan.incremental.snapshot.metadata.release.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "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 binlog phase,
to reduce JobManager memory"
+ + " pressure on jobs with a very large
number of snapshot splits. Disabled"
+ + " by default. The release happens only
after a checkpoint completes, so it has no"
+ + " effect without checkpointing. Cannot
be enabled together with"
+ + " scan.newly-added-table.enabled (the
job fails to start), since that"
+ + " flow needs the metadata that would be
released.");
+
@Experimental
public static final ConfigOption<String>
SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN =
ConfigOptions.key("scan.incremental.snapshot.chunk.key-column")
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java
index 0451db76aa..ff0279e4aa 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java
@@ -27,6 +27,7 @@ import
org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSplitAssigner
import
org.apache.flink.cdc.connectors.mysql.source.assigners.state.PendingSplitsState;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitAssignedEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaAssembledEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaRequestEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitUpdateAckEvent;
@@ -126,6 +127,17 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
if (binlogSplit.isPresent()) {
LOG.info("The enumerator adds add binlog split back: {}",
binlogSplit);
this.binlogSplitTaskId = null;
+ // The split is coming back to the coordinator; addSplits below
resets the assembly.
+ } else if (binlogSplitTaskId != null
+ && binlogSplitTaskId.equals(subtaskId)
+ && splitAssigner instanceof MySqlHybridSplitAssigner) {
+ // The binlog holder reset but handed nothing back (a
checkpoint-covered reset the
+ // reader restores itself), so addSplits will not see it;
invalidate the assembly here.
+ LOG.info(
+ "The binlog reader on subtask {} reset without handing its
split back; "
+ + "invalidating any assembled binlog split
metadata.",
+ subtaskId);
+ ((MySqlHybridSplitAssigner) splitAssigner).onBinlogReaderReset();
}
if (!CollectionUtil.isNullOrEmpty(splits)) {
splitAssigner.addSplits(splits);
@@ -179,7 +191,28 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
LOG.info(
"The enumerator receives notice from subtask {} for the
binlog split assignment. ",
subtaskId);
- binlogSplitTaskId = subtaskId;
+ // Only learn the holder from this event when it is unknown (e.g.
after a coordinator
+ // restart). A real reassignment sets it via assignSplits, so an
event naming a
+ // different subtask than the known holder is stale and must not
overwrite it.
+ if (binlogSplitTaskId == null) {
+ binlogSplitTaskId = subtaskId;
+ } else if (!binlogSplitTaskId.equals(subtaskId)) {
+ LOG.info(
+ "Ignoring a stale binlog split assigned event from
subtask {}; the binlog "
+ + "split is currently held by subtask {}.",
+ subtaskId,
+ binlogSplitTaskId);
+ }
+ } else if (sourceEvent instanceof BinlogSplitMetaAssembledEvent) {
+ LOG.info(
+ "The enumerator receives notice from subtask {} that the
binlog split metadata has been fully assembled. ",
+ subtaskId);
+ if (splitAssigner instanceof MySqlHybridSplitAssigner) {
+ ((MySqlHybridSplitAssigner) splitAssigner)
+ .onBinlogSplitMetaAssembled(
+ ((BinlogSplitMetaAssembledEvent) sourceEvent)
+ .getBinlogAssignmentGeneration());
+ }
}
}
@@ -191,6 +224,13 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
@Override
public void notifyCheckpointComplete(long checkpointId) {
splitAssigner.notifyCheckpointComplete(checkpointId);
+ // Once the assigner has released the snapshot metadata, drop the
enumerator's cached binlog
+ // split meta groups too: it is a second copy of the same
finished-split infos.
+ if (binlogSplitMeta != null
+ && splitAssigner instanceof MySqlHybridSplitAssigner
+ && ((MySqlHybridSplitAssigner)
splitAssigner).isSnapshotMetaReleased()) {
+ binlogSplitMeta = null;
+ }
// binlog split may be available after checkpoint complete
assignSplits();
}
@@ -291,6 +331,19 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
}
private void sendBinlogMeta(int subTask, BinlogSplitMetaRequestEvent
requestEvent) {
+ // After the snapshot metadata has been released, the finished-split
infos are gone. A meta
+ // request arriving now can only come from a failed reader attempt: a
live reader that has
+ // assembled the binlog split recovers it from its own checkpointed
state and never
+ // re-requests. Ignore it instead of rebuilding from an emptied
assigner, which would throw
+ // FlinkRuntimeException and fail the job.
+ if (splitAssigner instanceof MySqlHybridSplitAssigner
+ && ((MySqlHybridSplitAssigner)
splitAssigner).isSnapshotMetaReleased()) {
+ LOG.info(
+ "Ignoring a stale binlog split meta request from subtask
{} received after the "
+ + "snapshot metadata was released.",
+ subTask);
+ return;
+ }
// initialize once
if (binlogSplitMeta == null) {
final List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos =
@@ -319,7 +372,8 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
requestEvent.getSplitId(),
requestMetaGroupId,
null,
- totalFinishedSplitSizeOfEnumerator);
+ totalFinishedSplitSizeOfEnumerator,
+ currentBinlogAssignmentGeneration());
context.sendEventToSourceReader(subTask, metadataEvent);
} else if (binlogSplitMeta.size() > requestMetaGroupId) {
List<FinishedSnapshotSplitInfo> metaToSend =
binlogSplitMeta.get(requestMetaGroupId);
@@ -330,7 +384,8 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
metaToSend.stream()
.map(FinishedSnapshotSplitInfo::serialize)
.collect(Collectors.toList()),
- totalFinishedSplitSizeOfEnumerator);
+ totalFinishedSplitSizeOfEnumerator,
+ currentBinlogAssignmentGeneration());
context.sendEventToSourceReader(subTask, metadataEvent);
} else {
throw new FlinkRuntimeException(
@@ -343,6 +398,16 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
}
}
+ /**
+ * The current binlog split assignment generation, or 0 when not in hybrid
mode. Stamped onto
+ * the meta groups served to a reader so a stale assembled event can be
filtered out.
+ */
+ private long currentBinlogAssignmentGeneration() {
+ return splitAssigner instanceof MySqlHybridSplitAssigner
+ ? ((MySqlHybridSplitAssigner)
splitAssigner).getBinlogAssignmentGeneration()
+ : 0L;
+ }
+
private void handleLatestFinishedSplitNumberRequest(int subTask) {
if (splitAssigner instanceof MySqlHybridSplitAssigner) {
context.sendEventToSourceReader(
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/events/BinlogSplitMetaAssembledEvent.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/events/BinlogSplitMetaAssembledEvent.java
new file mode 100644
index 0000000000..79a902caa1
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/events/BinlogSplitMetaAssembledEvent.java
@@ -0,0 +1,56 @@
+/*
+ * 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.mysql.source.events;
+
+import org.apache.flink.api.connector.source.SourceEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.enumerator.MySqlSourceEnumerator;
+import org.apache.flink.cdc.connectors.mysql.source.reader.MySqlSourceReader;
+import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit;
+
+/**
+ * The {@link SourceEvent} that {@link MySqlSourceReader} sends to {@link
MySqlSourceEnumerator}
+ * once it holds a complete {@link MySqlBinlogSplit}, signalling that all
finished snapshot split
+ * metadata has been assembled by the reader. This lets the enumerator release
the snapshot split
+ * metadata it retains. Idempotent: re-sent whenever a complete binlog split
(re-)enters reading.
+ *
+ * <p>Carries the generation the reader assembled under, or {@link
+ * #COMPLETE_WITHOUT_META_GENERATION} for an inline split. The coordinator
uses it to drop a stale
+ * event left over from a failed reader attempt, which would otherwise arm the
release before a
+ * freshly-assigned reader has re-assembled.
+ */
+public class BinlogSplitMetaAssembledEvent implements SourceEvent {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Reported instead of a real generation when the reader held a complete
split inline and never
+ * requested meta groups. The coordinator releases on it regardless of
generation, since an
+ * inline reader never needs the coordinator's metadata.
+ */
+ public static final long COMPLETE_WITHOUT_META_GENERATION = -1L;
+
+ private final long binlogAssignmentGeneration;
+
+ public BinlogSplitMetaAssembledEvent(long binlogAssignmentGeneration) {
+ this.binlogAssignmentGeneration = binlogAssignmentGeneration;
+ }
+
+ public long getBinlogAssignmentGeneration() {
+ return binlogAssignmentGeneration;
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/events/BinlogSplitMetaEvent.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/events/BinlogSplitMetaEvent.java
index bdd553a73a..02bf1f5a4a 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/events/BinlogSplitMetaEvent.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/events/BinlogSplitMetaEvent.java
@@ -48,15 +48,24 @@ public class BinlogSplitMetaEvent implements SourceEvent {
private final int totalFinishedSplitSize;
+ /**
+ * The binlog assignment generation current at the coordinator when this
meta group was served.
+ * The reader echoes it back in {@link BinlogSplitMetaAssembledEvent} so
the coordinator can
+ * tell a fresh assembled signal from a stale one left by a failed reader
attempt.
+ */
+ private final long binlogAssignmentGeneration;
+
public BinlogSplitMetaEvent(
String splitId,
int metaGroupId,
@Nullable List<byte[]> metaGroup,
- int totalFinishedSplitSize) {
+ int totalFinishedSplitSize,
+ long binlogAssignmentGeneration) {
this.splitId = splitId;
this.metaGroupId = metaGroupId;
this.metaGroup = metaGroup;
this.totalFinishedSplitSize = totalFinishedSplitSize;
+ this.binlogAssignmentGeneration = binlogAssignmentGeneration;
}
public String getSplitId() {
@@ -74,4 +83,8 @@ public class BinlogSplitMetaEvent implements SourceEvent {
public int getTotalFinishedSplitSize() {
return totalFinishedSplitSize;
}
+
+ public long getBinlogAssignmentGeneration() {
+ return binlogAssignmentGeneration;
+ }
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
index b3f421ceec..4d6f4a8620 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
@@ -22,6 +22,7 @@ import
org.apache.flink.cdc.common.annotation.VisibleForTesting;
import org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitAssignedEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaAssembledEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaRequestEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitUpdateAckEvent;
@@ -81,6 +82,16 @@ public class MySqlSourceReader<T>
private final MySqlRecordEmitter<T> recordEmitter;
private final MySqlPartition partition;
+ /**
+ * Generation the reader learned from its served meta groups, echoed in
{@link
+ * BinlogSplitMetaAssembledEvent} so the coordinator can drop a stale
signal from a failed
+ * attempt. Stays {@link
BinlogSplitMetaAssembledEvent#COMPLETE_WITHOUT_META_GENERATION} for an
+ * inline split that requested none. Not checkpointed: on restart it
resets to that sentinel,
+ * which is safe because a reader reports the sentinel only while it holds
a complete split.
+ */
+ private long binlogAssignmentGeneration =
+ BinlogSplitMetaAssembledEvent.COMPLETE_WITHOUT_META_GENERATION;
+
public MySqlSourceReader(
Supplier<MySqlSplitReader> splitReaderSupplier,
MySqlRecordEmitter<T> recordEmitter,
@@ -285,6 +296,11 @@ public class MySqlSourceReader<T>
discoverTableSchemasForBinlogSplit(
binlogSplit, sourceConfig,
checkNewlyAddedTableSchema);
unfinishedSplits.add(mySqlBinlogSplit);
+ // binlog split is complete: tell the enumerator so it can
release the snapshot
+ // metadata. Also fires on restore of a complete split,
which is fine. The
+ // generation ties this signal to the current assignment.
+ context.sendSourceEventToCoordinator(
+ new
BinlogSplitMetaAssembledEvent(binlogAssignmentGeneration));
}
LOG.info(
"Source reader {} received the binlog split : {}.",
subtaskId, binlogSplit);
@@ -400,6 +416,9 @@ public class MySqlSourceReader<T>
private void fillMetadataForBinlogSplit(BinlogSplitMetaEvent
metadataEvent) {
MySqlBinlogSplit binlogSplit =
uncompletedBinlogSplits.get(metadataEvent.getSplitId());
if (binlogSplit != null) {
+ // Track the generation this metadata was served under, so the
assembled event we send
+ // once the split is complete is tied to the current assignment.
+ this.binlogAssignmentGeneration =
metadataEvent.getBinlogAssignmentGeneration();
final int receivedMetaGroupId = metadataEvent.getMetaGroupId();
final int receivedTotalFinishedSplitSize =
metadataEvent.getTotalFinishedSplitSize();
final int expectedMetaGroupId =
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java
index 2a1f051943..d44d0d4fe7 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java
@@ -93,6 +93,7 @@ public class MySqlTableSource implements ScanTableSource,
SupportsReadingMetadat
private final double distributionFactorLower;
private final StartupOptions startupOptions;
private final boolean scanNewlyAddedTableEnabled;
+ private final boolean releaseSnapshotMetadataEnabled;
private final boolean closeIdleReaders;
private final Properties jdbcProperties;
private final Duration heartbeatInterval;
@@ -136,6 +137,7 @@ public class MySqlTableSource implements ScanTableSource,
SupportsReadingMetadat
double distributionFactorLower,
StartupOptions startupOptions,
boolean scanNewlyAddedTableEnabled,
+ boolean releaseSnapshotMetadataEnabled,
boolean closeIdleReaders,
Properties jdbcProperties,
Duration heartbeatInterval,
@@ -166,6 +168,7 @@ public class MySqlTableSource implements ScanTableSource,
SupportsReadingMetadat
this.distributionFactorLower = distributionFactorLower;
this.startupOptions = startupOptions;
this.scanNewlyAddedTableEnabled = scanNewlyAddedTableEnabled;
+ this.releaseSnapshotMetadataEnabled = releaseSnapshotMetadataEnabled;
this.closeIdleReaders = closeIdleReaders;
this.jdbcProperties = jdbcProperties;
this.parseOnlineSchemaChanges = parseOnlineSchemaChanges;
@@ -233,6 +236,7 @@ public class MySqlTableSource implements ScanTableSource,
SupportsReadingMetadat
.startupOptions(startupOptions)
.deserializer(deserializer)
.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
+
.releaseSnapshotMetadataEnabled(releaseSnapshotMetadataEnabled)
.closeIdleReaders(closeIdleReaders)
.jdbcProperties(jdbcProperties)
.heartbeatInterval(heartbeatInterval)
@@ -322,6 +326,7 @@ public class MySqlTableSource implements ScanTableSource,
SupportsReadingMetadat
distributionFactorLower,
startupOptions,
scanNewlyAddedTableEnabled,
+ releaseSnapshotMetadataEnabled,
closeIdleReaders,
jdbcProperties,
heartbeatInterval,
@@ -353,6 +358,7 @@ public class MySqlTableSource implements ScanTableSource,
SupportsReadingMetadat
&& distributionFactorUpper == that.distributionFactorUpper
&& distributionFactorLower == that.distributionFactorLower
&& scanNewlyAddedTableEnabled ==
that.scanNewlyAddedTableEnabled
+ && releaseSnapshotMetadataEnabled ==
that.releaseSnapshotMetadataEnabled
&& closeIdleReaders == that.closeIdleReaders
&& Objects.equals(physicalSchema, that.physicalSchema)
&& Objects.equals(hostname, that.hostname)
@@ -405,6 +411,7 @@ public class MySqlTableSource implements ScanTableSource,
SupportsReadingMetadat
producedDataType,
metadataKeys,
scanNewlyAddedTableEnabled,
+ releaseSnapshotMetadataEnabled,
closeIdleReaders,
jdbcProperties,
heartbeatInterval,
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java
index 5ea430d94e..e3fd290ff1 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java
@@ -90,6 +90,8 @@ public class MySqlTableSourceFactory implements
DynamicTableSourceFactory {
config.get(MySqlSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
boolean scanNewlyAddedTableEnabled =
config.get(MySqlSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ boolean releaseSnapshotMetadataEnabled =
+
config.get(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
Duration heartbeatInterval =
config.get(MySqlSourceOptions.HEARTBEAT_INTERVAL);
String chunkKeyColumn =
config.getOptional(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN)
@@ -148,6 +150,7 @@ public class MySqlTableSourceFactory implements
DynamicTableSourceFactory {
distributionFactorLower,
startupOptions,
scanNewlyAddedTableEnabled,
+ releaseSnapshotMetadataEnabled,
closeIdleReaders,
JdbcUrlUtils.getJdbcProperties(context.getCatalogTable().getOptions()),
heartbeatInterval,
@@ -198,6 +201,7 @@ public class MySqlTableSourceFactory implements
DynamicTableSourceFactory {
options.add(MySqlSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
options.add(MySqlSourceOptions.CONNECT_MAX_RETRIES);
options.add(MySqlSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED);
+
options.add(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
options.add(MySqlSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
options.add(MySqlSourceOptions.HEARTBEAT_INTERVAL);
options.add(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java
index 62f51e4951..aa29e49f1e 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java
@@ -293,6 +293,45 @@ class MySqlSourceITCase extends MySqlSourceTestBase {
assignEndingFirst));
}
+ @ParameterizedTest
+ @MethodSource("parameters")
+ void testTaskManagerFailoverInBinlogPhaseWithMetadataRelease(
+ String tableName, String chunkColumnName, String
assignEndingFirst) throws Exception {
+ // Releasing the snapshot metadata after entering the binlog phase
must not lose or
+ // duplicate binlog data across a reader (TaskManager) failover. The
released metadata is
+ // checkpoint-covered, so the binlog split is recovered from the
reader's own state rather
+ // than re-created from the emptied assigner.
+ Map<String, String> options = new HashMap<>();
+ options.put("scan.incremental.snapshot.unbounded-chunk-first.enabled",
assignEndingFirst);
+ options.put("scan.incremental.snapshot.metadata.release.enabled",
"true");
+ testMySqlParallelSource(
+ FailoverType.TM,
+ FailoverPhase.BINLOG,
+ new String[] {tableName, "customers_1"},
+ tableName,
+ chunkColumnName,
+ options);
+ }
+
+ @ParameterizedTest
+ @MethodSource("parameters")
+ void testJobManagerFailoverInBinlogPhaseWithMetadataRelease(
+ String tableName, String chunkColumnName, String
assignEndingFirst) throws Exception {
+ // Same as above but across a JobManager (coordinator) failover, which
restores the
+ // enumerator from a released "light" checkpoint and must not
re-discover tables or rebuild
+ // the binlog split from empty metadata.
+ Map<String, String> options = new HashMap<>();
+ options.put("scan.incremental.snapshot.unbounded-chunk-first.enabled",
assignEndingFirst);
+ options.put("scan.incremental.snapshot.metadata.release.enabled",
"true");
+ testMySqlParallelSource(
+ FailoverType.JM,
+ FailoverPhase.BINLOG,
+ new String[] {tableName, "customers_1"},
+ tableName,
+ chunkColumnName,
+ options);
+ }
+
@ParameterizedTest
@MethodSource("parameters")
void testTaskManagerFailoverFromLatestOffset(
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotMetadataReleaseTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotMetadataReleaseTest.java
new file mode 100644
index 0000000000..475e1520a1
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotMetadataReleaseTest.java
@@ -0,0 +1,625 @@
+/*
+ * 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.mysql.source.assigners;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.cdc.connectors.mysql.MySqlTestUtils;
+import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
+import
org.apache.flink.cdc.connectors.mysql.source.assigners.state.ChunkSplitterState;
+import
org.apache.flink.cdc.connectors.mysql.source.assigners.state.HybridPendingSplitsState;
+import
org.apache.flink.cdc.connectors.mysql.source.assigners.state.PendingSplitsStateSerializer;
+import
org.apache.flink.cdc.connectors.mysql.source.assigners.state.SnapshotPendingSplitsState;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
+import
org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import
org.apache.flink.cdc.connectors.mysql.source.enumerator.MySqlSourceEnumerator;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitAssignedEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaAssembledEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaRequestEvent;
+import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
+import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit;
+import
org.apache.flink.cdc.connectors.mysql.source.split.MySqlSchemalessSnapshotSplit;
+import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplit;
+import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplitSerializer;
+import
org.apache.flink.cdc.connectors.mysql.source.utils.MockMySqlSplitEnumeratorEnumeratorContext;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
+
+import io.debezium.relational.TableId;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.Test;
+
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static
org.apache.flink.cdc.connectors.mysql.testutils.MetricsUtils.getMySqlSplitEnumeratorContext;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for releasing the heavyweight snapshot split metadata in {@link
MySqlHybridSplitAssigner}
+ * after the source has safely entered the binlog phase.
+ *
+ * <p>These tests restore the assigner from a finished snapshot checkpoint, so
they exercise the
+ * whole release protocol purely in memory without needing a running MySQL
instance.
+ */
+class MySqlSnapshotMetadataReleaseTest {
+
+ private static final String DB = "release_test_db";
+ private static final String TABLE = "customers";
+ private static final int NUM_FINISHED_SPLITS = 5;
+
+ @Test
+ void testReleaseAfterMetaAssembledAndCheckpointComplete() {
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+
+ // enter the binlog phase
+
assertThat(assigner.getNext()).isPresent().get().isInstanceOf(MySqlBinlogSplit.class);
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+ assertAssignedSplitsSize(assigner, 1, NUM_FINISHED_SPLITS);
+
+ // reader reports it has assembled the complete binlog split metadata
+
assigner.onBinlogSplitMetaAssembled(assigner.getBinlogAssignmentGeneration());
+
+ // snapshotState schedules the release but does not clear yet
+ assigner.snapshotState(2L);
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+ assertAssignedSplitsSize(assigner, 3, NUM_FINISHED_SPLITS);
+
+ // the covering checkpoint completes -> release happens now
+ assigner.notifyCheckpointComplete(2L);
+ assertThat(assigner.isSnapshotMetaReleased()).isTrue();
+
+ HybridPendingSplitsState state = (HybridPendingSplitsState)
assigner.snapshotState(4L);
+
assertThat(state.getSnapshotPendingSplits().getAssignedSplits()).isEmpty();
+
assertThat(state.getSnapshotPendingSplits().getSplitFinishedOffsets()).isEmpty();
+
assertThat(state.getSnapshotPendingSplits().getTableSchemas()).isEmpty();
+ // bookkeeping that must be kept so a restore does not re-discover
tables
+
assertThat(state.getSnapshotPendingSplits().getAlreadyProcessedTables()).isNotEmpty();
+ assertThat(state.isBinlogSplitAssigned()).isTrue();
+
+ assigner.close();
+ }
+
+ @Test
+ void testNoReleaseWithoutMetaAssembledAck() {
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+ assigner.getNext();
+
+ // no onBinlogSplitMetaAssembled() call -> never release
+ assigner.snapshotState(1L);
+ assigner.notifyCheckpointComplete(1L);
+
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+ assertAssignedSplitsSize(assigner, 2, NUM_FINISHED_SPLITS);
+ assigner.close();
+ }
+
+ @Test
+ void testNoReleaseBeforeCheckpointComplete() {
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+ assigner.getNext();
+
assigner.onBinlogSplitMetaAssembled(assigner.getBinlogAssignmentGeneration());
+
+ // schedule release at checkpoint 5, but complete an earlier
checkpoint only
+ assigner.snapshotState(5L);
+ assigner.notifyCheckpointComplete(4L);
+
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+ assertAssignedSplitsSize(assigner, 6, NUM_FINISHED_SPLITS);
+ assigner.close();
+ }
+
+ @Test
+ void testFailFastWhenBothOptionsEnabled() {
+ // Enabling both metadata release and scan.newly-added-table is
contradictory: the release
+ // would drop the metadata the newly-added-table flow needs. Reject it
at config build time
+ // rather than silently disabling the release.
+ assertThatThrownBy(() -> buildConfig(true, true))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("cannot both be enabled");
+ }
+
+ @Test
+ void testNoReleaseWhenReleaseMetadataDisabled() {
+ // The release optimization is opt-in. With it disabled (the default)
the metadata is
+ // retained exactly as before this change, so a job can still enable
scan.newly-added-table
+ // later. This guards against the regression raised in review.
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false, false);
+ assigner.getNext();
+
assigner.onBinlogSplitMetaAssembled(assigner.getBinlogAssignmentGeneration());
+
+ assigner.snapshotState(1L);
+ assigner.notifyCheckpointComplete(1L);
+
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+ assertAssignedSplitsSize(assigner, 2, NUM_FINISHED_SPLITS);
+ assigner.close();
+ }
+
+ @Test
+ void testAddBackResetsScheduledReleaseAndKeepsMetadata() {
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+ MySqlSplit binlogSplit = assigner.getNext().get();
+
assigner.onBinlogSplitMetaAssembled(assigner.getBinlogAssignmentGeneration());
+ assigner.snapshotState(1L); // schedules release at checkpoint 1
+
+ // the binlog split is added back (reader failover before the release
checkpoint completes)
+ assigner.addSplits(Collections.singletonList(binlogSplit));
+
+ // completing checkpoint 1 must NOT release, because the schedule was
reset on add-back
+ assigner.notifyCheckpointComplete(1L);
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+ assertAssignedSplitsSize(assigner, 2, NUM_FINISHED_SPLITS);
+
+ // and the binlog split can be recreated with the full metadata intact
+ Optional<MySqlSplit> recreated = assigner.getNext();
+
assertThat(recreated).isPresent().get().isInstanceOf(MySqlBinlogSplit.class);
+ assertThat(recreated.get().asBinlogSplit().getTotalFinishedSplitSize())
+ .isEqualTo(NUM_FINISHED_SPLITS);
+ assigner.close();
+ }
+
+ @Test
+ void testStaleAssembledEventFromOldGenerationDoesNotRelease() {
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+ MySqlSplit binlogSplit = assigner.getNext().get();
+
+ // the first reader assembles under generation 0 and reports it
+ long staleGeneration = assigner.getBinlogAssignmentGeneration();
+ assigner.onBinlogSplitMetaAssembled(staleGeneration);
+ assigner.snapshotState(1L); // schedules release at checkpoint 1
+
+ // the reader fails: the binlog split is added back, which bumps the
generation
+ assigner.addSplits(Collections.singletonList(binlogSplit));
+
assertThat(assigner.getBinlogAssignmentGeneration()).isGreaterThan(staleGeneration);
+
+ // the binlog split is re-created for a fresh reader that has not
assembled yet
+
assertThat(assigner.getNext()).isPresent().get().isInstanceOf(MySqlBinlogSplit.class);
+
+ // a stale assembled event from the failed attempt (old generation)
arrives late. It must be
+ // ignored, otherwise the metadata would be released while the new
reader still needs it.
+ assigner.onBinlogSplitMetaAssembled(staleGeneration);
+ assigner.snapshotState(2L);
+ assigner.notifyCheckpointComplete(2L);
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+ assertAssignedSplitsSize(assigner, 3, NUM_FINISHED_SPLITS);
+
+ // once the freshly-assigned reader reports under the current
generation, release proceeds
+
assigner.onBinlogSplitMetaAssembled(assigner.getBinlogAssignmentGeneration());
+ assigner.snapshotState(4L);
+ assigner.notifyCheckpointComplete(4L);
+ assertThat(assigner.isSnapshotMetaReleased()).isTrue();
+
+ assigner.close();
+ }
+
+ @Test
+ void testInlineSplitStillReleasesAfterAddBackBumpedGeneration() {
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+ MySqlSplit binlogSplit = assigner.getNext().get();
+
+ // a binlog split failover bumps the generation past the initial 0
+ assigner.addSplits(Collections.singletonList(binlogSplit));
+ assertThat(assigner.getBinlogAssignmentGeneration()).isGreaterThan(0L);
+
assertThat(assigner.getNext()).isPresent().get().isInstanceOf(MySqlBinlogSplit.class);
+
+ // an inline reader never requests meta groups, so it never learns the
bumped generation
+ // and reports COMPLETE_WITHOUT_META_GENERATION. Release must still
happen, otherwise the
+ // optimization silently stops working for small tables after any
failover.
+ assigner.onBinlogSplitMetaAssembled(
+
BinlogSplitMetaAssembledEvent.COMPLETE_WITHOUT_META_GENERATION);
+ assigner.snapshotState(2L);
+ assigner.notifyCheckpointComplete(2L);
+ assertThat(assigner.isSnapshotMetaReleased()).isTrue();
+
+ assigner.close();
+ }
+
+ @Test
+ void testEmptyAddBackFromBinlogReaderInvalidatesStaleAssembledEvent() {
+ // A reader holding the binlog split can reset without its split being
handed back: when the
+ // assignment is already checkpoint-covered, Flink restores the split
from the reader's own
+ // state and calls addSplitsBack with an empty list. The generation
must still be bumped and
+ // any scheduled release cancelled, otherwise a delayed assembled
event from the failed
+ // attempt would arm an unsafe release while the restored reader still
needs the metadata.
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+ assigner.getNext(); // enter the binlog phase; the binlog split is
assigned
+
+ MySqlSourceEnumerator enumerator =
+ new MySqlSourceEnumerator(
+ getMySqlSplitEnumeratorContext(),
+ buildConfig(false, true),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+
+ // subtask 0 reports it holds the binlog split, then assembles it and
schedules a release
+ enumerator.handleSourceEvent(0, new BinlogSplitAssignedEvent());
+ long staleGeneration = assigner.getBinlogAssignmentGeneration();
+ assigner.onBinlogSplitMetaAssembled(staleGeneration);
+ assigner.snapshotState(1L); // schedules release at checkpoint 1
+
+ // subtask 0 resets, but its assignment was checkpoint-covered, so
Flink hands nothing back
+ enumerator.addSplitsBack(Collections.emptyList(), 0);
+
+ // completing checkpoint 1 must NOT release: the reset cancelled the
schedule
+ assigner.notifyCheckpointComplete(1L);
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+
assertThat(assigner.getBinlogAssignmentGeneration()).isGreaterThan(staleGeneration);
+
+ // a delayed assembled event from the failed attempt (old generation)
is now ignored
+ assigner.onBinlogSplitMetaAssembled(staleGeneration);
+ assigner.snapshotState(2L);
+ assigner.notifyCheckpointComplete(2L);
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+
+ assigner.close();
+ }
+
+ @Test
+ void testStaleBinlogSplitAssignedEventDoesNotCorruptHolder() {
+ // A delayed BinlogSplitAssignedEvent from a failed attempt must not
overwrite the live
+ // binlog holder. If it did, the empty-add-back reset detection (which
keys off
+ // binlogSplitTaskId) would be defeated and a stale assembled event
could arm an unsafe
+ // release. Here the real holder is subtask 1; a stale assigned event
from subtask 0 arrives
+ // and must be ignored, so the later empty add-back from subtask 1
still cancels the
+ // release.
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+ assigner.getNext(); // enter the binlog phase
+
+ MySqlSourceEnumerator enumerator =
+ new MySqlSourceEnumerator(
+ getMySqlSplitEnumeratorContext(),
+ buildConfig(false, true),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+
+ // subtask 1 is the live binlog holder; a stale assigned event from
subtask 0 must be
+ // ignored
+ enumerator.handleSourceEvent(1, new BinlogSplitAssignedEvent());
+ enumerator.handleSourceEvent(0, new BinlogSplitAssignedEvent());
+
+ long staleGeneration = assigner.getBinlogAssignmentGeneration();
+ assigner.onBinlogSplitMetaAssembled(staleGeneration);
+ assigner.snapshotState(1L); // schedules release at checkpoint 1
+
+ // the real holder (subtask 1) resets with an empty add-back; because
binlogSplitTaskId was
+ // not corrupted to 0, the reset is still detected and the scheduled
release is cancelled
+ enumerator.addSplitsBack(Collections.emptyList(), 1);
+ assigner.notifyCheckpointComplete(1L);
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+
assertThat(assigner.getBinlogAssignmentGeneration()).isGreaterThan(staleGeneration);
+
+ assigner.close();
+ }
+
+ @Test
+ void testRestoreFromReleasedLightState() {
+ MySqlHybridSplitAssigner assigner = buildReleasedLightAssigner();
+
+ // No binlog split is re-created and no tables are re-discovered after
a light restore.
+ assertThat(assigner.getNext()).isEmpty();
+ // The released flag is restored from the persisted checkpoint value,
so after restore the
+ // assigner knows the metadata was already released.
+ assertThat(assigner.isSnapshotMetaReleased()).isTrue();
+
+ // The state stays light (empty heavy maps, binlog split still
assigned).
+ HybridPendingSplitsState state = (HybridPendingSplitsState)
assigner.snapshotState(10L);
+
assertThat(state.getSnapshotPendingSplits().getAssignedSplits()).isEmpty();
+
assertThat(state.getSnapshotPendingSplits().getSplitFinishedOffsets()).isEmpty();
+
assertThat(state.getSnapshotPendingSplits().getAlreadyProcessedTables()).isNotEmpty();
+ assertThat(state.isBinlogSplitAssigned()).isTrue();
+
+ // Already released after restore; re-sending the assembled event and
completing a
+ // checkpoint is a no-op, and the assigner stays in the released state.
+
assigner.onBinlogSplitMetaAssembled(assigner.getBinlogAssignmentGeneration());
+ assigner.snapshotState(11L);
+ assigner.notifyCheckpointComplete(11L);
+ assertThat(assigner.isSnapshotMetaReleased()).isTrue();
+
+ assigner.close();
+ }
+
+ @Test
+ void testReleasedFlagNotInferredWhenSnapshotSplitsRemain() {
+ // The released flag is persisted (PendingSplitsStateSerializer v6),
not inferred from state
+ // shape. A finished parallelism-1 state that never released (flag
defaults false) must
+ // restore as not-released even with non-empty remainingSplits and
empty assignedSplits.
+ // This guards against anyone re-introducing shape-based inference,
which previously
+ // false-positived here and crash-looped createBinlogSplit,
feature-off jobs included.
+ List<TableId> alreadyProcessedTables = Lists.newArrayList(new
TableId(null, DB, TABLE));
+ RowType splitKeyType =
+ (RowType) DataTypes.ROW(DataTypes.FIELD("id",
DataTypes.BIGINT())).getLogicalType();
+ List<MySqlSchemalessSnapshotSplit> remainingSplits = new ArrayList<>();
+ remainingSplits.add(
+ new MySqlSchemalessSnapshotSplit(
+ new TableId(null, DB, TABLE),
+ DB + "." + TABLE + ":0",
+ splitKeyType,
+ null,
+ new Object[] {2},
+ BinlogOffset.ofBinlogFilePosition("mysql-bin.00001",
1)));
+ SnapshotPendingSplitsState snapshotState =
+ new SnapshotPendingSplitsState(
+ alreadyProcessedTables,
+ remainingSplits,
+ new LinkedHashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ AssignerStatus.INITIAL_ASSIGNING_FINISHED,
+ new ArrayList<>(),
+ false,
+ true,
+ ChunkSplitterState.NO_SPLITTING_TABLE_STATE);
+ HybridPendingSplitsState checkpoint = new
HybridPendingSplitsState(snapshotState, false);
+ MySqlHybridSplitAssigner assigner =
+ new MySqlHybridSplitAssigner(
+ buildConfig(false, true), 1, checkpoint,
getMySqlSplitEnumeratorContext());
+
+ assertThat(assigner.isSnapshotMetaReleased()).isFalse();
+ assigner.close();
+ }
+
+ @Test
+ void testFailFastWhenNewlyAddedEnabledOnReleasedRestore() {
+ // A job that released its metadata cannot later enable
scan.newly-added-table, because the
+ // metadata that flow needs is gone. Restoring such a state with the
flag on must fail fast
+ // with a clear error rather than silently corrupting the binlog split.
+ assertThatThrownBy(() -> buildReleasedLightAssigner(true, false))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("scan.newly-added-table.enabled cannot
be turned on");
+ }
+
+ @Test
+ void testFailFastWhenReleaseDisabledOnReleasedRestore() {
+ // A job that already released its snapshot metadata cannot be
restarted with the release
+ // option turned off. The metadata is gone from state, and writing the
lighter v5 format
+ // would drop the released marker, so restoring such a state with the
option off must fail
+ // fast rather than silently corrupt the binlog split.
+ assertThatThrownBy(() -> buildReleasedLightAssigner(false, false))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining(
+ "scan.incremental.snapshot.metadata.release.enabled
cannot be turned off");
+ }
+
+ @Test
+ void testEnumeratorSerializerWritesV6OnlyWhenReleaseEnabled() {
+ // The real wiring: MySqlSource.getEnumeratorCheckpointSerializer()
picks the serializer
+ // version from the option, so a release-enabled job writes v6 and a
default-off job keeps
+ // writing v5.
+
assertThat(buildSource(true).getEnumeratorCheckpointSerializer().getVersion()).isEqualTo(6);
+
assertThat(buildSource(false).getEnumeratorCheckpointSerializer().getVersion())
+ .isEqualTo(5);
+ }
+
+ @Test
+ void testReleasedStateSerializedThenRestoredWithOptionOffFailsFast()
throws Exception {
+ // End to end: a released state serialized by the release-on
serializer (v6, flag true),
+ // then
+ // restored with the option off, must deserialize the flag as true and
fail fast rather than
+ // silently dropping the marker by rewriting the v5 format.
+ HybridPendingSplitsState released = buildReleasedHybridCheckpoint();
+ PendingSplitsStateSerializer serializer =
+ new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE, true);
+ byte[] bytes = serializer.serialize(released);
+ HybridPendingSplitsState restored =
+ (HybridPendingSplitsState)
serializer.deserialize(serializer.getVersion(), bytes);
+
assertThat(restored.getSnapshotPendingSplits().isSnapshotMetaReleased()).isTrue();
+
+ assertThatThrownBy(
+ () ->
+ new MySqlHybridSplitAssigner(
+ buildConfig(false, false),
+ 4,
+ restored,
+ getMySqlSplitEnumeratorContext()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining(
+ "scan.incremental.snapshot.metadata.release.enabled
cannot be turned off");
+ }
+
+ @Test
+ void testStaleBinlogMetaRequestAfterLightRestoreIsIgnored() {
+ // After restoring from a released light checkpoint the released flag
is restored from the
+ // persisted checkpoint value, so the enumerator still ignores a stray
meta request instead
+ // of rebuilding and throwing.
+ MySqlHybridSplitAssigner assigner = buildReleasedLightAssigner();
+ assertThat(assigner.isSnapshotMetaReleased()).isTrue();
+
+ MySqlSourceEnumerator enumerator =
+ new MySqlSourceEnumerator(
+ getMySqlSplitEnumeratorContext(),
+ buildConfig(false, true),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+
+ assertThatCode(
+ () ->
+ enumerator.handleSourceEvent(
+ 0,
+ new BinlogSplitMetaRequestEvent(
+ "binlog-split", 0,
NUM_FINISHED_SPLITS)))
+ .doesNotThrowAnyException();
+
+ assigner.close();
+ }
+
+ @Test
+ void testStaleBinlogMetaRequestAfterReleaseIsIgnored() {
+ // A stale BinlogSplitMetaRequestEvent from a failed reader attempt
can arrive after the
+ // snapshot metadata has been released. The enumerator must ignore it
instead of rebuilding
+ // from the emptied assigner, which would throw FlinkRuntimeException
and fail the job.
+ MySqlHybridSplitAssigner assigner =
buildFinishedSnapshotAssigner(false);
+ assigner.getNext();
+
assigner.onBinlogSplitMetaAssembled(assigner.getBinlogAssignmentGeneration());
+ assigner.snapshotState(1L);
+ assigner.notifyCheckpointComplete(1L);
+ assertThat(assigner.isSnapshotMetaReleased()).isTrue();
+
+ MySqlSourceEnumerator enumerator =
+ new MySqlSourceEnumerator(
+ getMySqlSplitEnumeratorContext(),
+ buildConfig(false, true),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+
+ assertThatCode(
+ () ->
+ enumerator.handleSourceEvent(
+ 0,
+ new BinlogSplitMetaRequestEvent(
+ "binlog-split", 0,
NUM_FINISHED_SPLITS)))
+ .doesNotThrowAnyException();
+
+ assigner.close();
+ }
+
+ //
------------------------------------------------------------------------------------------
+
+ private void assertAssignedSplitsSize(
+ MySqlHybridSplitAssigner assigner, long checkpointId, int
expectedSize) {
+ HybridPendingSplitsState state =
+ (HybridPendingSplitsState)
assigner.snapshotState(checkpointId);
+
assertThat(state.getSnapshotPendingSplits().getAssignedSplits()).hasSize(expectedSize);
+ assertThat(state.getSnapshotPendingSplits().getSplitFinishedOffsets())
+ .hasSize(expectedSize);
+ }
+
+ private MySqlSourceConfig buildConfig(
+ boolean scanNewlyAddedTableEnabled, boolean
releaseSnapshotMetadataEnabled) {
+ return new MySqlSourceConfigFactory()
+ .startupOptions(StartupOptions.initial())
+ .databaseList(DB)
+ .tableList(DB + "." + TABLE)
+ .hostname("localhost")
+ .port(3306)
+ .username("root")
+ .password("")
+ .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
+ .releaseSnapshotMetadataEnabled(releaseSnapshotMetadataEnabled)
+ .serverTimeZone(ZoneId.of("UTC").toString())
+ .createConfig(0);
+ }
+
+ private MySqlHybridSplitAssigner buildFinishedSnapshotAssigner(
+ boolean scanNewlyAddedTableEnabled) {
+ return buildFinishedSnapshotAssigner(scanNewlyAddedTableEnabled, true);
+ }
+
+ private MySqlHybridSplitAssigner buildFinishedSnapshotAssigner(
+ boolean scanNewlyAddedTableEnabled, boolean
releaseSnapshotMetadataEnabled) {
+ MySqlSourceConfig config =
+ buildConfig(scanNewlyAddedTableEnabled,
releaseSnapshotMetadataEnabled);
+
+ TableId tableId = new TableId(null, DB, TABLE);
+ RowType splitKeyType =
+ (RowType) DataTypes.ROW(DataTypes.FIELD("id",
DataTypes.BIGINT())).getLogicalType();
+
+ List<TableId> alreadyProcessedTables = Lists.newArrayList(tableId);
+ LinkedHashMap<String, MySqlSchemalessSnapshotSplit> assignedSplits =
new LinkedHashMap<>();
+ Map<String, BinlogOffset> splitFinishedOffsets = new HashMap<>();
+ for (int i = 0; i < NUM_FINISHED_SPLITS; i++) {
+ String splitId = DB + "." + TABLE + ":" + i;
+ Object[] splitStart = i == 0 ? null : new Object[] {i * 2};
+ Object[] splitEnd = new Object[] {i * 2 + 2};
+ BinlogOffset highWatermark =
+ BinlogOffset.ofBinlogFilePosition("mysql-bin.00001", i +
1);
+ assignedSplits.put(
+ splitId,
+ new MySqlSchemalessSnapshotSplit(
+ tableId, splitId, splitKeyType, splitStart,
splitEnd, highWatermark));
+ splitFinishedOffsets.put(splitId, highWatermark);
+ }
+
+ SnapshotPendingSplitsState snapshotState =
+ new SnapshotPendingSplitsState(
+ alreadyProcessedTables,
+ new ArrayList<>(),
+ assignedSplits,
+ new HashMap<>(),
+ splitFinishedOffsets,
+ AssignerStatus.INITIAL_ASSIGNING_FINISHED,
+ new ArrayList<>(),
+ false,
+ true,
+ ChunkSplitterState.NO_SPLITTING_TABLE_STATE);
+ HybridPendingSplitsState checkpoint = new
HybridPendingSplitsState(snapshotState, false);
+ MockMySqlSplitEnumeratorEnumeratorContext enumeratorContext =
+ getMySqlSplitEnumeratorContext();
+ return new MySqlHybridSplitAssigner(config, 4, checkpoint,
enumeratorContext);
+ }
+
+ /**
+ * Builds an assigner restored from a "light" checkpoint: the metadata was
already released, so
+ * the maps are empty, the binlog split is assigned, and only
alreadyProcessedTables and the
+ * finished status are retained.
+ */
+ private MySqlHybridSplitAssigner buildReleasedLightAssigner() {
+ return buildReleasedLightAssigner(false, true);
+ }
+
+ private MySqlHybridSplitAssigner buildReleasedLightAssigner(
+ boolean scanNewlyAddedTableEnabled, boolean
releaseSnapshotMetadataEnabled) {
+ return new MySqlHybridSplitAssigner(
+ buildConfig(scanNewlyAddedTableEnabled,
releaseSnapshotMetadataEnabled),
+ 4,
+ buildReleasedHybridCheckpoint(),
+ getMySqlSplitEnumeratorContext());
+ }
+
+ private HybridPendingSplitsState buildReleasedHybridCheckpoint() {
+ List<TableId> alreadyProcessedTables = Lists.newArrayList(new
TableId(null, DB, TABLE));
+ SnapshotPendingSplitsState snapshotState =
+ new SnapshotPendingSplitsState(
+ alreadyProcessedTables,
+ new ArrayList<>(),
+ new LinkedHashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ AssignerStatus.INITIAL_ASSIGNING_FINISHED,
+ new ArrayList<>(),
+ false,
+ true,
+ ChunkSplitterState.NO_SPLITTING_TABLE_STATE,
+ true);
+ return new HybridPendingSplitsState(snapshotState, true);
+ }
+
+ private MySqlSource<SourceRecord> buildSource(boolean
releaseSnapshotMetadataEnabled) {
+ return MySqlSource.<SourceRecord>builder()
+ .hostname("localhost")
+ .port(3306)
+ .databaseList(DB)
+ .tableList(DB + "." + TABLE)
+ .username("user")
+ .password("password")
+ .serverTimeZone("UTC")
+ .deserializer(new MySqlTestUtils.ForwardDeserializeSchema())
+ .releaseSnapshotMetadataEnabled(releaseSnapshotMetadataEnabled)
+ .build();
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializerTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializerTest.java
index 60abb56ff1..7193962b4e 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializerTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializerTest.java
@@ -30,6 +30,7 @@ import io.debezium.relational.TableEditor;
import io.debezium.relational.TableId;
import io.debezium.relational.history.TableChanges;
import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -95,7 +96,7 @@ class PendingSplitsStateSerializerTest {
@MethodSource("params")
void testRepeatedSerializationCache(PendingSplitsState state) throws
Exception {
final PendingSplitsStateSerializer serializer =
- new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE);
+ new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE, true);
final byte[] ser1 = serializer.serialize(state);
final byte[] ser2 = serializer.serialize(state);
@@ -107,7 +108,7 @@ class PendingSplitsStateSerializerTest {
@MethodSource("params")
void testOutputIsFinallyCleared(PendingSplitsState state) throws Exception
{
final PendingSplitsStateSerializer serializer =
- new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE);
+ new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE, true);
final byte[] ser1 = serializer.serialize(state);
state.serializedFormCache = null;
@@ -121,10 +122,100 @@ class PendingSplitsStateSerializerTest {
Assertions.assertThat(ser1).isEqualTo(ser2);
}
+ @Test
+ void testSerializeAndDeserializeReleasedHybridState() throws Exception {
+ // The "light" state produced after releasing the snapshot metadata
must round-trip, and
+ // the persisted released flag (PendingSplitsStateSerializer v6) must
survive.
+ HybridPendingSplitsState released =
getTestReleasedHybridPendingSplitsState();
+ PendingSplitsState roundTripped =
serializeAndDeserializeSourceEnumState(released);
+ Assertions.assertThat(roundTripped).isEqualTo(released);
+
+ SnapshotPendingSplitsState snapshot =
+ ((HybridPendingSplitsState)
roundTripped).getSnapshotPendingSplits();
+ Assertions.assertThat(snapshot.getAssignedSplits()).isEmpty();
+ Assertions.assertThat(snapshot.getSplitFinishedOffsets()).isEmpty();
+ Assertions.assertThat(snapshot.getTableSchemas()).isEmpty();
+
Assertions.assertThat(snapshot.getAlreadyProcessedTables()).isNotEmpty();
+ Assertions.assertThat(snapshot.isSnapshotMetaReleased()).isTrue();
+ Assertions.assertThat(((HybridPendingSplitsState)
roundTripped).isBinlogSplitAssigned())
+ .isTrue();
+ }
+
+ @Test
+ void testDeserializeV5SnapshotStateDefaultsReleasedFlagToFalse() throws
Exception {
+ // A v5 checkpoint written before FLINK-39775 has no released-flag
byte. For a snapshot
+ // state that byte is the last one written, so a v6 stream without its
final byte is exactly
+ // the v5 wire format. Deserializing it under the current version must
default
+ // snapshotMetaReleased to false and otherwise round-trip unchanged.
+ SnapshotPendingSplitsState state =
getTestSnapshotPendingSplitsState(false);
+ PendingSplitsStateSerializer serializer =
+ new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE, true);
+ byte[] v6 = serializer.serialize(state);
+ byte[] v5 = Arrays.copyOf(v6, v6.length - 1);
+
+ PendingSplitsState restored = serializer.deserialize(5, v5);
+
+
Assertions.assertThat(restored).isInstanceOf(SnapshotPendingSplitsState.class);
+ Assertions.assertThat(((SnapshotPendingSplitsState)
restored).isSnapshotMetaReleased())
+ .isFalse();
+ Assertions.assertThat(restored).isEqualTo(state);
+ }
+
+ @Test
+ void testReleaseDisabledJobKeepsWritingV5Format() throws Exception {
+ // A job that leaves
scan.incremental.snapshot.metadata.release.enabled off must keep
+ // writing
+ // the v5 format (no released flag), so it can still be restored by an
older connector
+ // build.
+ SnapshotPendingSplitsState state =
getTestSnapshotPendingSplitsState(false);
+ PendingSplitsStateSerializer v5Serializer =
+ new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE, false);
+ PendingSplitsStateSerializer v6Serializer =
+ new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE, true);
+
+ Assertions.assertThat(v5Serializer.getVersion()).isEqualTo(5);
+ Assertions.assertThat(v6Serializer.getVersion()).isEqualTo(6);
+
+ byte[] v6 = v6Serializer.serialize(state);
+ // Clear the lazily cached serialized form so the v5 serializer
re-serializes the state.
+ state.serializedFormCache = null;
+ byte[] v5 = v5Serializer.serialize(state);
+
+ // For a snapshot state the released flag is the final byte, so the v5
bytes are exactly the
+ // v6 bytes without that trailing byte, i.e. the format an older
connector wrote.
+ Assertions.assertThat(v5).isEqualTo(Arrays.copyOf(v6, v6.length - 1));
+
+ // A default-off job restores its own v5 checkpoint with the flag
defaulting to false.
+ PendingSplitsState restored = v5Serializer.deserialize(5, v5);
+ Assertions.assertThat(restored).isEqualTo(state);
+ Assertions.assertThat(((SnapshotPendingSplitsState)
restored).isSnapshotMetaReleased())
+ .isFalse();
+ }
+
+ @Test
+ void testReleaseDisabledSerializerRoundTripsHybridStateAtV5() throws
Exception {
+ // The option-off serializer round-trips a hybrid state at v5 (the
released flag is written
+ // for neither the snapshot part nor read back), leaving the flag
false.
+ HybridPendingSplitsState state =
getTestHybridPendingSplitsState(false);
+ PendingSplitsStateSerializer v5Serializer =
+ new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE, false);
+
+ byte[] serialized = v5Serializer.serialize(state);
+ PendingSplitsState restored =
+ v5Serializer.deserialize(v5Serializer.getVersion(),
serialized);
+
+ Assertions.assertThat(restored).isEqualTo(state);
+ Assertions.assertThat(
+ ((HybridPendingSplitsState) restored)
+ .getSnapshotPendingSplits()
+ .isSnapshotMetaReleased())
+ .isFalse();
+ }
+
static PendingSplitsState
serializeAndDeserializeSourceEnumState(PendingSplitsState state)
throws Exception {
final PendingSplitsStateSerializer serializer =
- new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE);
+ new
PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE, true);
byte[] serialized = serializer.serialize(state);
return serializer.deserialize(serializer.getVersion(), serialized);
}
@@ -203,6 +294,28 @@ class PendingSplitsStateSerializerTest {
getTestSnapshotPendingSplitsState(checkpointWhenSplitting),
false);
}
+ private static HybridPendingSplitsState
getTestReleasedHybridPendingSplitsState() {
+ // After releasing the snapshot metadata, the hybrid state is just a
normal state
+ // with empty assigned/finished/schema maps and
isBinlogSplitAssigned=true.
+ final List<TableId> alreadyProcessedTables = new ArrayList<>();
+ alreadyProcessedTables.add(tableId0);
+ alreadyProcessedTables.add(tableId1);
+ SnapshotPendingSplitsState snapshotState =
+ new SnapshotPendingSplitsState(
+ alreadyProcessedTables,
+ new ArrayList<>(),
+ new LinkedHashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ AssignerStatus.INITIAL_ASSIGNING_FINISHED,
+ new ArrayList<>(),
+ false,
+ true,
+ ChunkSplitterState.NO_SPLITTING_TABLE_STATE,
+ true);
+ return new HybridPendingSplitsState(snapshotState, true);
+ }
+
private static BinlogPendingSplitsState getTestBinlogPendingSplitsState() {
return new BinlogPendingSplitsState(true);
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumeratorTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumeratorTest.java
new file mode 100644
index 0000000000..7fb2206645
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumeratorTest.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.mysql.source.enumerator;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import
org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlBinlogSplitAssigner;
+import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
+import
org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaAssembledEvent;
+import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.time.ZoneId;
+
+import static
org.apache.flink.cdc.connectors.mysql.testutils.MetricsUtils.getMySqlSplitEnumeratorContext;
+
+/** Tests for {@link MySqlSourceEnumerator}. */
+class MySqlSourceEnumeratorTest {
+
+ @Test
+ void testBinlogSplitMetaAssembledEventIsNoOpForBinlogOnlyAssigner() {
+ // For binlog-only startup modes the assigner is a
MySqlBinlogSplitAssigner, not the hybrid
+ // one. The reader still sends BinlogSplitMetaAssembledEvent, and the
enumerator must treat
+ // it as a safe no-op rather than failing.
+ MySqlSourceConfig config =
+ new MySqlSourceConfigFactory()
+ .startupOptions(StartupOptions.latest())
+ .databaseList("db")
+ .tableList("db.t")
+ .hostname("localhost")
+ .port(3306)
+ .username("root")
+ .password("")
+ .serverTimeZone(ZoneId.of("UTC").toString())
+ .createConfig(0);
+ MySqlSourceEnumerator enumerator =
+ new MySqlSourceEnumerator(
+ getMySqlSplitEnumeratorContext(),
+ config,
+ new MySqlBinlogSplitAssigner(config),
+ Boundedness.CONTINUOUS_UNBOUNDED);
+
+ Assertions.assertThatCode(
+ () ->
+ enumerator.handleSourceEvent(
+ 0, new
BinlogSplitMetaAssembledEvent(0L)))
+ .doesNotThrowAnyException();
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java
index f2fc642669..137df37c57 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java
@@ -28,9 +28,12 @@ import
org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlBinlogSplitAs
import
org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import
org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaAssembledEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.FinishedSnapshotSplitsAckEvent;
import
org.apache.flink.cdc.connectors.mysql.source.metrics.MySqlSourceReaderMetrics;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
+import
org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSnapshotSplit;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplit;
@@ -379,6 +382,106 @@ class MySqlSourceReaderTest extends MySqlSourceTestBase {
restartReader.close();
}
+ @Test
+ void testReaderSendsBinlogSplitMetaAssembledEventForCompleteSplit() throws
Exception {
+ // When a complete binlog split enters reading, the reader must notify
the
+ // enumerator that its finished snapshot split metadata is fully
assembled, so the
+ // coordinator can release the retained snapshot metadata.
+ customerDatabase.createAndInitialize();
+ final MySqlSourceConfig sourceConfig = getConfig(new String[]
{"customers"});
+ MySqlSplit binlogSplit;
+ try (MySqlConnection jdbc =
DebeziumUtils.createMySqlConnection(sourceConfig)) {
+ Map<TableId, TableChanges.TableChange> tableSchemas =
+ TableDiscoveryUtils.discoverSchemaForCapturedTables(
+ new MySqlPartition(
+
sourceConfig.getMySqlConnectorConfig().getLogicalName()),
+ sourceConfig,
+ jdbc);
+ binlogSplit =
+ MySqlBinlogSplit.fillTableSchemas(
+ createBinlogSplit(sourceConfig).asBinlogSplit(),
tableSchemas);
+ }
+ // a binlog split created from the latest startup offset carries no
finished snapshot split
+ // info, so it is already complete on receipt
+
Assertions.assertThat(binlogSplit.asBinlogSplit().isCompletedSplit()).isTrue();
+
+ TestingReaderContext readerContext = new TestingReaderContext();
+ MySqlSourceReader<SourceRecord> reader = createReader(sourceConfig,
readerContext, 1);
+ reader.start();
+ reader.addSplits(Collections.singletonList(binlogSplit));
+
+ // the split was complete inline (no meta groups requested), so the
reader reports the
+ // COMPLETE_WITHOUT_META_GENERATION sentinel rather than a served
generation
+ Assertions.assertThat(readerContext.getSentEvents())
+ .filteredOn(event -> event instanceof
BinlogSplitMetaAssembledEvent)
+ .extracting(
+ event ->
+ ((BinlogSplitMetaAssembledEvent) event)
+ .getBinlogAssignmentGeneration())
+
.containsExactly(BinlogSplitMetaAssembledEvent.COMPLETE_WITHOUT_META_GENERATION);
+ reader.close();
+ }
+
+ @Test
+ void testReaderEchoesBinlogAssignmentGenerationInAssembledEvent() throws
Exception {
+ // A binlog split assembled from divided meta groups must echo the
binlog assignment
+ // generation it was served under, so the coordinator can reject a
stale assembled event
+ // from a failed attempt. Drive the divided-meta path: give the reader
an incomplete binlog
+ // split, answer its meta request with a group stamped with a
generation, and assert the
+ // assembled event the reader sends back carries that same generation.
+ customerDatabase.createAndInitialize();
+ final MySqlSourceConfig sourceConfig = getConfig(new String[]
{"customers"});
+ final TableId tableId =
TableId.parse(customerDatabase.getDatabaseName() + ".customers");
+
+ // one finished split info is still outstanding, so the split is
incomplete on receipt
+ MySqlBinlogSplit incompleteSplit =
+ new MySqlBinlogSplit(
+ "binlog-split",
+ BinlogOffset.ofEarliest(),
+ null,
+ new ArrayList<>(),
+ new HashMap<>(),
+ 1,
+ false);
+ Assertions.assertThat(incompleteSplit.isCompletedSplit()).isFalse();
+
+ TestingReaderContext readerContext = new TestingReaderContext();
+ MySqlSourceReader<SourceRecord> reader = createReader(sourceConfig,
readerContext, 0);
+ reader.start();
+ reader.addSplits(Collections.singletonList(incompleteSplit));
+
+ // the coordinator answers the reader's meta request with the
outstanding group, stamped
+ // with the current binlog assignment generation
+ long generation = 7L;
+ FinishedSnapshotSplitInfo finishedInfo =
+ new FinishedSnapshotSplitInfo(
+ tableId,
+ tableId + ":0",
+ new Object[] {0},
+ new Object[] {100},
+ BinlogOffset.ofBinlogFilePosition("mysql-bin.000001",
4L));
+ reader.handleSourceEvents(
+ new BinlogSplitMetaEvent(
+ incompleteSplit.splitId(),
+ 0,
+ Collections.singletonList(
+
FinishedSnapshotSplitInfo.serialize(finishedInfo)),
+ 1,
+ generation));
+
+ BinlogSplitMetaAssembledEvent assembled =
+ readerContext.getSentEvents().stream()
+ .filter(event -> event instanceof
BinlogSplitMetaAssembledEvent)
+ .map(event -> (BinlogSplitMetaAssembledEvent) event)
+ .reduce((first, second) -> second)
+ .orElseThrow(
+ () ->
+ new AssertionError(
+ "reader did not send a
BinlogSplitMetaAssembledEvent"));
+
Assertions.assertThat(assembled.getBinlogAssignmentGeneration()).isEqualTo(generation);
+ reader.close();
+ }
+
@Test
void testRemoveSplitAccordingToNewFilter() throws Exception {
inventoryDatabase.createAndInitialize();
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java
index 01c2dff84d..33abc21ddb 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java
@@ -122,6 +122,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.initial(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -172,6 +173,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.initial(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
"testCol",
@@ -218,6 +220,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.initial(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -262,6 +265,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.latest(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -322,6 +326,7 @@ class MySqlTableSourceFactoryTest {
CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.defaultValue(),
StartupOptions.initial(),
true,
+ false,
true,
jdbcProperties,
Duration.ofMillis(15213),
@@ -382,6 +387,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.specificOffset(offsetFile, offsetPos),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -424,6 +430,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.initial(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -467,6 +474,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.earliest(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -511,6 +519,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.timestamp(0L),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -553,6 +562,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.latest(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -600,6 +610,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.initial(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -803,6 +814,7 @@ class MySqlTableSourceFactoryTest {
StartupOptions.initial(),
false,
false,
+ false,
new Properties(),
HEARTBEAT_INTERVAL.defaultValue(),
null,
@@ -814,6 +826,19 @@ class MySqlTableSourceFactoryTest {
Assertions.assertThat(actualSource).isEqualTo(expectedSource);
}
+ @Test
+ void testReleaseSnapshotMetadataOption() {
+ // Setting the option must be accepted (it is registered in
optionalOptions, so there is no
+ // "unsupported options" validation error) and must change the
produced source.
+ Map<String, String> withRelease = getAllOptions();
+ withRelease.put("scan.incremental.snapshot.metadata.release.enabled",
"true");
+ DynamicTableSource sourceWithRelease = createTableSource(withRelease);
+
+ Assertions.assertThat(sourceWithRelease)
+ .isInstanceOf(MySqlTableSource.class)
+ .isNotEqualTo(createTableSource(getAllOptions()));
+ }
+
private Map<String, String> getAllOptions() {
Map<String, String> options = new HashMap<>();
options.put("connector", "mysql-cdc");
