[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?

tmp

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) {

Co-Authored-By: Qoder <noreply@qoder.com> AI-Contributed/Feature: 62/62 AI-Contributed/UT: 0/0

leonardBang

Thanks @loserwang1024 for the contribution, I left some comments

leonardBang · 2026-08-20T10:04:29Z


    <properties>
-        <fluss.version>0.9.0-incubating</fluss.version>
+        <fluss.version>1.0-SNAPSHOT</fluss.version>


Could we clarify which publicly available Fluss artifact this PR is expected to build against?
As of August 20, 2026, the Apache snapshot repository still resolves the relevant Fluss artifacts to 1.0-20260802.061439-1; it does not currently show a newer publication. A locally installed artifact or private mirror would not be reproducible for community builds.

leonardBang · 2026-08-20T10:05:09Z

-                CompletableFuture<?> writeFuture =
-                        write(writerMap.get(tablePath), opType, row, tablePath);
+                MultiTableWriteRecord writeRecord = toWriteRecord(opType, tablePath, row, schemaId);
+                LOG.info("------writeRecord  " + writeRecord);


Could we remove this per-record INFO log?

This is on the writer's hot path and eagerly converts every MultiTableWriteRecord to a string. Under normal production throughput it may generate a large amount of logging, add per-record overhead, and expose business row contents.

If diagnostic logging is needed, a guarded DEBUG log that avoids printing the complete record would be safer.

leonardBang · 2026-08-20T10:06:07Z

+                        : this.flussSourceReaderMetricGroup.addGroup(
+                                PARTITION_GROUP, String.valueOf(tableBucket.getPartitionId()));
+        final MetricGroup bucketGroup =
+                metricGroup.addGroup(BUCKET_GROUP, String.valueOf(tableBucket.getBucket()));


Should the metric group include the table identity?

A CDC source reader may consume multiple tables, and different tables can have the same partition and bucket IDs. In that case they register the same metric identifier, so the gauges may collide or become indistinguishable.

Could we add the database/table path or table ID to the metric group and cover two tables with the same bucket number in a test?

leonardBang · 2026-08-20T10:07:22Z

+     */
+    private void handleTableBucketChanges(List<FlussSplitBase> newSplits, Throwable error) {
+        if (error != null) {
+            LOG.error("Error creating splits for new table-buckets", error);


Could this split-initialization failure be propagated rather than returning after logging it?
Failures while resolving startup offsets or creating snapshot/log splits currently leave the discovered buckets unassigned. In one-shot discovery mode the operation is never retried, so the job may remain RUNNING without producing data.
A test that injects an offset-initialization failure and verifies that the source fails, rather than silently becoming idle, would help cover this behavior.

leonardBang · 2026-08-20T10:07:45Z

+     */
+    private void checkTableBucketChanges(List<TableBucketInfo> allBuckets, Throwable error) {
+        if (error != null) {
+            LOG.error("Error discovering subscribed table-buckets", error);


Should the initial discovery failure be propagated to the coordinator instead of only being logged?
When periodic discovery is disabled, this callback is invoked only once. If table discovery or a metadata request fails, returning here leaves the job RUNNING without assigning any splits, and there is no subsequent retry.
Could we fail the job for initial/one-shot discovery failures? For periodic discovery, an explicit bounded retry policy may be more appropriate.

leonardBang · 2026-08-20T11:17:12Z

+    public List<Event> restoreState(TablePath tablePath, int schemaId, RowType rowType) {
+        ensureCacheInitialized();
+        // Multiple splits may read log with different schemaIds; only reserve the first one.
+        if (!latestSchemaIdCache.containsKey(tablePath)) {


Could this split-initialization failure be propagated rather than returning after logging it?
Failures while resolving startup offsets or creating snapshot/log splits currently leave the discovered buckets unassigned. In one-shot discovery mode the operation is never retried, so the job may remain RUNNING without producing data.
A test that injects an offset-initialization failure and verifies that the source fails, rather than silently becoming idle, would help cover this behavior.

leonardBang · 2026-08-20T11:18:25Z

+
+        // A reader maybe subscribe multiple split, thus only inferred by the latest schema(also the
+        // widest)
+        if (isSchemaChangeEvent) {


Would it make sense to project the row whenever initialRowType differs from latestRowType, rather than only when the current record advances the cached schema ID?

The Fluss MultiTableLogScanner uses dynamic schema resolution and preserves each record's original schema ID. With multiple buckets, records may therefore arrive in the order schema 1 → schema 2 → schema 1. After schema 2 updates the cache, the final schema 1 row is read using schema 2's field converters and currently fails with:
ArrayIndexOutOfBoundsException: Index 2 out of bounds for length 2

Could we project every older-schema row to the cached latest schema? It would also be helpful to add a regression test covering schema 1 → schema 2 → schema 1.

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

loserwang1024 force-pushed the fluss-source branch 4 times, most recently from d1f6013 to 80f622c Compare August 4, 2026 12:18

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

5a7ce32

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

support complex type

291705c

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

Refactor FlussPipelineITCase from fluss to fluss.

ed5fc73

tmp

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

Expose metric for fluss source

da0efff

Co-Authored-By: Qoder <noreply@qoder.com> AI-Contributed/Feature: 62/62 AI-Contributed/UT: 0/0

leonardBang reviewed Aug 20, 2026

View reviewed changes

Navigation Menu

Uh oh!

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

[FLINK-39722][pipeline-connector][fluss] Support Fluss CDC Yaml Source.#4494
loserwang1024 wants to merge 4 commits into
apache:masterfrom
loserwang1024:fluss-source

loserwang1024 commented Aug 3, 2026

Uh oh!

fxbing commented Aug 6, 2026

Uh oh!

loserwang1024 commented Aug 7, 2026 •

edited

Loading

Uh oh!

fxbing Aug 12, 2026

Uh oh!

leonardBang left a comment

Uh oh!

leonardBang Aug 20, 2026

Uh oh!

leonardBang Aug 20, 2026

Uh oh!

leonardBang Aug 20, 2026

Uh oh!

leonardBang Aug 20, 2026 •

edited

Loading

Uh oh!

leonardBang Aug 20, 2026 •

edited

Loading

Uh oh!

leonardBang Aug 20, 2026

Uh oh!

leonardBang Aug 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

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!

leonardBang Aug 20, 2026

Choose a reason for hiding this comment

Uh oh!

leonardBang Aug 20, 2026

Choose a reason for hiding this comment

Uh oh!

leonardBang Aug 20, 2026

Choose a reason for hiding this comment

Uh oh!

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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leonardBang Aug 20, 2026

Choose a reason for hiding this comment

Uh oh!

leonardBang Aug 20, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

loserwang1024 commented Aug 7, 2026 •

edited

Loading

leonardBang Aug 20, 2026 •

edited

Loading

leonardBang Aug 20, 2026 •

edited

Loading