You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "spuru9 (via GitHub)" <gi...@apache.org> on 2026/09/20 22:17:47 UTC
[PR] [FLINK-40725][tests] Fix changelog-source sink ITCase for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
spuru9 opened a new pull request, #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232
## What is the purpose of the change
Flink 2.3.0 (FLIP-558, FLINK-38926) tightened the planner: a changelog query whose upsert key differs from the sink primary key is now rejected unless the `INSERT` declares an `ON CONFLICT` clause. `JdbcDynamicTableSinkITCase.testReadingFromChangelogSource` — the shared base test inherited by every database module's sink ITCase — therefore fails on Flink 2.3.0 with a `ValidationException`, while passing on 2.1.x/2.2.x. See FLINK-40725.
This makes the test compatible across the supported Flink range and adds Flink 2.3.0 to the PR CI matrix so the path is actually exercised.
## Brief change log
- `JdbcDynamicTableSinkITCase.testReadingFromChangelogSource`: append `ON CONFLICT DO DEDUPLICATE` to the `INSERT` only when running against Flink >= 2.3.0. `DO DEDUPLICATE` keeps the latest record per key, matching the prior upsert semantics.
- Version detection uses `EnvironmentInformation.getVersion()` (leading `major.minor`), which compiles against every supported Flink version — the `FlinkVersion` enum constants for 2.2/2.3 do not exist in the connector's `flink.version=2.0.0` baseline, so they cannot be referenced.
- Add `2.3.0` to the `push_pr.yml` `flink` matrix so this fix is exercised in CI.
## Verifying this change
This change is already covered by existing tests.
`DerbyDynamicTableSinkITCase#testReadingFromChangelogSource` (the core, Docker-free realization of the shared base test) was run locally and passes on Flink `2.1.3`, `2.2.1`, and `2.3.0`; the default `flink.version=2.0.0` build still compiles. The `2.3.0` job added to `push_pr.yml` exercises the fixed path in CI.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency, including `flink.version` or a database driver version): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)` or `@Experimental`, or are the Table options or the JDBC dialect SPI changed: no
- Checkpointed state, its serializers, or exactly-once delivery: no
- The per-record code paths: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (Claude Code)
Generated-by: Claude Code (Claude Opus 4.8)
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#issuecomment-5757348647
that's true
however couldn't you still use just regular string comparison?
Given the fact there is a dot in version
2.2 is always < 2.3, same for 2.1, 1.20
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#discussion_r4060483684
##########
flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java:
##########
@@ -362,9 +363,14 @@ protected void testReadingFromChangelogSource() throws Exception {
"'sink.buffer-flush.max-rows' = '2'",
"'sink.buffer-flush.interval' = '0'")));
+ String onConflict =
+ FlinkVersion.current().toString().compareTo("2.3") >= 0
Review Comment:
```
FlinkVersion.current().toString().compareTo("2.3") >= 0
```
probably better to extract it into helper method something like `isFlink558Enabled`
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "spuru9 (via GitHub)" <gi...@apache.org>.
spuru9 commented on PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#issuecomment-5756976038
Just to check my understanding, when we compile against `-Dflink.version=2.1.2`, we will be overriding the pom.xml arg and get link-annotations:2.1.3, which does not have a `FlinkVersion.2_3`. Wont that be causing a compile time failure?
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "spuru9 (via GitHub)" <gi...@apache.org>.
spuru9 commented on code in PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#discussion_r4060537150
##########
flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java:
##########
@@ -362,9 +363,14 @@ protected void testReadingFromChangelogSource() throws Exception {
"'sink.buffer-flush.max-rows' = '2'",
"'sink.buffer-flush.interval' = '0'")));
+ String onConflict =
+ FlinkVersion.current().toString().compareTo("2.3") >= 0
Review Comment:
Ok, I will just extract in a private function as a Util class seems to be overkill for 2 liner.
--
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-40725][tests] Fix changelog-source sink ITCase for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "spuru9 (via GitHub)" <gi...@apache.org>.
spuru9 commented on code in PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#discussion_r4058270223
##########
.github/workflows/push_pr.yml:
##########
@@ -28,7 +28,7 @@ jobs:
compile_and_test:
strategy:
matrix:
- flink: [2.0.1, 2.1.1]
+ flink: [2.0.1, 2.1.1, 2.3.0]
Review Comment:
Temporary change to test the resolution against 2.3.x. To be added in #228.
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#discussion_r4060483684
##########
flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java:
##########
@@ -362,9 +363,14 @@ protected void testReadingFromChangelogSource() throws Exception {
"'sink.buffer-flush.max-rows' = '2'",
"'sink.buffer-flush.interval' = '0'")));
+ String onConflict =
+ FlinkVersion.current().toString().compareTo("2.3") >= 0
Review Comment:
```
FlinkVersion.current().toString().compareTo("2.3") >= 0
```
probably better to extract it into helper method something like `isFlip558Enabled`
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#discussion_r4060546813
##########
flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java:
##########
@@ -362,9 +363,14 @@ protected void testReadingFromChangelogSource() throws Exception {
"'sink.buffer-flush.max-rows' = '2'",
"'sink.buffer-flush.interval' = '0'")));
+ String onConflict =
+ FlinkVersion.current().toString().compareTo("2.3") >= 0
Review Comment:
yes, helper method in utility should be enough
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#discussion_r4060834399
##########
flink-connector-jdbc-mysql/src/test/java/org/apache/flink/connector/jdbc/mysql/database/catalog/MySqlCatalogTestBase.java:
##########
@@ -513,12 +514,14 @@ void testSelectToInsert() throws Exception {
@Test
void testGroupByInsert() throws Exception {
// Changes primary key for the next record.
+ String onConflict = isFlink558Enabled() ? " ON CONFLICT DO DEDUPLICATE" : "";
Review Comment:
```suggestion
String onConflict = isFlip558Enabled() ? " ON CONFLICT DO DEDUPLICATE" : "";
```
based on https://issues.apache.org/jira/browse/FLINK-38926
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "spuru9 (via GitHub)" <gi...@apache.org>.
spuru9 commented on PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#issuecomment-5756568722
> why do we need FlinkVersionUtils and can't use FlinkVersion?
I've switched the helper to use the public FlinkVersion.current() instead of EnvironmentInformation.
It still needs to be a small helper rather than an inline FlinkVersion.current().isNewerVersionThan(FlinkVersion.v2_3), because the connector compiles against its baseline flink.version and that flink-annotations only declares FlinkVersion up to the baseline's own version. Referencing v2_3 fails to compile there — and this holds for the planned 2.1.2 baseline too, whose enum stops at v2_1:
```
error: cannot find symbol
return FlinkVersion.current().isNewerVersionThan(FlinkVersion.v2_3);
^
symbol: variable v2_3 (flink-annotations 2.1.2 declares up to v2_1)
```
And as a inline is not possible, and its a block of code, I am adding a Util. Can add a TODO for cleanup post the compile version has moved 2.3+.
If there is another way let me know.
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232#issuecomment-5756694217
I still don't understand why we need to do any kind of `toString` and `split`
why can't we just use
```java
if (FlinkVersion.current().isNewerVersionThan(FlinkVersion.2_3)) {
...
} else {
...
}
```
?
--
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-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 ON CONFLICT validation [flink-connector-jdbc]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin merged PR #232:
URL: https://github.com/apache/flink-connector-jdbc/pull/232
--
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