This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 5e3b5a8b [FLINK-40121][Connectors/Kafka] Build and test against Flink
2.3 (#335)
5e3b5a8b is described below
commit 5e3b5a8b133b48118b3b22acc2d1f88c93b44486
Author: Martijn Visser <[email protected]>
AuthorDate: Fri Oct 23 21:56:24 2026 +0200
[FLINK-40121][Connectors/Kafka] Build and test against Flink 2.3 (#335)
---
.github/workflows/push_pr.yml | 2 +-
.github/workflows/weekly.yml | 3 +++
.../connector/kafka/sink/KafkaSinkITCase.java | 7 +-----
.../kafka/table/KafkaChangelogTableITCase.java | 6 +++--
.../connectors/kafka/table/KafkaTableITCase.java | 4 +++-
.../connectors/kafka/table/KafkaTableTestBase.java | 13 +++++++++++
.../kafka/table/UpsertKafkaTableITCase.java | 27 ++++++++++++++++------
7 files changed, 45 insertions(+), 17 deletions(-)
diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml
index 28e22c9e..218bf2be 100644
--- a/.github/workflows/push_pr.yml
+++ b/.github/workflows/push_pr.yml
@@ -42,7 +42,7 @@ jobs:
compile_and_test:
strategy:
matrix:
- flink: &flink_versions [ 2.2.1 ]
+ flink: &flink_versions [ 2.2.1, 2.3.0 ]
jdk: &jdk_versions [ '11, 17, 21' ]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
diff --git a/.github/workflows/weekly.yml b/.github/workflows/weekly.yml
index e02986e7..b9ed23be 100644
--- a/.github/workflows/weekly.yml
+++ b/.github/workflows/weekly.yml
@@ -34,6 +34,9 @@ jobs:
flink_branches: &flink_branches_matrix [{
flink: 2.2-SNAPSHOT,
branch: main
+ }, {
+ flink: 2.3-SNAPSHOT,
+ branch: main
}, {
flink: 2.2.1,
branch: v5.0,
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
index 6e382e06..7e20d8fd 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
@@ -52,7 +52,6 @@ import
org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SinkTestSuiteBase;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.minicluster.MiniCluster;
@@ -539,11 +538,7 @@ class KafkaSinkITCase {
clusterClient);
JobResult jobResult = clusterClient.requestJobResult(thirdJobId).get();
- assertThat(jobResult.getApplicationStatus())
- .isEqualTo(
- supportedMigration
- ? ApplicationStatus.SUCCEEDED
- : ApplicationStatus.FAILED);
+ assertThat(jobResult.isSuccess()).isEqualTo(supportedMigration);
if (supportedMigration) {
final List<Long> committedRecords =
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java
index d77e3dfb..97a4e8cf 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaChangelogTableITCase.java
@@ -251,7 +251,8 @@ class KafkaChangelogTableITCase extends KafkaTableTestBase {
"INSERT INTO sink "
+ "SELECT origin_topic, origin_database,
origin_table, origin_sql_type, "
+ "origin_pk_names, origin_ts, origin_es, name
"
- + "FROM canal_source");
+ + "FROM canal_source"
+ + onConflictDeduplicate());
/*
* Canal captures change data on the `products` table:
@@ -386,7 +387,8 @@ class KafkaChangelogTableITCase extends KafkaTableTestBase {
"INSERT INTO sink "
+ "SELECT origin_topic, origin_database,
origin_table, origin_primary_key_columns, "
+ "origin_ts, name "
- + "FROM maxwell_source");
+ + "FROM maxwell_source"
+ + onConflictDeduplicate());
/*
* Maxwell captures change data on the `products` table:
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
index cff654f8..965df1a9 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
@@ -155,7 +155,9 @@ class KafkaTableITCase extends KafkaTableTestBase {
"SELECT\n"
+ " CAST(TUMBLE_END(ts, INTERVAL '5' SECOND) AS
VARCHAR),\n"
+ " CAST(MAX(log_date) AS VARCHAR),\n"
- + " CAST(MAX(log_time) AS VARCHAR),\n"
+ // Flink 2.3 renders the precision of TIME
(FLINK-17224), TIME(0) renders
+ // the same on every supported version.
+ + " CAST(CAST(MAX(log_time) AS TIME(0)) AS
VARCHAR),\n"
+ " CAST(MAX(ts) AS VARCHAR),\n"
+ " COUNT(*),\n"
+ " CAST(MAX(price) AS DECIMAL(10, 2))\n"
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
index 32e76f68..32145589 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.connectors.kafka.table;
+import org.apache.flink.FlinkVersion;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.connector.kafka.testutils.KafkaUtil;
@@ -106,6 +107,18 @@ abstract class KafkaTableTestBase extends AbstractTestBase
{
cancelTimeoutLogger();
}
+ /**
+ * Since Flink 2.3 (FLIP-558), an insert whose upsert key differs from the
primary key of the
+ * sink fails unless it names a conflict strategy. A query without an
upsert key, such as an
+ * append-only insert, counts as differing. Flink 2.2 does not know the
clause. DEDUPLICATE
+ * keeps the behaviour of earlier versions.
+ */
+ protected static String onConflictDeduplicate() {
+ return FlinkVersion.current().isNewerVersionThan(FlinkVersion.v2_2)
+ ? " ON CONFLICT DO DEDUPLICATE"
+ : "";
+ }
+
public Properties getStandardProps() {
Properties standardProps = new Properties();
standardProps.put("bootstrap.servers",
KAFKA_CONTAINER.getBootstrapServers());
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaTableITCase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaTableITCase.java
index 7677f984..1e2d126e 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaTableITCase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaTableITCase.java
@@ -176,7 +176,11 @@ class UpsertKafkaTableITCase extends KafkaTableTestBase {
tEnv.executeSql(createTable);
- table.executeInsert("upsert_kafka").await();
+ tEnv.createTemporaryView("upsert_source", table);
+ tEnv.executeSql(
+ "INSERT INTO upsert_kafka SELECT * FROM upsert_source"
+ + onConflictDeduplicate())
+ .await();
final List<Row> result = collectRows(tEnv.sqlQuery("SELECT * FROM
upsert_kafka"), 3);
final List<Row> expected =
@@ -254,7 +258,11 @@ class UpsertKafkaTableITCase extends KafkaTableTestBase {
tEnv.executeSql(createTable);
- table.executeInsert("upsert_kafka").await();
+ tEnv.createTemporaryView("upsert_source", table);
+ tEnv.executeSql(
+ "INSERT INTO upsert_kafka SELECT * FROM upsert_source"
+ + onConflictDeduplicate())
+ .await();
final List<Row> result = collectRows(tEnv.sqlQuery("SELECT * FROM
upsert_kafka"), 3);
final List<Row> expected =
@@ -312,7 +320,8 @@ class UpsertKafkaTableITCase extends KafkaTableTestBase {
+ " (1, 'name 1', TIMESTAMP '2020-03-08 13:12:11.123',
100, 41, 'payload 1'),\n"
+ " (2, 'name 2', TIMESTAMP '2020-03-09 13:12:11.123',
101, 42, 'payload 2'),\n"
+ " (3, 'name 3', TIMESTAMP '2020-03-10 13:12:11.123',
102, 43, 'payload 3'),\n"
- + " (2, 'name 2', TIMESTAMP '2020-03-11 13:12:11.123',
101, 42, 'payload')";
+ + " (2, 'name 2', TIMESTAMP '2020-03-11 13:12:11.123',
101, 42, 'payload')"
+ + onConflictDeduplicate();
tEnv.executeSql(initialValues).await();
// ---------- Consume stream from Kafka -------------------
@@ -413,7 +422,8 @@ class UpsertKafkaTableITCase extends KafkaTableTestBase {
+ " (1, 'name 1', TIMESTAMP '2020-03-08 13:12:11.123',
100, 'payload 1'),\n"
+ " (2, 'name 2', TIMESTAMP '2020-03-09 13:12:11.123',
101, 'payload 2'),\n"
+ " (3, 'name 3', TIMESTAMP '2020-03-10 13:12:11.123',
102, 'payload 3'),\n"
- + " (1, 'name 1', TIMESTAMP '2020-03-11 13:12:11.123',
100, 'payload')";
+ + " (1, 'name 1', TIMESTAMP '2020-03-11 13:12:11.123',
100, 'payload')"
+ + onConflictDeduplicate();
tEnv.executeSql(initialValues).await();
// ---------- Consume stream from Kafka -------------------
@@ -502,7 +512,8 @@ class UpsertKafkaTableITCase extends KafkaTableTestBase {
+ " (1, 100, 'payload 1'),\n"
+ " (1, 100, 'payload 1-new'),\n"
+ " (2, 101, 'payload 2'),\n"
- + " (3, 102, 'payload 3')";
+ + " (3, 102, 'payload 3')"
+ + onConflictDeduplicate();
tEnv.executeSql(insertValuesSql).await();
// results should only have records up to offset=2
@@ -564,7 +575,8 @@ class UpsertKafkaTableITCase extends KafkaTableTestBase {
+ " (1, TIMESTAMP '2023-03-08 08:10:10.666', 100,
'payload 1'),\n"
+ " (2, TIMESTAMP '2023-03-09 13:12:11.123', 101,
'payload 2'),\n"
+ " (1, TIMESTAMP '2023-03-10 12:09:50.321', 100,
'payload 1-new'),\n"
- + " (2, TIMESTAMP '2023-03-11 17:15:13.457', 101,
'payload 2-new')";
+ + " (2, TIMESTAMP '2023-03-11 17:15:13.457', 101,
'payload 2-new')"
+ + onConflictDeduplicate();
tEnv.executeSql(insertValuesSql).await();
// results should only have records up to timestamp
2023-03-10T14:00:00.000
@@ -652,7 +664,8 @@ class UpsertKafkaTableITCase extends KafkaTableTestBase {
+ " (1, TIMESTAMP '2023-03-11 08:10:10.666', 100,
'payload 1'),\n"
+ " (2, TIMESTAMP '2023-03-12 13:12:11.123', 101,
'payload 2'),\n"
+ " (1, TIMESTAMP '2023-03-13 12:09:50.321', 100,
'payload 1-new'),\n"
- + " (2, TIMESTAMP '2023-03-14 17:15:13.457', 101,
'payload 2-new')";
+ + " (2, TIMESTAMP '2023-03-14 17:15:13.457', 101,
'payload 2-new')"
+ + onConflictDeduplicate();
tEnv.executeSql(insertValuesSql).await();
// results should be empty
