[FLINK-39722][pipeline-connector][fluss] Support Fluss CDC Yaml Source. by loserwang1024 · Pull Request #4494 · apache/flink-cdc

loserwang1024 · 2026-08-03T14:28:17Z

What is the purpose of this pull request?

Briefly describe the problem this PR fixes or the feature it introduces. Reference Flink JIRA ticket when possible.

Brief change log

Support Fluss yaml source.

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

This change added tests and can be verified as follows:

Added/Updated unit tests in ...*
Added/Updated integration tests in ...*
Manually tested by ...

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

fxbing · 2026-08-06T02:13:51Z

Ran this PR (5a7ce32) end-to-end as a Fluss→Fluss pipeline, built against apache/fluss main (22cb1969). Build is fine and the PR's own tests are green (74 UT + 43 IT). Two issues showed up in E2E:

Source deserializer doesn't handle ARRAY / MAP / ROW

FlussRecordDeserializer.convertFlussField covers the 16 scalar types and throws for everything else:

java.lang.UnsupportedOperationException: Unsupported Fluss data type for deserialization: ArrayType
at ...FlussRecordDeserializer.convertFlussField(FlussRecordDeserializer.java:396)
at ...FlussRecordDeserializer.convertFlussRowToCdcRecord(FlussRecordDeserializer.java:201)

Repro: a source table with tags ARRAY (same for MAP / ROW) — the job fails on the first record and goes into a restart loop. The sink side already supports these types via CdcAsFlussArray /
CdcAsFlussMap / CdcAsFlussRow, so the gap looks source-only.

CreateTableEvent isn't reconciled against an existing sink table

If the sink table already exists with fewer columns than the source, FlussMetaDataApplier treats CreateTableEvent as create-if-not-exists and leaves the physical schema as is. Upstream-only columns are then
silently dropped while the job stays RUNNING with no error.

Repro: sink (id, payload), source (id, payload, extra_col) → the row lands as (1, 'cold-start') and extra_col is lost. The coordinator does pass the full schema down:

Step 3.5 - Corresponding schema changes are: [CreateTableEvent{... extra_col STRING ...}]

A worse variant: when the sink is missing a middle column (source (id, name, sale_quantity) vs sink (id, sale_quantity)), the write still succeeds with name projected away rather than failing fast.

Would it make sense for the metadata applier to reload the real TableInfo on CreateTableEvent, append missing nullable trailing columns, and fail fast when the difference can't be reconciled that way?

loserwang1024 · 2026-08-07T06:57:45Z

@fxbing

Source deserializer doesn't handle ARRAY / MAP / ROW

support it now.

CreateTableEvent isn't reconciled against an existing sink table

This is a sink behavior, thus is not included in this PR. I have raise a new issue to solve it: https://issues.apache.org/jira/browse/FLINK-40349

fxbing · 2026-08-12T12:57:13Z

+    public SourceReader<T, FlussSplitBase> createReader(SourceReaderContext readerContext) {
+        FlussSourceReaderMetrics sourceReaderMetrics =
+                new FlussSourceReaderMetrics(readerContext.metricGroup());
+        WrapperFlussMetricRegistry metricRegistry =


WrapperFlussMetricRegistry wraps Fluss METERs as WrapperFlussMeter, but it never registers a MetricViewUpdater (or otherwise calls update() on the underlying MeterView). WrapperFlussMeter#getRate() therefore only exposes the cached value.

The new source creates this registry in FlussSource#createReader, so scanner rate metrics such as fetchRequestsPerSecond and remoteFetch* will remain at their initial 0 even while records are being fetched. Could we register and close a MetricViewUpdater for this registry, as the Fluss metric integration requires?

flink-cdc/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrapperFlussMetricRegistry.java

Line 116 in ed5fc73

private void registerMetric(MetricGroup metricGroup, Metric metric, String metricName) {

leonardBang

Thanks @loserwang1024 for the contribution, I left some comments

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 1785/4380 AI-Contributed/UT: 934/3936

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 225/225 AI-Contributed/UT: 88/429

tmp Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 5/5 AI-Contributed/UT: 0/902

Co-Authored-By: Qoder <noreply@qoder.com> Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 51/62 AI-Contributed/UT: 0/0

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 208/208 AI-Contributed/UT: 296/296

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 2/2 AI-Contributed/UT: 0/0

leonardBang

Thanks for the update. I found three additional issues during the full review.

leonardBang · 2026-08-26T09:00:20Z

@@ -26,10 +26,6 @@
 public interface TableDiscovererFactory extends ObjectIdDiscovererFactory<TableId> {

    /** Compatibility alias for the old single identifier, now meaning storage-side type. */


Could we keep the identifier() compatibility alias here? TableDiscovererFactory is @PublicEvolving, and the current master API intentionally retains this default method as the legacy name for type(). Removing it in an unrelated Fluss source PR can break third-party discoverer code compiled against the current SPI and also leaves the compatibility-alias Javadoc attached to objectIdClass(). Unless this removal is part of a separately reviewed SPI migration, restoring the alias would avoid an unnecessary compatibility regression.

Since https://issues.apache.org/jira/browse/FLINK-39732 is for the same version of the Fluss YAML source, semantic simplicity takes priority.

leonardBang · 2026-08-26T09:02:25Z

+        List<FlussSplitBase> splits = new ArrayList<>();
+
+        // Get KV snapshots for this table/partition
+        KvSnapshots kvSnapshots =


Could we acquire and checkpoint a KV snapshot lease before creating these hybrid splits? The split stores a concrete snapshot ID, but the Fluss server retains only a bounded number of completed snapshots, so an unleased snapshot may be pruned while a slow full scan is still running or before the job restores from a checkpoint/savepoint. I reproduced this with kv.snapshot.num-retained=1: after the enumerator checkpointed snapshot 0 and two newer snapshots were created, createBatchScanner(tableBucket, snapshotId) failed with KvSnapshotNotExistException; acquiring a lease made the same test pass. The native Fluss source persists its lease ID and releases snapshots only after their consumption is checkpointed. Could we implement the corresponding lease lifecycle here and add this recovery test?

leonardBang

Thanks @loserwang1024 for the update and @fxbing for the review, LGTM now

leonardBang · 2026-08-27T07:28:57Z

Will merge once CI green....

github-actions Bot added common runtime base labels Aug 3, 2026

loserwang1024 force-pushed the fluss-source branch 5 times, most recently from 80f622c to 5a7ce32 Compare August 5, 2026 01:52

loserwang1024 force-pushed the fluss-source branch from a93c81f to 291705c Compare August 6, 2026 06:24

fxbing approved these changes Aug 11, 2026

View reviewed changes

github-actions Bot added the reviewed label Aug 11, 2026

leonardBang self-requested a review August 12, 2026 09:16

fxbing reviewed Aug 12, 2026

View reviewed changes

leonardBang reviewed Aug 20, 2026

View reviewed changes

Comment thread .../java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java Outdated

Comment thread .../java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java Outdated

Hongshun Wang and others added 5 commits August 25, 2026 11:47

[FLINK-39722][pipeline-connector][fluss] Support Fluss CDC Yaml Source.

b955c6f

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 1785/4380 AI-Contributed/UT: 934/3936

support complex type

4b5d360

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 225/225 AI-Contributed/UT: 88/429

Refactor FlussPipelineITCase from fluss to fluss.

dd50c35

tmp Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 5/5 AI-Contributed/UT: 0/902

Expose metric for fluss source

05499e4

Co-Authored-By: Qoder <noreply@qoder.com> Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 51/62 AI-Contributed/UT: 0/0

modified based on leonard's advice.

3b21c89

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 208/208 AI-Contributed/UT: 296/296

loserwang1024 force-pushed the fluss-source branch from da0efff to 3b21c89 Compare August 25, 2026 06:02

loserwang1024 requested a review from leonardBang August 25, 2026 06:04

fix fluss fink for different version.

b22d3ea

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 2/2 AI-Contributed/UT: 0/0

github-actions Bot added the build label Aug 25, 2026

leonardBang reviewed Aug 25, 2026

View reviewed changes

modified based on leonard's advice 2.

24610b5

loserwang1024 requested a review from leonardBang August 26, 2026 08:22

leonardBang reviewed Aug 26, 2026

View reviewed changes

modified based on leonard'advice 3

b963116

leonardBang approved these changes Aug 27, 2026

View reviewed changes

github-actions Bot added the approved label Aug 27, 2026

leonardBang merged commit 5d86d11 into apache:master Aug 27, 2026
21 checks passed

		@@ -26,10 +26,6 @@
		public interface TableDiscovererFactory extends ObjectIdDiscovererFactory<TableId> {

		/** Compatibility alias for the old single identifier, now meaning storage-side type. */

Uh oh!

Conversation

loserwang1024 commented Aug 3, 2026

What is the purpose of this pull request?

Brief change log

Verifying this change

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

fxbing commented Aug 6, 2026

Uh oh!

loserwang1024 commented Aug 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fxbing Aug 12, 2026

Choose a reason for hiding this comment

Uh oh!

leonardBang left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

leonardBang left a comment

Choose a reason for hiding this comment

Uh oh!

leonardBang Aug 26, 2026

Choose a reason for hiding this comment

Uh oh!

loserwang1024 Aug 27, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

leonardBang Aug 26, 2026

Choose a reason for hiding this comment

Uh oh!

leonardBang left a comment

Choose a reason for hiding this comment

Uh oh!

leonardBang commented Aug 27, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

loserwang1024 commented Aug 7, 2026 •

edited

Loading