[Feature][Flink] Add flink version 1.20.1 support #9576

yzeng1618 · 2025-07-15T11:53:13Z

Purpose of this pull request

This pull request fixes the Flink 1.20 compatibility issue in the SeaTunnel Flink translation layer. The current implementation uses reflection to access internal StreamingRuntimeContext from SourceReaderContext, which is fragile and may break with Flink version updates.

Does this PR introduce any user-facing change?

No. This is an internal implementation fix that maintains the same public API behavior. Users will continue to use the same SeaTunnel connector APIs without any changes. The metrics functionality remains identical from the user perspective.

How was this patch tested?

Manual Testing:
Built SeaTunnel with Flink 1.20 dependencies
Ran sample connector jobs to verify metrics collection
Confirmed no reflection-related errors in logs
Validated metrics appear correctly in Flink Web UI

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update plugin-mapping.properties and add new connector information in it
2. Update the pom file of seatunnel-dist
3. Add ci label in label-scope-conf
4. Add e2e testcase in seatunnel-e2e
5. Update connector plugin_config

Hisoka-X · 2025-07-15T11:54:04Z

cc @TyrantLucifer

TyrantLucifer · 2025-07-15T11:55:31Z

Good pr

...-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java

...ation-flink-20/src/main/java/org/apache/seatunnel/translation/flink/source/SplitWrapper.java

Hisoka-X

We need new flink 1.20.1 test container to verify it. https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/flink

…n-flink-common

yzeng1618 · 2025-09-03T01:44:08Z

We need new flink 1.20.1 test container to verify it. https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/flink

The test container for Flink 1.20 has been added.

Copilot

Pull Request Overview

This PR adds support for Flink 1.20.1 to SeaTunnel's translation layer by creating new Flink 1.20-specific implementations that use Flink's official sink2 API instead of the fragile reflection-based approach. The changes ensure compatibility while maintaining the same public API for users.

Creates Flink 1.20 translation layer with modern sink2 API integration
Adds comprehensive Flink 1.20 starter components and build configuration
Updates E2E testing infrastructure to include Flink 1.20 support

Reviewed Changes

Copilot reviewed 49 out of 51 changed files in this pull request and generated 3 comments.

File	Description
seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-20/	Complete Flink 1.20 translation layer implementation using sink2 API
seatunnel-core/seatunnel-flink-starter/seatunnel-flink-20-starter/	Flink 1.20 starter implementation with refactored architecture
seatunnel-e2e/	E2E test infrastructure updates for Flink 1.20 compatibility
seatunnel-dist/	Distribution assembly configuration for Flink 1.20 components

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

...ansforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestEmbeddingIT.java

...ink-20/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterContext.java

...e-common/src/test/java/org/apache/seatunnel/e2e/common/container/flink/Flink20Container.java

...er/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java

...ain/java/org/apache/seatunnel/core/starter/flink/execution/AbstractSinkExecuteProcessor.java

...ink-20/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterContext.java

…apper

yzeng1618

It has been replied to now.

yzeng1618 · 2025-08-26T01:54:50Z

...-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java

+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class FlinkExecution implements TaskExecution {


Because there is additional logic for the createFlink20JobMetricsSummary method in FlinkExecution

...rc/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceReader.java

...tor-v2-e2e/connector-doris-e2e/src/test/resources/doris_source_and_sink_with_custom_sql.conf

...connector-v2-e2e/connector-file-sftp-e2e/src/test/resources/text/fake_to_sftp_file_text.conf

...2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/AbstractTestContainer.java

...ink-20/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterContext.java

...ain/java/org/apache/seatunnel/core/starter/flink/execution/AbstractSinkExecuteProcessor.java

...er/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java

...ansforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestEmbeddingIT.java

yzeng1618 · 2025-09-23T01:16:38Z

We need new flink 1.20.1 test container to verify it. https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/flink

It has been added.

Carl-Zhou-CN

LGTM

yzeng1618 · 2025-09-26T12:08:13Z

LGTM

Because this commit is merged into dev, we need to resolve conflicts and then re-run the CI process. Please approve it again.

.github/workflows/backend.yml

...-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java

...-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkAccumulatorCounter.java

...ation-flink-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkCounter.java

...k-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkJobMetricsSummary.java

...slation-flink-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkMeter.java

Hisoka-X · 2025-09-26T13:20:10Z

...flink-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkMetricContext.java

+        }
+
+        if (metricGroup == null) {
+            Counter noOpCounter = new NoOpCounter();


why we need NoOpCounter?

why we need NoOpCounter?

When metricGroup == null (e.g., in scenarios such as certain test cases, initialization phases, or when the context is not in the Flink Runtime environment), returning a NoOp (No Operation) instance prevents NullPointerException and ensures the business logic remains operational.

It's werid. Why other flink version does not need this?

…PL-support1.20.1-test

Hisoka-X · 2025-09-28T14:58:34Z

...-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java

+    }
+
+    @Override
+    protected FlinkJobMetricsSummary createJobMetricsSummary(


After remove logs, I found this method same as method in common module. We shoud do some refactor.

Hisoka-X · 2025-09-28T15:00:33Z

...-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkAccumulatorCounter.java

+        return accumulator;
+    }
+
+    public void sync() {


what's for?

Hisoka-X · 2025-09-28T15:00:57Z

...-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkAccumulatorCounter.java

+    private final org.apache.flink.metrics.Counter flinkCounter;
+    private final LongCounter accumulator;
+    private final RuntimeContext runtimeContext;
+    private final java.util.concurrent.atomic.AtomicLong localCount =
+            new java.util.concurrent.atomic.AtomicLong(0L);


why we need two fields to store value?

Hisoka-X · 2025-09-28T15:02:05Z

...-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkAccumulatorCounter.java

+    }
+
+    private String getStandardAccumulatorName(String originalName) {
+        if (originalName.contains("SinkWriteCount")


same as #9576 (comment)

Hisoka-X · 2025-09-28T15:03:24Z

...flink-20/src/main/java/org/apache/seatunnel/translation/flink/metric/FlinkMetricContext.java

+        }
+
+        if (metricGroup == null) {
+            Counter noOpCounter = new NoOpCounter();


It's werid. Why other flink version does not need this?

zengyi added 6 commits July 3, 2025 17:40

[improve][start] version1.0

cd26913

[improve][start] version2.0

9dba128

[improve][start] version3.0

e020b82

[improve][start] version4.0

f870cef

[improve][start] version5.0

bef73ee

[improve][start] support 1.20.1 demo

777d513

github-actions bot added core SeaTunnel core module flink api labels Jul 15, 2025

Hisoka-X changed the title ~~[improve][start] Seatunnel version support 1.20.1~~ [Feature][Flink] Add flink version 1.20.1 support Jul 15, 2025

TyrantLucifer reviewed Jul 15, 2025

View reviewed changes

...-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java Outdated Show resolved Hide resolved

zengyi added 2 commits July 16, 2025 11:56

[improve][start] refactor the code

bc46786

Merge branch 'dev' into BDPL-support1.20.1-test

87b0319

nielifeng requested a review from Copilot July 17, 2025 09:46

This comment was marked as outdated.

Sign in to view

zengyi added 5 commits July 19, 2025 17:38

[improve][start] Remove unnecessary classes and comment content.

4229fc1

[improve][start] Remove unnecessary classes and comment content.

943c501

[improve][start] Remove unnecessary classes.

0b6ad1f

[improve][start] Fix some details

e3afc1d

[improve][start] Fix some details in the flink translation module

4bc665d

yzeng1618 requested a review from TyrantLucifer July 19, 2025 10:57

Carl-Zhou-CN reviewed Jul 26, 2025

View reviewed changes

...ation-flink-20/src/main/java/org/apache/seatunnel/translation/flink/source/SplitWrapper.java Outdated Show resolved Hide resolved

yzeng1618 requested a review from Carl-Zhou-CN July 27, 2025 09:50

Hisoka-X reviewed Jul 31, 2025

View reviewed changes

zengyi added 3 commits August 2, 2025 17:54

[improve][start] removed content consistent with seatunnel-translatio…

2a6c2ef

…n-flink-common

[improve][start] Optimized the details

96d9ebc

[improve][start] add Flink20Container

baf00f6

github-actions bot added the e2e label Aug 3, 2025

zengyi added 2 commits August 31, 2025 11:56

Merge branch 'dev' into BDPL-support1.20.1-test

bb73acd

[feature] fix ci build

f386103

yzeng1618 requested a review from Copilot September 8, 2025 01:29

Copilot AI reviewed Sep 8, 2025

View reviewed changes

Carl-Zhou-CN reviewed Sep 14, 2025

View reviewed changes

...er/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java Outdated Show resolved Hide resolved

Carl-Zhou-CN reviewed Sep 14, 2025

View reviewed changes

...ain/java/org/apache/seatunnel/core/starter/flink/execution/AbstractSinkExecuteProcessor.java Outdated Show resolved Hide resolved

Carl-Zhou-CN reviewed Sep 14, 2025

View reviewed changes

...ink-20/src/main/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterContext.java Outdated Show resolved Hide resolved

[feature] unified to prioritize LoggerFactory and remove tryGetFromWr…

c799532

…apper

yzeng1618 requested a review from Carl-Zhou-CN September 16, 2025 02:38

Merge branch 'dev' into BDPL-support1.20.1-test

6624ba3

yzeng1618 commented Sep 23, 2025

View reviewed changes

[feature] restore TestEmbeddingIT

f2dd36b

Carl-Zhou-CN previously approved these changes Sep 24, 2025

View reviewed changes

github-actions bot added approved reviewed labels Sep 24, 2025

Merge branch 'dev' into BDPL-support1.20.1-test

d257efa

yzeng1618 dismissed Carl-Zhou-CN’s stale review via d257efa September 26, 2025 11:59

github-actions bot removed approved reviewed labels Sep 26, 2025

yzeng1618 requested a review from Carl-Zhou-CN September 26, 2025 12:08

Hisoka-X reviewed Sep 26, 2025

View reviewed changes

zengyi and others added 3 commits September 28, 2025 11:24

[feature] apply reviewer feedback

35cceda

Merge branch 'apache:dev' into BDPL-support1.20.1-test

d46e2c2

Merge remote-tracking branch 'origin/BDPL-support1.20.1-test' into BD…

197324e

…PL-support1.20.1-test

yzeng1618 requested a review from Hisoka-X September 28, 2025 06:08

[Feature][hive] increase timeout

4ff372f

Hisoka-X reviewed Sep 28, 2025

View reviewed changes

[Feature][Flink] Add flink version 1.20.1 support #9576

Are you sure you want to change the base?

[Feature][Flink] Add flink version 1.20.1 support #9576

Conversation

yzeng1618 commented Jul 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

Uh oh!

Hisoka-X commented Jul 15, 2025

Uh oh!

TyrantLucifer commented Jul 15, 2025

Uh oh!

Uh oh!

This comment was marked as outdated.

Uh oh!

Uh oh!

Hisoka-X left a comment

Choose a reason for hiding this comment

Uh oh!

yzeng1618 commented Sep 3, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

yzeng1618 left a comment

Choose a reason for hiding this comment

Uh oh!

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!

yzeng1618 commented Sep 23, 2025

Uh oh!

Carl-Zhou-CN left a comment

Choose a reason for hiding this comment

Uh oh!

yzeng1618 commented Sep 26, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

yzeng1618 commented Jul 15, 2025 •

edited

Loading