This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch v4.1
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/v4.1 by this push:
new 9bf59aca [FLINK-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0
`ON CONFLICT` validation
9bf59aca is described below
commit 9bf59acaf74b06c8fbd7b75e6b5f5fc636785112
Author: Purushottam Sinha <[email protected]>
AuthorDate: Mon Sep 21 16:44:18 2026 +0530
[FLINK-40725][tests] Fix sink/catalog ITCases for Flink 2.3.0 `ON CONFLICT`
validation
(cherry picked from commit 95816699623472e27797cc8901b4cf8c414e6d99)
---
.github/workflows/push_pr.yml | 2 +-
.../jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java | 10 +++++++++-
.../jdbc/mysql/database/catalog/MySqlCatalogTestBase.java | 12 ++++++++++--
.../database/catalog/OceanBaseMysqlCatalogITCase.java | 12 ++++++++++--
4 files changed, 30 insertions(+), 6 deletions(-)
diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml
index 1f3a58ae..409b72a4 100644
--- a/.github/workflows/push_pr.yml
+++ b/.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]
jdk: [ '17' ]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java
index 5a36f435..52ce50ae 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.jdbc.core.table.sink;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.DefaultOpenContext;
import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -362,15 +363,22 @@ public abstract class JdbcDynamicTableSinkITCase extends
AbstractTestBase implem
"'sink.buffer-flush.max-rows' = '2'",
"'sink.buffer-flush.interval' = '0'")));
+ String onConflict = isFlip558Enabled() ? " ON CONFLICT DO DEDUPLICATE"
: "";
tEnv.executeSql(
String.format(
- "INSERT INTO %s SELECT * FROM %s",
userTableSink, userTableLogs))
+ "INSERT INTO %s SELECT * FROM %s%s",
+ userTableSink, userTableLogs, onConflict))
.await();
assertThat(userOutputTable.selectAllTable(getMetadata()))
.containsExactlyInAnyOrderElementsOf(testUserData());
}
+ /** Whether the Flink version under test has the FLIP-558 ON CONFLICT
validation (>= 2.3). */
+ private static boolean isFlip558Enabled() {
+ return FlinkVersion.current().toString().compareTo("2.3") >= 0;
+ }
+
protected Map<String, String> getOptions() {
Map<String, String> options = new HashMap<>();
options.put("connector", "jdbc");
diff --git
a/flink-connector-jdbc-mysql/src/test/java/org/apache/flink/connector/jdbc/mysql/database/catalog/MySqlCatalogTestBase.java
b/flink-connector-jdbc-mysql/src/test/java/org/apache/flink/connector/jdbc/mysql/database/catalog/MySqlCatalogTestBase.java
index 07477811..c00e920d 100644
---
a/flink-connector-jdbc-mysql/src/test/java/org/apache/flink/connector/jdbc/mysql/database/catalog/MySqlCatalogTestBase.java
+++
b/flink-connector-jdbc-mysql/src/test/java/org/apache/flink/connector/jdbc/mysql/database/catalog/MySqlCatalogTestBase.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.jdbc.mysql.database.catalog;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.connector.jdbc.testutils.DatabaseTest;
import org.apache.flink.connector.jdbc.testutils.JdbcITCaseBase;
import org.apache.flink.connector.jdbc.testutils.TableManaged;
@@ -513,12 +514,14 @@ abstract class MySqlCatalogTestBase implements
JdbcITCaseBase, DatabaseTest {
@Test
void testGroupByInsert() throws Exception {
// Changes primary key for the next record.
+ String onConflict = isFlip558Enabled() ? " ON CONFLICT DO DEDUPLICATE"
: "";
tEnv.executeSql(
String.format(
"insert into `%s` select max(`pid`) `pid`,
`col_bigint` from `%s` "
- + "group by `col_bigint` ",
+ + "group by `col_bigint` %s",
TABLE_GROUPED_BY_SINK.getTableName(),
- TABLE_ALL_TYPES.getTableName()))
+ TABLE_ALL_TYPES.getTableName(),
+ onConflict))
.await();
List<Row> results =
@@ -532,4 +535,9 @@ abstract class MySqlCatalogTestBase implements
JdbcITCaseBase, DatabaseTest {
assertThat(results)
.isEqualTo(Collections.singletonList(Row.ofKind(RowKind.INSERT, 2L, -1L)));
}
+
+ /** Whether the Flink version under test has the FLIP-558 ON CONFLICT
validation (>= 2.3). */
+ private static boolean isFlip558Enabled() {
+ return FlinkVersion.current().toString().compareTo("2.3") >= 0;
+ }
}
diff --git
a/flink-connector-jdbc-oceanbase/src/test/java/org/apache/flink/connector/jdbc/oceanbase/database/catalog/OceanBaseMysqlCatalogITCase.java
b/flink-connector-jdbc-oceanbase/src/test/java/org/apache/flink/connector/jdbc/oceanbase/database/catalog/OceanBaseMysqlCatalogITCase.java
index dec4733a..b3db7e89 100644
---
a/flink-connector-jdbc-oceanbase/src/test/java/org/apache/flink/connector/jdbc/oceanbase/database/catalog/OceanBaseMysqlCatalogITCase.java
+++
b/flink-connector-jdbc-oceanbase/src/test/java/org/apache/flink/connector/jdbc/oceanbase/database/catalog/OceanBaseMysqlCatalogITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.jdbc.oceanbase.database.catalog;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.connector.jdbc.oceanbase.OceanBaseMysqlTestBase;
import
org.apache.flink.connector.jdbc.oceanbase.database.dialect.OceanBaseCompatibleMode;
import org.apache.flink.connector.jdbc.testutils.TableManaged;
@@ -354,12 +355,14 @@ public class OceanBaseMysqlCatalogITCase extends
OceanBaseCatalogITCaseBase
@Test
void testGroupByInsert() throws Exception {
// Changes primary key for the next record.
+ String onConflict = isFlip558Enabled() ? " ON CONFLICT DO DEDUPLICATE"
: "";
tEnv.executeSql(
String.format(
"insert into `%s` select max(`pid`) `pid`,
`col_bigint` from `%s` "
- + "group by `col_bigint` ",
+ + "group by `col_bigint` %s",
TABLE_GROUPED_BY_SINK.getTableName(),
- TABLE_ALL_TYPES.getTableName()))
+ TABLE_ALL_TYPES.getTableName(),
+ onConflict))
.await();
List<Row> results =
@@ -373,4 +376,9 @@ public class OceanBaseMysqlCatalogITCase extends
OceanBaseCatalogITCaseBase
assertThat(results)
.isEqualTo(Collections.singletonList(Row.ofKind(RowKind.INSERT, 2L, -1L)));
}
+
+ /** Whether the Flink version under test has the FLIP-558 ON CONFLICT
validation (>= 2.3). */
+ private static boolean isFlip558Enabled() {
+ return FlinkVersion.current().toString().compareTo("2.3") >= 0;
+ }
}
