From a66b8d4978885382c327d0fec2684284b03e6f99 Mon Sep 17 00:00:00 2001
From: Zhenqiu Huang <huangzhenqiu0825@gmail.com>
Date: Sat, 10 Aug 2024 19:35:31 -0700
Subject: [PATCH] [FLINK-34468] bump flink version to 1.20.0

---
 .gitignore                                          |  1 +
 .../cassandra/source/CassandraSourceITCase.java     |  6 ------
 .../cassandra/CassandraConnectorITCase.java         | 13 +------------
 pom.xml                                             |  2 +-
 4 files changed, 3 insertions(+), 19 deletions(-)

diff --git a/.gitignore b/.gitignore
index acbe2176..1fa6f3af 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,5 +1,6 @@
 .eslintcache
 .cache
+.java-version
 scalastyle-output.xml
 .classpath
 .idea/*
diff --git a/flink-connector-cassandra/src/test/java/org/apache/flink/connector/cassandra/source/CassandraSourceITCase.java b/flink-connector-cassandra/src/test/java/org/apache/flink/connector/cassandra/source/CassandraSourceITCase.java
index d6eecb17..e203fa8f 100644
--- a/flink-connector-cassandra/src/test/java/org/apache/flink/connector/cassandra/source/CassandraSourceITCase.java
+++ b/flink-connector-cassandra/src/test/java/org/apache/flink/connector/cassandra/source/CassandraSourceITCase.java
@@ -176,7 +176,6 @@ public void testGenerateSplitsWithTooHighMaximumSplitSize(
     }
 
     // overridden to use unordered checks
-    @Override
     protected void checkResultWithSemantic(
             CloseableIterator<Pojo> resultIterator,
             List<List<Pojo>> testData,
@@ -197,7 +196,6 @@ protected void checkResultWithSemantic(
     }
 
     @Disabled("Not a unbounded source")
-    @Override
     public void testSourceMetrics(
             TestEnvironment testEnv,
             DataStreamSourceExternalContext<Pojo> externalContext,
@@ -205,28 +203,24 @@ public void testSourceMetrics(
             throws Exception {}
 
     @Disabled("Not a unbounded source")
-    @Override
     public void testSavepoint(
             TestEnvironment testEnv,
             DataStreamSourceExternalContext<Pojo> externalContext,
             CheckpointingMode semantic) {}
 
     @Disabled("Not a unbounded source")
-    @Override
     public void testScaleUp(
             TestEnvironment testEnv,
             DataStreamSourceExternalContext<Pojo> externalContext,
             CheckpointingMode semantic) {}
 
     @Disabled("Not a unbounded source")
-    @Override
     public void testScaleDown(
             TestEnvironment testEnv,
             DataStreamSourceExternalContext<Pojo> externalContext,
             CheckpointingMode semantic) {}
 
     @Disabled("Not a unbounded source")
-    @Override
     public void testTaskManagerFailure(
             TestEnvironment testEnv,
             DataStreamSourceExternalContext<Pojo> externalContext,
diff --git a/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java b/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
index f12e595d..6f9dcd77 100644
--- a/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
+++ b/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
@@ -42,7 +42,6 @@
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
-import org.apache.flink.streaming.runtime.operators.WriteAheadSinkTestBase;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.table.api.internal.TableEnvironmentInternal;
 import org.apache.flink.testutils.junit.extensions.retry.RetryExtension;
@@ -80,10 +79,7 @@
 @SuppressWarnings("serial")
 @Testcontainers
 @ExtendWith(RetryExtension.class)
-class CassandraConnectorITCase
-        extends WriteAheadSinkTestBase<
-                Tuple3<String, Integer, Integer>,
-                CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>>> {
+class CassandraConnectorITCase {
 
     private static final CassandraTestEnvironment cassandraTestEnvironment =
             new CassandraTestEnvironment(false);
@@ -284,7 +280,6 @@ void testAnnotatePojoWithTable() {
     //  Exactly-once Tests
     // ------------------------------------------------------------------------
 
-    @Override
     protected CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>> createSink()
             throws Exception {
         return new CassandraTupleWriteAheadSink<>(
@@ -295,17 +290,14 @@ protected CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>> createS
                 new CassandraCommitter(cassandraTestEnvironment.getBuilderForReading()));
     }
 
-    @Override
     protected TupleTypeInfo<Tuple3<String, Integer, Integer>> createTypeInfo() {
         return TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class, Integer.class);
     }
 
-    @Override
     protected Tuple3<String, Integer, Integer> generateValue(int counter, int checkpointID) {
         return new Tuple3<>(UUID.randomUUID().toString(), counter, checkpointID);
     }
 
-    @Override
     protected void verifyResultsIdealCircumstances(
             CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>> sink) {
 
@@ -325,7 +317,6 @@ protected void verifyResultsIdealCircumstances(
                 .isEmpty();
     }
 
-    @Override
     protected void verifyResultsDataPersistenceUponMissedNotify(
             CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>> sink) {
 
@@ -345,7 +336,6 @@ protected void verifyResultsDataPersistenceUponMissedNotify(
                 .isEmpty();
     }
 
-    @Override
     protected void verifyResultsDataDiscardingUponRestore(
             CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>> sink) {
 
@@ -368,7 +358,6 @@ protected void verifyResultsDataDiscardingUponRestore(
                 .isEmpty();
     }
 
-    @Override
     protected void verifyResultsWhenReScaling(
             CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>> sink,
             int startElementCounter,
diff --git a/pom.xml b/pom.xml
index 858bc049..73308742 100644
--- a/pom.xml
+++ b/pom.xml
@@ -42,7 +42,7 @@ under the License.
 	</scm>
 
 	<properties>
-		<flink.version>1.18.0</flink.version>
+		<flink.version>1.20.0</flink.version>
 		<japicmp.referenceVersion>3.1.0-1.17</japicmp.referenceVersion>
 		<guava.version>19.0</guava.version>
 	</properties>