You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2024/10/10 09:54:11 UTC
(flink-connector-kafka) branch main updated (5eeafd6a -> aedc7905)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
from 5eeafd6a [FLINK-36441] Fix KafkaWriterITCase#testIncreasingRecordBasedCounters
new 429fe0c4 [FLINK-35109] Fix SmokeKafkaITCase for later java versions
new aedc7905 [FLINK-35109] Bump to Flink 1.19 and support Flink 1.20
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/push_pr.yml | 8 +-
.github/workflows/weekly.yml | 21 +-
.../flink/tests/util/kafka/SmokeKafkaITCase.java | 4 +
flink-connector-kafka/pom.xml | 12 +-
.../kafka/FlinkKafkaConsumerBaseMigrationTest.java | 2 +-
.../kafka/FlinkKafkaProducerMigrationTest.java | 2 +-
.../kafka/KafkaSerializerUpgradeTest.java | 2 +-
.../kafka/table/KafkaChangelogTableITCase.java | 4 -
...edMapper.java => ThreadContextClassLoader.java} | 22 +-
.../kafka/testutils/TypeSerializerMatchers.java | 225 +++++++++++++++++++++
.../testutils/TypeSerializerUpgradeTestBase.java | 21 +-
.../context-state-serializer-1.11/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.12/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.13/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.14/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.15/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.16/test-data | Bin 19 -> 19 bytes
.../serializer-snapshot | Bin
.../test-data | Bin
.../serializer-snapshot | Bin
.../test-data | Bin
.../serializer-snapshot | Bin
.../test-data | Bin
...-migration-test-flink1.17-empty-state-snapshot} | Bin 1465 -> 1465 bytes
...fka-consumer-migration-test-flink1.17-snapshot} | Bin 1519 -> 1519 bytes
...r-migration-test-flink1.18-empty-state-snapshot | Bin 0 -> 1466 bytes
...afka-consumer-migration-test-flink1.18-snapshot | Bin 0 -> 1520 bytes
...r-migration-test-flink1.19-empty-state-snapshot | Bin 0 -> 1466 bytes
...afka-consumer-migration-test-flink1.19-snapshot | Bin 0 -> 1520 bytes
...r-migration-test-flink1.20-empty-state-snapshot | Bin 0 -> 1466 bytes
...afka-consumer-migration-test-flink1.20-snapshot | Bin 0 -> 1520 bytes
...a-migration-kafka-producer-flink-1.17-snapshot} | Bin 1242 -> 1242 bytes
...ka-migration-kafka-producer-flink-1.18-snapshot | Bin 0 -> 1243 bytes
...ka-migration-kafka-producer-flink-1.19-snapshot | Bin 0 -> 1243 bytes
...ka-migration-kafka-producer-flink-1.20-snapshot | Bin 0 -> 1243 bytes
.../serializer-snapshot | Bin
.../test-data | Bin
.../serializer-snapshot | Bin
.../test-data | Bin
.../serializer-snapshot | Bin
.../test-data | Bin
pom.xml | 9 +-
42 files changed, 266 insertions(+), 66 deletions(-)
copy flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/{ThrottledMapper.java => ThreadContextClassLoader.java} (56%)
create mode 100644 flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerMatchers.java
copy flink-connector-kafka/src/test/resources/{context-state-serializer-1.11 => context-state-serializer-1.18}/serializer-snapshot (100%)
copy flink-connector-kafka/src/test/resources/{context-state-serializer-1.11 => context-state-serializer-1.18}/test-data (100%)
copy flink-connector-kafka/src/test/resources/{context-state-serializer-1.11 => context-state-serializer-1.19}/serializer-snapshot (100%)
copy flink-connector-kafka/src/test/resources/{context-state-serializer-1.11 => context-state-serializer-1.19}/test-data (100%)
copy flink-connector-kafka/src/test/resources/{context-state-serializer-1.11 => context-state-serializer-1.20}/serializer-snapshot (100%)
copy flink-connector-kafka/src/test/resources/{context-state-serializer-1.11 => context-state-serializer-1.20}/test-data (100%)
copy flink-connector-kafka/src/test/resources/{kafka-consumer-migration-test-flink1.13-empty-state-snapshot => kafka-consumer-migration-test-flink1.17-empty-state-snapshot} (88%)
copy flink-connector-kafka/src/test/resources/{kafka-consumer-migration-test-flink1.15-snapshot => kafka-consumer-migration-test-flink1.17-snapshot} (94%)
create mode 100644 flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.18-empty-state-snapshot
create mode 100644 flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.18-snapshot
create mode 100644 flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.19-empty-state-snapshot
create mode 100644 flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.19-snapshot
create mode 100644 flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.20-empty-state-snapshot
create mode 100644 flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.20-snapshot
copy flink-connector-kafka/src/test/resources/{kafka-migration-kafka-producer-flink-1.15-snapshot => kafka-migration-kafka-producer-flink-1.17-snapshot} (66%)
create mode 100644 flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.18-snapshot
create mode 100644 flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.19-snapshot
create mode 100644 flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.20-snapshot
copy flink-connector-kafka/src/test/resources/{transaction-state-serializer-1.11 => transaction-state-serializer-1.18}/serializer-snapshot (100%)
copy flink-connector-kafka/src/test/resources/{transaction-state-serializer-1.11 => transaction-state-serializer-1.18}/test-data (100%)
copy flink-connector-kafka/src/test/resources/{transaction-state-serializer-1.11 => transaction-state-serializer-1.19}/serializer-snapshot (100%)
copy flink-connector-kafka/src/test/resources/{transaction-state-serializer-1.11 => transaction-state-serializer-1.19}/test-data (100%)
copy flink-connector-kafka/src/test/resources/{transaction-state-serializer-1.11 => transaction-state-serializer-1.20}/serializer-snapshot (100%)
copy flink-connector-kafka/src/test/resources/{transaction-state-serializer-1.11 => transaction-state-serializer-1.20}/test-data (100%)
(flink-connector-kafka) 02/02: [FLINK-35109] Bump to Flink 1.19 and support Flink 1.20
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit aedc79051323ed3c75d54859cdaa5770b224f4ce
Author: Arvid Heise <AH...@users.noreply.github.com>
AuthorDate: Mon Sep 30 14:33:46 2024 +0200
[FLINK-35109] Bump to Flink 1.19 and support Flink 1.20
Also
- adds the migration support tests up to 1.20.
- bumps Kafka-client to 3.6.2
---
.github/workflows/push_pr.yml | 8 +-
.github/workflows/weekly.yml | 21 +-
flink-connector-kafka/pom.xml | 12 +-
.../kafka/FlinkKafkaConsumerBaseMigrationTest.java | 2 +-
.../kafka/FlinkKafkaProducerMigrationTest.java | 2 +-
.../kafka/KafkaSerializerUpgradeTest.java | 2 +-
.../kafka/table/KafkaChangelogTableITCase.java | 4 -
.../kafka/testutils/ThreadContextClassLoader.java | 41 ++++
.../kafka/testutils/TypeSerializerMatchers.java | 225 +++++++++++++++++++++
.../testutils/TypeSerializerUpgradeTestBase.java | 21 +-
.../context-state-serializer-1.11/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.12/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.13/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.14/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.15/test-data | Bin 19 -> 19 bytes
.../context-state-serializer-1.16/test-data | Bin 19 -> 19 bytes
.../serializer-snapshot | Bin 0 -> 126 bytes
.../test-data | Bin
.../serializer-snapshot | Bin 0 -> 126 bytes
.../test-data | Bin
.../serializer-snapshot | Bin 0 -> 126 bytes
.../test-data | Bin
...r-migration-test-flink1.17-empty-state-snapshot | Bin 0 -> 1465 bytes
...afka-consumer-migration-test-flink1.17-snapshot | Bin 0 -> 1519 bytes
...r-migration-test-flink1.18-empty-state-snapshot | Bin 0 -> 1466 bytes
...afka-consumer-migration-test-flink1.18-snapshot | Bin 0 -> 1520 bytes
...r-migration-test-flink1.19-empty-state-snapshot | Bin 0 -> 1466 bytes
...afka-consumer-migration-test-flink1.19-snapshot | Bin 0 -> 1520 bytes
...r-migration-test-flink1.20-empty-state-snapshot | Bin 0 -> 1466 bytes
...afka-consumer-migration-test-flink1.20-snapshot | Bin 0 -> 1520 bytes
...ka-migration-kafka-producer-flink-1.17-snapshot | Bin 0 -> 1242 bytes
...ka-migration-kafka-producer-flink-1.18-snapshot | Bin 0 -> 1243 bytes
...ka-migration-kafka-producer-flink-1.19-snapshot | Bin 0 -> 1243 bytes
...ka-migration-kafka-producer-flink-1.20-snapshot | Bin 0 -> 1243 bytes
.../serializer-snapshot | Bin 0 -> 134 bytes
.../transaction-state-serializer-1.18/test-data | Bin 0 -> 17 bytes
.../serializer-snapshot | Bin 0 -> 134 bytes
.../transaction-state-serializer-1.19/test-data | Bin 0 -> 17 bytes
.../serializer-snapshot | Bin 0 -> 134 bytes
.../transaction-state-serializer-1.20/test-data | Bin 0 -> 17 bytes
pom.xml | 9 +-
41 files changed, 293 insertions(+), 54 deletions(-)
diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml
index 95157940..fd102871 100644
--- a/.github/workflows/push_pr.yml
+++ b/.github/workflows/push_pr.yml
@@ -28,10 +28,10 @@ jobs:
compile_and_test:
strategy:
matrix:
- flink: [ 1.18.1 ]
- jdk: [ '8, 11, 17' ]
+ flink: [ 1.19.1 ]
+ jdk: [ '8, 11, 17, 21' ]
include:
- - flink: 1.19.0
+ - flink: 1.20.0
jdk: '8, 11, 17, 21'
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
@@ -40,7 +40,7 @@ jobs:
python_test:
strategy:
matrix:
- flink: [ 1.18.1, 1.19.0 ]
+ flink: [ 1.19.1, 1.20.0 ]
uses: apache/flink-connector-shared-utils/.github/workflows/python_ci.yml@ci_utils
with:
flink_version: ${{ matrix.flink }}
\ No newline at end of file
diff --git a/.github/workflows/weekly.yml b/.github/workflows/weekly.yml
index 1caecd5c..da4426df 100644
--- a/.github/workflows/weekly.yml
+++ b/.github/workflows/weekly.yml
@@ -30,34 +30,21 @@ jobs:
strategy:
matrix:
flink_branches: [{
- flink: 1.18-SNAPSHOT,
- branch: main
- }, {
flink: 1.19-SNAPSHOT,
- jdk: '8, 11, 17, 21',
branch: main
}, {
flink: 1.20-SNAPSHOT,
- jdk: '8, 11, 17, 21',
branch: main
}, {
- flink: 1.18.1,
+ flink: 1.19.1,
branch: v3.2
}, {
- flink: 1.19.0,
- branch: v3.2,
- jdk: '8, 11, 17, 21',
- }, {
- flink: 1.18.1,
- branch: v3.1
- }, {
- flink: 1.19.0,
- branch: v3.1,
- jdk: '8, 11, 17, 21',
+ flink: 1.20.0,
+ branch: v3.2
}]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
flink_version: ${{ matrix.flink_branches.flink }}
connector_branch: ${{ matrix.flink_branches.branch }}
- jdk_version: ${{ matrix.flink_branches.jdk || '8, 11, 17' }}
+ jdk_version: ${{ matrix.flink_branches.jdk || '8, 11, 17, 21' }}
run_dependency_convergence: false
\ No newline at end of file
diff --git a/flink-connector-kafka/pom.xml b/flink-connector-kafka/pom.xml
index 4a10bdc9..85d43741 100644
--- a/flink-connector-kafka/pom.xml
+++ b/flink-connector-kafka/pom.xml
@@ -39,7 +39,8 @@ under the License.
FlinkKafkaProducerBaseTest --> --add-opens=java.base/java.lang=ALL-UNNAMED <!--
FlinkKafkaProducerBaseTest --> --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED <!--
FlinkKafkaConsumerBaseTest --> --add-opens=java.base/java.util=ALL-UNNAMED <!--
- KafkaProducerExactlyOnceITCase --> --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED </flink.connector.module.config>
+ KafkaProducerExactlyOnceITCase --> --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED
+ </flink.connector.module.config>
</properties>
<dependencies>
@@ -81,10 +82,10 @@ under the License.
<version>${kafka.version}</version>
</dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
<!-- Tests -->
@@ -211,6 +212,7 @@ under the License.
<version>${flink.version}</version>
<scope>test</scope>
</dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
index 296545ca..47bce8bd 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
@@ -92,7 +92,7 @@ public class FlinkKafkaConsumerBaseMigrationTest {
@Parameterized.Parameters(name = "Migration Savepoint: {0}")
public static Collection<FlinkVersion> parameters() {
- return FlinkVersion.rangeOf(FlinkVersion.v1_8, FlinkVersion.v1_16);
+ return FlinkVersion.rangeOf(FlinkVersion.v1_8, FlinkVersion.current());
}
public FlinkKafkaConsumerBaseMigrationTest(FlinkVersion testMigrateVersion) {
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationTest.java
index 8a413f42..32c380c7 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationTest.java
@@ -42,7 +42,7 @@ import java.util.Properties;
public class FlinkKafkaProducerMigrationTest extends KafkaMigrationTestBase {
@Parameterized.Parameters(name = "Migration Savepoint: {0}")
public static Collection<FlinkVersion> parameters() {
- return FlinkVersion.rangeOf(FlinkVersion.v1_8, FlinkVersion.v1_16);
+ return FlinkVersion.rangeOf(FlinkVersion.v1_8, FlinkVersion.current());
}
public FlinkKafkaProducerMigrationTest(FlinkVersion testMigrateVersion) {
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializerUpgradeTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializerUpgradeTest.java
index c9e82be5..90ce2e5e 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializerUpgradeTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializerUpgradeTest.java
@@ -20,9 +20,9 @@ package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.FlinkVersion;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerMatchers;
import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.streaming.connectors.kafka.internals.FlinkKafkaInternalProducer;
+import org.apache.flink.streaming.connectors.kafka.testutils.TypeSerializerMatchers;
import org.apache.flink.streaming.connectors.kafka.testutils.TypeSerializerUpgradeTestBase;
import org.hamcrest.Matcher;
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 e8bc9e37..632b74ac 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
@@ -29,7 +29,6 @@ import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartiti
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.junit.Before;
@@ -65,7 +64,6 @@ public class KafkaChangelogTableITCase extends KafkaTableTestBase {
tableConf.set(
ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY, Duration.ofSeconds(1));
tableConf.set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, 5000L);
- tableConf.set(OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY, "TWO_PHASE");
// ---------- Write the Debezium json into Kafka -------------------
List<String> lines = readLines("debezium-data-schema-exclude.txt");
@@ -194,7 +192,6 @@ public class KafkaChangelogTableITCase extends KafkaTableTestBase {
tableConf.set(
ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY, Duration.ofSeconds(1));
tableConf.set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, 5000L);
- tableConf.set(OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY, "TWO_PHASE");
// ---------- Write the Canal json into Kafka -------------------
List<String> lines = readLines("canal-data.txt");
@@ -335,7 +332,6 @@ public class KafkaChangelogTableITCase extends KafkaTableTestBase {
tableConf.set(
ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY, Duration.ofSeconds(1));
tableConf.set(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, 5000L);
- tableConf.set(OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY, "TWO_PHASE");
// ---------- Write the Maxwell json into Kafka -------------------
List<String> lines = readLines("maxwell-data.txt");
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/ThreadContextClassLoader.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/ThreadContextClassLoader.java
new file mode 100644
index 00000000..cdaec31c
--- /dev/null
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/ThreadContextClassLoader.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.kafka.testutils;
+
+import java.io.Closeable;
+
+/**
+ * Utility class to temporarily use a different classloader as the thread context classloader.
+ *
+ * <p>Temporarily copied from flink-core to avoid dependency on flink-core.
+ */
+public class ThreadContextClassLoader implements Closeable {
+
+ private final ClassLoader originalThreadContextClassLoader;
+
+ public ThreadContextClassLoader(ClassLoader newThreadContextClassLoader) {
+ this.originalThreadContextClassLoader = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(newThreadContextClassLoader);
+ }
+
+ @Override
+ public void close() {
+ Thread.currentThread().setContextClassLoader(originalThreadContextClassLoader);
+ }
+}
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerMatchers.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerMatchers.java
new file mode 100644
index 00000000..0815e299
--- /dev/null
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerMatchers.java
@@ -0,0 +1,225 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.kafka.testutils;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+
+import org.hamcrest.CoreMatchers;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.TypeSafeDiagnosingMatcher;
+import org.hamcrest.TypeSafeMatcher;
+
+import java.util.function.Predicate;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A Collection of useful {@link Matcher}s for {@link TypeSerializer} and {@link
+ * TypeSerializerSchemaCompatibility}.
+ *
+ * <p>Note copied from Flink 1.19. Remove when we drop 1.19 support.
+ */
+public final class TypeSerializerMatchers {
+
+ private TypeSerializerMatchers() {}
+
+ // -------------------------------------------------------------------------------------------------------------
+ // Matcher Factories
+ // -------------------------------------------------------------------------------------------------------------
+
+ /**
+ * Matches {@code compatibleAsIs} {@link TypeSerializerSchemaCompatibility}.
+ *
+ * @param <T> element type
+ * @return a {@code Matcher} that matches {@code compatibleAsIs} {@link
+ * TypeSerializerSchemaCompatibility}.
+ */
+ public static <T> Matcher<TypeSerializerSchemaCompatibility<T>> isCompatibleAsIs() {
+ return propertyMatcher(
+ TypeSerializerSchemaCompatibility::isCompatibleAsIs,
+ "type serializer schema that is a compatible as is");
+ }
+
+ /**
+ * Matches {@code isIncompatible} {@link TypeSerializerSchemaCompatibility}.
+ *
+ * @param <T> element type
+ * @return a {@code Matcher} that matches {@code isIncompatible} {@link
+ * TypeSerializerSchemaCompatibility}.
+ */
+ public static <T> Matcher<TypeSerializerSchemaCompatibility<T>> isIncompatible() {
+ return propertyMatcher(
+ TypeSerializerSchemaCompatibility::isIncompatible,
+ "type serializer schema that is incompatible");
+ }
+
+ /**
+ * Matches {@code isCompatibleAfterMigration} {@link TypeSerializerSchemaCompatibility}.
+ *
+ * @param <T> element type
+ * @return a {@code Matcher} that matches {@code isCompatibleAfterMigration} {@link
+ * TypeSerializerSchemaCompatibility}.
+ */
+ public static <T> Matcher<TypeSerializerSchemaCompatibility<T>> isCompatibleAfterMigration() {
+ return propertyMatcher(
+ TypeSerializerSchemaCompatibility::isCompatibleAfterMigration,
+ "type serializer schema that is compatible after migration");
+ }
+
+ /**
+ * Matches {@code isCompatibleWithReconfiguredSerializer} {@link
+ * TypeSerializerSchemaCompatibility}.
+ *
+ * @param <T> element type
+ * @return a {@code Matcher} that matches {@code isCompatibleWithReconfiguredSerializer} {@link
+ * TypeSerializerSchemaCompatibility}.
+ */
+ public static <T>
+ Matcher<TypeSerializerSchemaCompatibility<T>> isCompatibleWithReconfiguredSerializer() {
+ @SuppressWarnings("unchecked")
+ Matcher<TypeSerializer<T>> anything =
+ (Matcher<TypeSerializer<T>>) (Matcher<?>) CoreMatchers.anything();
+
+ return new CompatibleAfterReconfiguration<>(anything);
+ }
+
+ /**
+ * Matches {@code isCompatibleWithReconfiguredSerializer} {@link
+ * TypeSerializerSchemaCompatibility}.
+ *
+ * @param reconfiguredSerializerMatcher matches the reconfigured serializer.
+ * @param <T> element type
+ * @return a {@code Matcher} that matches {@code isCompatibleWithReconfiguredSerializer} {@link
+ * TypeSerializerSchemaCompatibility}.
+ */
+ public static <T>
+ Matcher<TypeSerializerSchemaCompatibility<T>> isCompatibleWithReconfiguredSerializer(
+ Matcher<? extends TypeSerializer<T>> reconfiguredSerializerMatcher) {
+
+ return new CompatibleAfterReconfiguration<>(reconfiguredSerializerMatcher);
+ }
+
+ /**
+ * Matches if the expected {@code TypeSerializerSchemaCompatibility} has the same compatibility
+ * as {@code expectedCompatibility}.
+ *
+ * @param expectedCompatibility the compatibility to match to.
+ * @param <T> element type.
+ * @return a {@code Matcher} that matches if it has the same compatibility as {@code
+ * expectedCompatibility}.
+ */
+ public static <T> Matcher<TypeSerializerSchemaCompatibility<T>> hasSameCompatibilityAs(
+ TypeSerializerSchemaCompatibility<T> expectedCompatibility) {
+
+ return new SchemaCompatibilitySameAs<>(expectedCompatibility);
+ }
+
+ // -------------------------------------------------------------------------------------------------------------
+ // Helpers
+ // -------------------------------------------------------------------------------------------------------------
+
+ private static <T> Matcher<T> propertyMatcher(
+ Predicate<T> predicate, String matcherDescription) {
+ return new TypeSafeMatcher<T>() {
+
+ @Override
+ protected boolean matchesSafely(T item) {
+ return predicate.test(item);
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText(matcherDescription);
+ }
+ };
+ }
+
+ // -------------------------------------------------------------------------------------------------------------
+ // Matchers
+ // -------------------------------------------------------------------------------------------------------------
+
+ private static final class CompatibleAfterReconfiguration<T>
+ extends TypeSafeDiagnosingMatcher<TypeSerializerSchemaCompatibility<T>> {
+
+ private final Matcher<? extends TypeSerializer<T>> reconfiguredSerializerMatcher;
+
+ private CompatibleAfterReconfiguration(
+ Matcher<? extends TypeSerializer<T>> reconfiguredSerializerMatcher) {
+ this.reconfiguredSerializerMatcher = checkNotNull(reconfiguredSerializerMatcher);
+ }
+
+ @Override
+ protected boolean matchesSafely(
+ TypeSerializerSchemaCompatibility<T> item, Description mismatchDescription) {
+ if (!item.isCompatibleWithReconfiguredSerializer()) {
+ mismatchDescription.appendText(
+ "serializer schema is not compatible with a reconfigured serializer");
+ return false;
+ }
+ TypeSerializer<T> reconfiguredSerializer = item.getReconfiguredSerializer();
+ if (!reconfiguredSerializerMatcher.matches(reconfiguredSerializer)) {
+ reconfiguredSerializerMatcher.describeMismatch(
+ reconfiguredSerializer, mismatchDescription);
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description
+ .appendText("type serializer schema that is compatible after reconfiguration,")
+ .appendText("with a reconfigured serializer matching ")
+ .appendDescriptionOf(reconfiguredSerializerMatcher);
+ }
+ }
+
+ private static class SchemaCompatibilitySameAs<T>
+ extends TypeSafeMatcher<TypeSerializerSchemaCompatibility<T>> {
+
+ private final TypeSerializerSchemaCompatibility<T> expectedCompatibility;
+
+ private SchemaCompatibilitySameAs(
+ TypeSerializerSchemaCompatibility<T> expectedCompatibility) {
+ this.expectedCompatibility = checkNotNull(expectedCompatibility);
+ }
+
+ @Override
+ protected boolean matchesSafely(
+ TypeSerializerSchemaCompatibility<T> testResultCompatibility) {
+ if (expectedCompatibility.isCompatibleAsIs()) {
+ return testResultCompatibility.isCompatibleAsIs();
+ } else if (expectedCompatibility.isIncompatible()) {
+ return testResultCompatibility.isIncompatible();
+ } else if (expectedCompatibility.isCompatibleAfterMigration()) {
+ return testResultCompatibility.isCompatibleAfterMigration();
+ } else if (expectedCompatibility.isCompatibleWithReconfiguredSerializer()) {
+ return testResultCompatibility.isCompatibleWithReconfiguredSerializer();
+ }
+ return false;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("same compatibility as ").appendValue(expectedCompatibility);
+ }
+ }
+}
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerUpgradeTestBase.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerUpgradeTestBase.java
index 90e21838..ab3baade 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerUpgradeTestBase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerUpgradeTestBase.java
@@ -20,9 +20,7 @@ package org.apache.flink.streaming.connectors.kafka.testutils;
import org.apache.flink.FlinkVersion;
import org.apache.flink.api.common.typeutils.ClassRelocator;
-import org.apache.flink.api.common.typeutils.ThreadContextClassLoader;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerMatchers;
import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil;
@@ -61,7 +59,7 @@ import static org.hamcrest.CoreMatchers.not;
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public abstract class TypeSerializerUpgradeTestBase<PreviousElementT, UpgradedElementT> {
- public static final FlinkVersion CURRENT_VERSION = FlinkVersion.v1_17;
+ public static final FlinkVersion CURRENT_VERSION = FlinkVersion.current();
public static final Set<FlinkVersion> MIGRATION_VERSIONS =
FlinkVersion.rangeOf(FlinkVersion.v1_11, CURRENT_VERSION);
@@ -136,9 +134,6 @@ public abstract class TypeSerializerUpgradeTestBase<PreviousElementT, UpgradedEl
try (ThreadContextClassLoader ignored =
new ThreadContextClassLoader(setupClassloader)) {
return delegateSetup.createPriorSerializer();
- } catch (IOException e) {
- throw new RuntimeException(
- "Error creating prior serializer via ClassLoaderSafePreUpgradeSetup.", e);
}
}
@@ -147,9 +142,6 @@ public abstract class TypeSerializerUpgradeTestBase<PreviousElementT, UpgradedEl
try (ThreadContextClassLoader ignored =
new ThreadContextClassLoader(setupClassloader)) {
return delegateSetup.createTestData();
- } catch (IOException e) {
- throw new RuntimeException(
- "Error creating test data via ThreadContextClassLoader.", e);
}
}
}
@@ -179,10 +171,6 @@ public abstract class TypeSerializerUpgradeTestBase<PreviousElementT, UpgradedEl
try (ThreadContextClassLoader ignored =
new ThreadContextClassLoader(verifierClassloader)) {
return delegateVerifier.createUpgradedSerializer();
- } catch (IOException e) {
- throw new RuntimeException(
- "Error creating upgraded serializer via ClassLoaderSafeUpgradeVerifier.",
- e);
}
}
@@ -191,9 +179,6 @@ public abstract class TypeSerializerUpgradeTestBase<PreviousElementT, UpgradedEl
try (ThreadContextClassLoader ignored =
new ThreadContextClassLoader(verifierClassloader)) {
return delegateVerifier.testDataMatcher();
- } catch (IOException e) {
- throw new RuntimeException(
- "Error creating expected test data via ClassLoaderSafeUpgradeVerifier.", e);
}
}
@@ -203,10 +188,6 @@ public abstract class TypeSerializerUpgradeTestBase<PreviousElementT, UpgradedEl
try (ThreadContextClassLoader ignored =
new ThreadContextClassLoader(verifierClassloader)) {
return delegateVerifier.schemaCompatibilityMatcher(version);
- } catch (IOException e) {
- throw new RuntimeException(
- "Error creating schema compatibility matcher via ClassLoaderSafeUpgradeVerifier.",
- e);
}
}
}
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data
index 3efe488b..d13f52b8 100644
Binary files a/flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.12/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.12/test-data
index 3efe488b..d13f52b8 100644
Binary files a/flink-connector-kafka/src/test/resources/context-state-serializer-1.12/test-data and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.12/test-data differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.13/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.13/test-data
index 3efe488b..d13f52b8 100644
Binary files a/flink-connector-kafka/src/test/resources/context-state-serializer-1.13/test-data and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.13/test-data differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.14/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.14/test-data
index 3efe488b..d13f52b8 100644
Binary files a/flink-connector-kafka/src/test/resources/context-state-serializer-1.14/test-data and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.14/test-data differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.15/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.15/test-data
index 3efe488b..d13f52b8 100644
Binary files a/flink-connector-kafka/src/test/resources/context-state-serializer-1.15/test-data and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.15/test-data differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.16/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.16/test-data
index 3efe488b..d13f52b8 100644
Binary files a/flink-connector-kafka/src/test/resources/context-state-serializer-1.16/test-data and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.16/test-data differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.18/serializer-snapshot b/flink-connector-kafka/src/test/resources/context-state-serializer-1.18/serializer-snapshot
new file mode 100644
index 00000000..85a71dc6
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.18/serializer-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.18/test-data
similarity index 100%
copy from flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data
copy to flink-connector-kafka/src/test/resources/context-state-serializer-1.18/test-data
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.19/serializer-snapshot b/flink-connector-kafka/src/test/resources/context-state-serializer-1.19/serializer-snapshot
new file mode 100644
index 00000000..85a71dc6
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.19/serializer-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.19/test-data
similarity index 100%
copy from flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data
copy to flink-connector-kafka/src/test/resources/context-state-serializer-1.19/test-data
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.20/serializer-snapshot b/flink-connector-kafka/src/test/resources/context-state-serializer-1.20/serializer-snapshot
new file mode 100644
index 00000000..85a71dc6
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/context-state-serializer-1.20/serializer-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data b/flink-connector-kafka/src/test/resources/context-state-serializer-1.20/test-data
similarity index 100%
copy from flink-connector-kafka/src/test/resources/context-state-serializer-1.11/test-data
copy to flink-connector-kafka/src/test/resources/context-state-serializer-1.20/test-data
diff --git a/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.17-empty-state-snapshot b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.17-empty-state-snapshot
new file mode 100644
index 00000000..79ca3887
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.17-empty-state-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.17-snapshot b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.17-snapshot
new file mode 100644
index 00000000..08f1f7da
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.17-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.18-empty-state-snapshot b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.18-empty-state-snapshot
new file mode 100644
index 00000000..0fc1a555
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.18-empty-state-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.18-snapshot b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.18-snapshot
new file mode 100644
index 00000000..36381498
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.18-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.19-empty-state-snapshot b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.19-empty-state-snapshot
new file mode 100644
index 00000000..d9bd99d5
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.19-empty-state-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.19-snapshot b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.19-snapshot
new file mode 100644
index 00000000..ca2179ca
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.19-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.20-empty-state-snapshot b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.20-empty-state-snapshot
new file mode 100644
index 00000000..78ba1578
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.20-empty-state-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.20-snapshot b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.20-snapshot
new file mode 100644
index 00000000..995b70e5
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-consumer-migration-test-flink1.20-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.17-snapshot b/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.17-snapshot
new file mode 100644
index 00000000..6d5d487d
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.17-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.18-snapshot b/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.18-snapshot
new file mode 100644
index 00000000..68c9d632
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.18-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.19-snapshot b/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.19-snapshot
new file mode 100644
index 00000000..03b1a24a
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.19-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.20-snapshot b/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.20-snapshot
new file mode 100644
index 00000000..a6a147ca
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/kafka-migration-kafka-producer-flink-1.20-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.18/serializer-snapshot b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.18/serializer-snapshot
new file mode 100644
index 00000000..b95eacb6
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.18/serializer-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.18/test-data b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.18/test-data
new file mode 100644
index 00000000..0936509e
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.18/test-data differ
diff --git a/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.19/serializer-snapshot b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.19/serializer-snapshot
new file mode 100644
index 00000000..b95eacb6
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.19/serializer-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.19/test-data b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.19/test-data
new file mode 100644
index 00000000..0936509e
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.19/test-data differ
diff --git a/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.20/serializer-snapshot b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.20/serializer-snapshot
new file mode 100644
index 00000000..b95eacb6
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.20/serializer-snapshot differ
diff --git a/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.20/test-data b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.20/test-data
new file mode 100644
index 00000000..0936509e
Binary files /dev/null and b/flink-connector-kafka/src/test/resources/transaction-state-serializer-1.20/test-data differ
diff --git a/pom.xml b/pom.xml
index aeb1a4ec..8a8575be 100644
--- a/pom.xml
+++ b/pom.xml
@@ -50,7 +50,7 @@ under the License.
</modules>
<properties>
- <flink.version>1.18.0</flink.version>
+ <flink.version>1.19.1</flink.version>
<kafka.version>3.4.0</kafka.version>
<confluent.version>7.4.4</confluent.version>
@@ -403,6 +403,13 @@ under the License.
<version>2.1</version>
</dependency>
+ <!-- For dependency convergence -->
+ <dependency>
+ <groupId>org.yaml</groupId>
+ <artifactId>snakeyaml</artifactId>
+ <version>2.2</version>
+ </dependency>
+
<!-- For dependency convergence -->
<dependency>
<groupId>org.apache.commons</groupId>
(flink-connector-kafka) 01/02: [FLINK-35109] Fix SmokeKafkaITCase for later java versions
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 429fe0c48e4485ab12836ebfa1643c31e62e97c2
Author: Arvid Heise <AH...@users.noreply.github.com>
AuthorDate: Wed Oct 9 14:25:35 2024 +0200
[FLINK-35109] Fix SmokeKafkaITCase for later java versions
Workaround for FLINK-36454.
---
.../test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java b/flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java
index eef63e57..a8c416b0 100644
--- a/flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java
+++ b/flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java
@@ -101,6 +101,10 @@ class SmokeKafkaITCase {
final Configuration flinkConfig = new Configuration();
flinkConfig.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 3);
flinkConfig.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ // Workaround for FLINK-36454 ; default config is entirely overwritten
+ flinkConfig.setString(
+ "env.java.opts.all",
+ "--add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=AL [...]
return flinkConfig;
}