You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "spoorthibasu (via GitHub)" <gi...@apache.org> on 2026/09/17 17:42:02 UTC
[PR] [FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase [flink-cdc]
spoorthibasu opened a new pull request, #4550:
URL: https://github.com/apache/flink-cdc/pull/4550
<!--
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.
-->
## What is the purpose of this pull request?
The incremental source framework (flink-cdc-base) keeps the full snapshot-split metadata (assigned splits, finished offsets, table schemas) in the split enumerator for the whole lifetime of the job and re-serializes it on every checkpoint. On a large multi-table job this keeps JobManager memory high and inflates checkpoint size.
FLINK-39775 fixed this for the MySQL connector. This PR generalizes the same mechanism into flink-cdc-base so the JDBC incremental dialects (Postgres, SqlServer, Oracle, Db2) get it, rather than each connector solving it separately.
The release is opt-in behind a new option `scan.incremental.snapshot.metadata.release.enabled` (default `false`), so existing jobs are unaffected. Once the stream split is assigned and the reader reports its metadata assembled, the enumerator schedules the release at the current checkpoint and clears it in `notifyCheckpointComplete`, so the assignment is always covered by a completed checkpoint. An assignment-generation guard rejects a stale assembled report from a reader that failed over. The checkpoint serializer is bumped to version 9 only for jobs that enable the option, so default-off jobs keep writing version 8 and stay restorable by an older build.
## Brief change log
- Add `scan.incremental.snapshot.metadata.release.enabled` (default `false`) to the base source options and config
- Add the release primitive to the snapshot assigner and thread it through the split and hybrid assigners
- Add the reader-to-enumerator assembled event and the enumerator release orchestration (arm in `snapshotState`, execute in `notifyCheckpointComplete`), with the assignment-generation guard
- Persist the released flag; `PendingSplitsStateSerializer` writes version 9 only when the option is enabled, else version 8
- Fail fast on release together with `scan.newly-added-table.enabled`, and on restoring an already-released state with the option off
- Expose the option through the DataStream builders and Flink SQL for Postgres, SqlServer, Oracle, and Db2
- Document the option (EN and ZH) for the four connectors
---
## Verifying this change
This change added tests and can be verified as follows:
- Unit tests in flink-cdc-base for the release protocol, the generation guard, the conditional serializer versioning (version 8 vs 9 by the option), and the fail-fast paths
- Integration tests on real Postgres and SqlServer databases covering the group-fetch release path plus JobManager and TaskManager failover in the stream phase, verifying no snapshot or stream data is lost through the release and restore
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs
The Chinese doc entries mirror the English and are a best-effort translation. Please correct the wording where needed.
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change the checkbox below to `[X]` followed by the name of the tool, and uncomment the
"Generated-by" line. See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed.
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name and Version]
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4550:
URL: https://github.com/apache/flink-cdc/pull/4550#discussion_r4044631992
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/enumerator/IncrementalSourceEnumeratorReleaseTest.java:
##########
@@ -0,0 +1,458 @@
+/*
+ * 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.base.source.enumerator;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderInfo;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.cdc.connectors.base.config.SourceConfig;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.assigner.AssignerStatus;
+import org.apache.flink.cdc.connectors.base.source.assigner.SplitAssigner;
+import org.apache.flink.cdc.connectors.base.source.assigner.state.PendingSplitsState;
+import org.apache.flink.cdc.connectors.base.source.assigner.state.StreamPendingSplitsState;
+import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitAssignedEvent;
+import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitMetaAssembledEvent;
+import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitMetaEvent;
+import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitMetaRequestEvent;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import org.apache.flink.cdc.connectors.base.source.meta.split.FinishedSnapshotSplitInfo;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
+import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
+
+import io.debezium.relational.TableId;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Coordinator-level end-to-end test for the snapshot-metadata release trigger. Drives the real
+ * {@link IncrementalSourceEnumerator} through the event protocol with a {@link
+ * MockSplitEnumeratorContext}: the enumerator serves the finished-split metadata, the reader
+ * reports it assembled, and the enumerator releases it from the assigner and its own cache once a
+ * checkpoint covers the assembled split. Generalizes the MySQL release (FLINK-39775) to the base
+ * framework.
+ */
+class IncrementalSourceEnumeratorReleaseTest {
+
+ private static final TableId TABLE_ID = new TableId("test_db", null, "customers");
+ private static final String STREAM_SPLIT_ID = "stream-split";
+ private static final OffsetFactory OFFSET_FACTORY = new NullOffsetFactory();
+
+ @Test
+ void testReleaseFiresAfterAssembledAndCheckpoint() throws Exception {
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ // 1) The reader requests the metadata group; the enumerator serves it from the assigner
+ // and builds its finishedSnapshotSplitMeta cache.
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ assertThat(sentEventsTo(context, 0)).anyMatch(e -> e instanceof StreamSplitMetaEvent);
+ assertThat(finishedSnapshotSplitMeta(enumerator)).isNotNull();
+
+ // 2) The reader reports the metadata fully assembled, but the release must NOT fire yet
+ // (no checkpoint has covered the assembled split).
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+ assertThat(assigner.released).isFalse();
+
+ // 3) A checkpoint is taken (arms the release) and then completes (fires it).
+ enumerator.snapshotState(100L);
+ assertThat(assigner.released).isFalse();
+ enumerator.notifyCheckpointComplete(100L);
+ assertThat(assigner.released).isTrue();
+ assertThat(assigner.releaseCalls).isEqualTo(1);
+
+ // 4) The enumerator-side metadata cache was cleared too (memory reclaimed).
+ assertThat(finishedSnapshotSplitMeta(enumerator)).isNull();
+
+ // 5) A stale metadata request after release is ignored gracefully: no exception, and no
+ // further metadata is sent to the reader.
+ int sentBefore = sentEventsTo(context, 0).size();
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ assertThat(sentEventsTo(context, 0)).hasSize(sentBefore);
+
+ // 6) Release is idempotent across further checkpoints.
+ enumerator.notifyCheckpointComplete(101L);
+ assertThat(assigner.releaseCalls).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void testReleaseDoesNotFireWhenOptionDisabled() throws Exception {
+ // The release is opt-in. With the option off (the default), the metadata is retained
+ // exactly as before, even after the assembled event and a checkpoint.
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(false, false), // release disabled
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+ enumerator.snapshotState(100L);
+ enumerator.notifyCheckpointComplete(100L);
+
+ assertThat(assigner.released).isFalse();
+ assertThat(finishedSnapshotSplitMeta(enumerator)).isNotNull();
+ }
+ }
+
+ @Test
+ void testStaleAssembledReportIsIgnoredAfterStreamSplitReassignment() throws Exception {
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ // The enumerator serves metadata at generation 0.
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+
+ // The stream split's reader fails over: the split is added back, bumping the
+ // generation.
+ List<SourceSplitBase> addedBack = new ArrayList<>();
+ addedBack.add(
+ new StreamSplit(
+ STREAM_SPLIT_ID, null, null, new ArrayList<>(), new HashMap<>(), 2));
+ enumerator.addSplitsBack(addedBack, 0);
+
+ // A stale assembled report (generation 0) must NOT trigger a release.
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+ enumerator.snapshotState(100L);
+ enumerator.notifyCheckpointComplete(100L);
+ assertThat(assigner.released).isFalse();
+
+ // A fresh report at the current generation (1) does trigger the release.
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 1));
+ enumerator.snapshotState(101L);
+ enumerator.notifyCheckpointComplete(101L);
+ assertThat(assigner.released).isTrue();
+ }
+ }
+
+ @Test
+ void testEmptyAddBackFromStreamReaderBumpsGeneration() throws Exception {
+ // If the stream reader's assignment was already checkpoint-covered, a reader reset
+ // arrives as an empty addSplitsBack. The generation must still bump so a stale assembled
+ // report from the failed attempt cannot arm an unsafe release.
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ // serve metadata at generation 0, and record that subtask 0 holds the stream split
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ enumerator.handleSourceEvent(0, new StreamSplitAssignedEvent());
+
+ // The reader resets without handing the split back (checkpoint-covered), which
+ // arrives as an empty add-back.
+ enumerator.addSplitsBack(new ArrayList<>(), 0);
+
+ // A stale assembled report at the old generation (0) must NOT trigger a release.
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+ enumerator.snapshotState(100L);
+ enumerator.notifyCheckpointComplete(100L);
+ assertThat(assigner.released).isFalse();
+
+ // A fresh report at the current generation (1) does trigger the release.
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 1));
+ enumerator.snapshotState(101L);
+ enumerator.notifyCheckpointComplete(101L);
+ assertThat(assigner.released).isTrue();
+ }
+ }
+
+ @Test
+ void testReleaseDoesNotFireWhenNewlyAddedTablesEnabled() throws Exception {
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(true), // scan.newly.added.tables enabled
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ // The reader serves and assembles the metadata exactly as in the happy path.
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+
+ // But because newly-added-tables capture can re-enter snapshot assignment later and the
+ // suspended stream reader must rebuild its split from this metadata, the release must
+ // NOT arm or fire even after a checkpoint completes. The metadata stays intact.
+ enumerator.snapshotState(100L);
+ enumerator.notifyCheckpointComplete(100L);
+ assertThat(assigner.released).isFalse();
+ assertThat(finishedSnapshotSplitMeta(enumerator)).isNotNull();
+ }
+ }
+
+ private static Object finishedSnapshotSplitMeta(IncrementalSourceEnumerator enumerator)
+ throws Exception {
+ java.lang.reflect.Field field =
+ IncrementalSourceEnumerator.class.getDeclaredField("finishedSnapshotSplitMeta");
Review Comment:
**nit: use a package-private getter instead of reflection to access `finishedSnapshotSplitMeta`**
`IncrementalSourceEnumeratorReleaseTest` uses reflection to inspect `IncrementalSourceEnumerator.finishedSnapshotSplitMeta` to verify the cache is cleared after release. Reflection is brittle — renaming the field would silently break the test.
Consider adding a package-private getter on `IncrementalSourceEnumerator`:
```java
@VisibleForTesting
List<FinishedSnapshotSplitInfo> getFinishedSnapshotSplitMeta() {
return finishedSnapshotSplitMeta;
}
```
Minor, not blocking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase [flink-cdc]
Posted by "spoorthibasu (via GitHub)" <gi...@apache.org>.
spoorthibasu commented on PR #4550:
URL: https://github.com/apache/flink-cdc/pull/4550#issuecomment-5749917274
Thanks @lvyanquan, both are addressed inline.
Three other changes in this push:
- The release clears its armed checkpoint once it runs, so a stream-only job, which has no snapshot metadata to drop, doesn't re-enter that path on every later checkpoint.
- `IncrementalSourceReader` coverage for the assembled event: `testReaderSendsStreamSplitMetaAssembledEventForCompleteSplit` on the inline path and `testReaderEchoesAssignmentGenerationInAssembledEvent` on the group-fetch path.
- Ticket references follow the same split throughout, FLINK-40697 for what this ticket adds and FLINK-39775 where it points back at the original MySQL change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase [flink-cdc]
Posted by "spoorthibasu (via GitHub)" <gi...@apache.org>.
spoorthibasu commented on PR #4550:
URL: https://github.com/apache/flink-cdc/pull/4550#issuecomment-5748533510
Thanks for the review @lvyanquan. I've also prepared the same wiring for the MongoDB connector, which is the one remaining incremental dialect. Happy to open that as a follow-up once this is merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4550:
URL: https://github.com/apache/flink-cdc/pull/4550#discussion_r4056591769
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java:
##########
@@ -481,6 +501,14 @@ private void requestStreamSplitMetaIfNeeded(StreamSplit streamSplit) {
} else {
LOG.info("The meta of stream split {} has been collected success", splitId);
this.addSplits(Collections.singletonList(streamSplit));
+ // Report the stream-split metadata assembled so the coordinator can release what it no
+ // longer needs to serve. Carries the assignment generation so a stale report from a
+ // failed attempt is ignored.
+ context.sendSourceEventToCoordinator(
Review Comment:
addSplits(streamSplit) enters the completed stream-split branch above, which already sends a StreamSplitMetaAssembledEvent. This therefore sends the event twice on the group-fetch path. Could we remove this duplicate notification and update the comment in the completed-split branch to also mention metadata assembled through group fetch? Non-blocking, since the coordinator handling is idempotent.
##########
docs/content/docs/connectors/flink-sources/db2-cdc.md:
##########
@@ -263,6 +263,13 @@ Db2 server.
If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true,
so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
</td>
+ </tr>
+ <tr>
+ <td>scan.incremental.snapshot.metadata.release.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to release the snapshot split metadata (assigned splits, finished offsets and table schemas) held by the source coordinator once the source has entered the stream phase, to reduce JobManager memory on jobs with a very large number of snapshot splits. Disabled by default. Incompatible with scan.newly-added-table.enabled: enabling both fails at startup, and a job that has released the metadata cannot later enable newly-added-table scanning. Release happens only after a successful checkpoint; if checkpointing is disabled or no checkpoint completes, the metadata is retained, so this option has no effect without checkpointing.</td>
Review Comment:
Could we document the downgrade restriction for all connectors that expose this option, including MySQL? Once a checkpoint or savepoint is created with this option enabled, it cannot be used to restore the job after downgrading to Flink CDC 3.6.0 or an earlier version. MySQL has the same state compatibility restriction and exposes this option in both the Flink Source and Pipeline Connector APIs, so the corresponding English and Chinese MySQL documentation should be updated as well. This is non-blocking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #4550:
URL: https://github.com/apache/flink-cdc/pull/4550
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase [flink-cdc]
Posted by "spoorthibasu (via GitHub)" <gi...@apache.org>.
spoorthibasu commented on code in PR #4550:
URL: https://github.com/apache/flink-cdc/pull/4550#discussion_r4056463154
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/enumerator/IncrementalSourceEnumeratorReleaseTest.java:
##########
@@ -0,0 +1,458 @@
+/*
+ * 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.base.source.enumerator;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderInfo;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.cdc.connectors.base.config.SourceConfig;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.assigner.AssignerStatus;
+import org.apache.flink.cdc.connectors.base.source.assigner.SplitAssigner;
+import org.apache.flink.cdc.connectors.base.source.assigner.state.PendingSplitsState;
+import org.apache.flink.cdc.connectors.base.source.assigner.state.StreamPendingSplitsState;
+import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitAssignedEvent;
+import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitMetaAssembledEvent;
+import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitMetaEvent;
+import org.apache.flink.cdc.connectors.base.source.meta.events.StreamSplitMetaRequestEvent;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import org.apache.flink.cdc.connectors.base.source.meta.split.FinishedSnapshotSplitInfo;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
+import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
+
+import io.debezium.relational.TableId;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Coordinator-level end-to-end test for the snapshot-metadata release trigger. Drives the real
+ * {@link IncrementalSourceEnumerator} through the event protocol with a {@link
+ * MockSplitEnumeratorContext}: the enumerator serves the finished-split metadata, the reader
+ * reports it assembled, and the enumerator releases it from the assigner and its own cache once a
+ * checkpoint covers the assembled split. Generalizes the MySQL release (FLINK-39775) to the base
+ * framework.
+ */
+class IncrementalSourceEnumeratorReleaseTest {
+
+ private static final TableId TABLE_ID = new TableId("test_db", null, "customers");
+ private static final String STREAM_SPLIT_ID = "stream-split";
+ private static final OffsetFactory OFFSET_FACTORY = new NullOffsetFactory();
+
+ @Test
+ void testReleaseFiresAfterAssembledAndCheckpoint() throws Exception {
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ // 1) The reader requests the metadata group; the enumerator serves it from the assigner
+ // and builds its finishedSnapshotSplitMeta cache.
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ assertThat(sentEventsTo(context, 0)).anyMatch(e -> e instanceof StreamSplitMetaEvent);
+ assertThat(finishedSnapshotSplitMeta(enumerator)).isNotNull();
+
+ // 2) The reader reports the metadata fully assembled, but the release must NOT fire yet
+ // (no checkpoint has covered the assembled split).
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+ assertThat(assigner.released).isFalse();
+
+ // 3) A checkpoint is taken (arms the release) and then completes (fires it).
+ enumerator.snapshotState(100L);
+ assertThat(assigner.released).isFalse();
+ enumerator.notifyCheckpointComplete(100L);
+ assertThat(assigner.released).isTrue();
+ assertThat(assigner.releaseCalls).isEqualTo(1);
+
+ // 4) The enumerator-side metadata cache was cleared too (memory reclaimed).
+ assertThat(finishedSnapshotSplitMeta(enumerator)).isNull();
+
+ // 5) A stale metadata request after release is ignored gracefully: no exception, and no
+ // further metadata is sent to the reader.
+ int sentBefore = sentEventsTo(context, 0).size();
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ assertThat(sentEventsTo(context, 0)).hasSize(sentBefore);
+
+ // 6) Release is idempotent across further checkpoints.
+ enumerator.notifyCheckpointComplete(101L);
+ assertThat(assigner.releaseCalls).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void testReleaseDoesNotFireWhenOptionDisabled() throws Exception {
+ // The release is opt-in. With the option off (the default), the metadata is retained
+ // exactly as before, even after the assembled event and a checkpoint.
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(false, false), // release disabled
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+ enumerator.snapshotState(100L);
+ enumerator.notifyCheckpointComplete(100L);
+
+ assertThat(assigner.released).isFalse();
+ assertThat(finishedSnapshotSplitMeta(enumerator)).isNotNull();
+ }
+ }
+
+ @Test
+ void testStaleAssembledReportIsIgnoredAfterStreamSplitReassignment() throws Exception {
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ // The enumerator serves metadata at generation 0.
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+
+ // The stream split's reader fails over: the split is added back, bumping the
+ // generation.
+ List<SourceSplitBase> addedBack = new ArrayList<>();
+ addedBack.add(
+ new StreamSplit(
+ STREAM_SPLIT_ID, null, null, new ArrayList<>(), new HashMap<>(), 2));
+ enumerator.addSplitsBack(addedBack, 0);
+
+ // A stale assembled report (generation 0) must NOT trigger a release.
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+ enumerator.snapshotState(100L);
+ enumerator.notifyCheckpointComplete(100L);
+ assertThat(assigner.released).isFalse();
+
+ // A fresh report at the current generation (1) does trigger the release.
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 1));
+ enumerator.snapshotState(101L);
+ enumerator.notifyCheckpointComplete(101L);
+ assertThat(assigner.released).isTrue();
+ }
+ }
+
+ @Test
+ void testEmptyAddBackFromStreamReaderBumpsGeneration() throws Exception {
+ // If the stream reader's assignment was already checkpoint-covered, a reader reset
+ // arrives as an empty addSplitsBack. The generation must still bump so a stale assembled
+ // report from the failed attempt cannot arm an unsafe release.
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(),
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ // serve metadata at generation 0, and record that subtask 0 holds the stream split
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ enumerator.handleSourceEvent(0, new StreamSplitAssignedEvent());
+
+ // The reader resets without handing the split back (checkpoint-covered), which
+ // arrives as an empty add-back.
+ enumerator.addSplitsBack(new ArrayList<>(), 0);
+
+ // A stale assembled report at the old generation (0) must NOT trigger a release.
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+ enumerator.snapshotState(100L);
+ enumerator.notifyCheckpointComplete(100L);
+ assertThat(assigner.released).isFalse();
+
+ // A fresh report at the current generation (1) does trigger the release.
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 1));
+ enumerator.snapshotState(101L);
+ enumerator.notifyCheckpointComplete(101L);
+ assertThat(assigner.released).isTrue();
+ }
+ }
+
+ @Test
+ void testReleaseDoesNotFireWhenNewlyAddedTablesEnabled() throws Exception {
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1)) {
+ context.registerReader(new ReaderInfo(0, "localhost"));
+
+ RecordingSplitAssigner assigner = new RecordingSplitAssigner();
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(
+ context,
+ new StubSourceConfig(true), // scan.newly.added.tables enabled
+ assigner,
+ Boundedness.CONTINUOUS_UNBOUNDED);
+ enumerator.start();
+
+ // The reader serves and assembles the metadata exactly as in the happy path.
+ enumerator.handleSourceEvent(0, new StreamSplitMetaRequestEvent(STREAM_SPLIT_ID, 0, 2));
+ enumerator.handleSourceEvent(
+ 0, new StreamSplitMetaAssembledEvent(STREAM_SPLIT_ID, 2, 0));
+
+ // But because newly-added-tables capture can re-enter snapshot assignment later and the
+ // suspended stream reader must rebuild its split from this metadata, the release must
+ // NOT arm or fire even after a checkpoint completes. The metadata stays intact.
+ enumerator.snapshotState(100L);
+ enumerator.notifyCheckpointComplete(100L);
+ assertThat(assigner.released).isFalse();
+ assertThat(finishedSnapshotSplitMeta(enumerator)).isNotNull();
+ }
+ }
+
+ private static Object finishedSnapshotSplitMeta(IncrementalSourceEnumerator enumerator)
+ throws Exception {
+ java.lang.reflect.Field field =
+ IncrementalSourceEnumerator.class.getDeclaredField("finishedSnapshotSplitMeta");
Review Comment:
Replaced the reflection with a package-private `@VisibleForTesting` `getFinishedSnapshotSplitMeta()` on the enumerator, and the test now calls it directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40697][cdc-base] Release finished snapshot split metadata after entering the stream phase [flink-cdc]
Posted by "spoorthibasu (via GitHub)" <gi...@apache.org>.
spoorthibasu commented on code in PR #4550:
URL: https://github.com/apache/flink-cdc/pull/4550#discussion_r4056979303
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java:
##########
@@ -481,6 +501,14 @@ private void requestStreamSplitMetaIfNeeded(StreamSplit streamSplit) {
} else {
LOG.info("The meta of stream split {} has been collected success", splitId);
this.addSplits(Collections.singletonList(streamSplit));
+ // Report the stream-split metadata assembled so the coordinator can release what it no
+ // longer needs to serve. Carries the assignment generation so a stale report from a
+ // failed attempt is ignored.
+ context.sendSourceEventToCoordinator(
Review Comment:
Removed the duplicate send, and the comment in the completed-split branch now mentions the group-fetch case along with the others that reach it.
##########
docs/content/docs/connectors/flink-sources/db2-cdc.md:
##########
@@ -263,6 +263,13 @@ Db2 server.
If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true,
so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
</td>
+ </tr>
+ <tr>
+ <td>scan.incremental.snapshot.metadata.release.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to release the snapshot split metadata (assigned splits, finished offsets and table schemas) held by the source coordinator once the source has entered the stream phase, to reduce JobManager memory on jobs with a very large number of snapshot splits. Disabled by default. Incompatible with scan.newly-added-table.enabled: enabling both fails at startup, and a job that has released the metadata cannot later enable newly-added-table scanning. Release happens only after a successful checkpoint; if checkpointing is disabled or no checkpoint completes, the metadata is retained, so this option has no effect without checkpointing.</td>
Review Comment:
Documented for the four dialects and for MySQL, in both the Flink Source and Pipeline Connector docs, EN and ZH.
The Db2 docs also drop the `scan.newly-added-table.enabled` line, since Db2 doesn't expose that option.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org