Skip to content

[flink] Expose Flink Source in FlinkSourceBuilder API - #10713

Open
eye-gu wants to merge 1 commit into
apache:masterfrom
eye-gu:issue-7395
Open

eye-gu wants to merge 1 commit into
apache:masterfrom
eye-gu:issue-7395

Conversation

@eye-gu

@eye-gu eye-gu commented Oct 21, 2026

Copy link
Copy Markdown
Member

Closes #7395.

Purpose

FlinkSourceBuilder can only produce a DataStream through build() and buildForRow(). Users who need a plain Source instance — for example to bootstrap a pipeline with historical data from a bounded Paimon source and continue streaming within one Flink HybridSource — currently have no way to obtain one.

This PR adds FlinkSourceBuilder.buildSource(), returning a Source<RowData, ?, ?> wrapped in PaimonDataStreamSource to preserve lineage (#8001). Unlike build(), no StreamExecutionEnvironment is required: boundedness is either inferred from the environment set via env() or must be given explicitly with sourceBounded(...).

build() and buildSource() share a single private construction path (createSource()), and the consumer-id validation moved into checkConsumerId() so both entry points keep it. For build() this is a pure refactor: the extra checks inside createSource() are unreachable from that path.

Only the default static and continuous scan paths are supported by buildSource(); postpone merge-on-read, scan.dedicated-split-generation, source.checkpoint-align.enabled and exactly-once consumer mode rely on other topologies and throw UnsupportedOperationException pointing to build(). The exception is postpone merge-on-read with scan.mode=compacted-full, which reads through the static path, mirroring the existing fall-through in build() (#10521).

The javadoc shows a HybridSource example and documents the pitfalls: an explicit sourceBounded(true) is needed when the bounded segment runs in a streaming job; there is no snapshot hand-off between segments, so scan.mode=latest may skip changes committed around the switch while the default latest-full re-reads and duplicates the bounded segment; and env.fromSource(...) requires an explicit TypeInformation because HybridSource is not ResultTypeQueryable.

Tests

FlinkSourceBuilderTest (26 tests): boundedness via sourceBounded(...), via env inference, missing boundedness → error; the four unsupported modes rejected; postpone merge-on-read with scan.mode=compacted-full still builds through the static path in both buildSource() and build() — locking the new checks to build()'s existing fall-through behavior.

HybridSourceITCase: a streaming job whose HybridSource embeds a bounded buildSource() segment followed by a continuous one. The continuous segment filters a <= 2, so rows 1–2 can only come from the bounded segment; DISTINCT tolerates the latest-full re-read overlap. All tests passed locally:

mvn -o -pl paimon-flink/paimon-flink-common -am -Pflink1 \
  -DfailIfNoTests=false -DwildcardSuites=none \
  '-Dtest=FlinkSourceBuilderTest,HybridSourceITCase' test

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Feature] Expose Flink Source in FlinkSourceBuilder API

1 participant