[FLINK-36586][build] Update flink version to 1.19 (#3660)

Co-authored-by: ConradJam <czy006@apache.com>
Co-authored-by: Hang Ruan <ruanhang1993@hotmail.com>

Author: 3 people

.github/workflows/flink_cdc.yml

@@ -71,7 +71,7 @@ jobs:
     strategy:
       matrix:
         java-version: [ '8' ]
-        flink-version: ['1.17.2', '1.18.1', '1.19.1', '1.20.0']
+        flink-version: ['1.19.1', '1.20.0']
         module: [ 'pipeline_e2e' ]
     name: Pipeline E2E Tests
     uses: ./.github/workflows/flink_cdc_base.yml
@@ -83,7 +83,7 @@ jobs:
     strategy:
       matrix:
         java-version: [ '8' ]
-        flink-version: ['1.16.3', '1.17.2', '1.18.1', '1.19.1', '1.20.0']
+        flink-version: ['1.19.1', '1.20.0']
         module: [ 'source_e2e' ]
     name: Source E2E Tests
     uses: ./.github/workflows/flink_cdc_base.yml

.github/workflows/flink_cdc_migration_test.yml

@@ -39,7 +39,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        flink-version: [ '1.18.1', '1.19.1', '1.20.0' ]
+        flink-version: [ '1.19.1', '1.20.0' ]
 
     steps:
       - uses: actions/checkout@v4
@@ -78,7 +78,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        flink-version: [ '1.18.1', '1.19.1', '1.20.0' ]
+        flink-version: [ '1.19.1', '1.20.0' ]
 
     steps:
       - uses: actions/checkout@v4

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/pom.xml

@@ -36,7 +36,7 @@ limitations under the License.
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <elasticsearch.version>8.12.1</elasticsearch.version>
-        <flink.version>1.18.0</flink.version>
+        <flink.version>1.19.1</flink.version>
         <scala.binary.version>4.0</scala.binary.version>
         <jackson.version>2.13.2</jackson.version>
         <surefire.module.config>--add-opens=java.base/java.util=ALL-UNNAMED</surefire.module.config>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/pom.xml

@@ -35,7 +35,7 @@ limitations under the License.
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-kafka</artifactId>
-            <version>3.0.2-${flink.major.version}</version>
+            <version>3.3.0-${flink.major.version}</version>
         </dependency>
 
         <dependency>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml

@@ -72,6 +72,13 @@ limitations under the License.
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-compress</artifactId>
+            <version>1.26.0</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-inline</artifactId>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java

@@ -18,6 +18,8 @@
 package org.apache.flink.cdc.connectors.paimon.sink.v2;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.TaskInfo;
 import org.apache.flink.api.common.operators.MailboxExecutor;
 import org.apache.flink.api.common.operators.ProcessingTimeService;
 import org.apache.flink.api.common.serialization.SerializationSchema;
@@ -40,6 +42,8 @@
 import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.metrics.groups.InternalSinkCommitterMetricGroup;
 import org.apache.flink.streaming.runtime.operators.sink.committables.CommitRequestImpl;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableEnvironment;
@@ -572,7 +576,10 @@ private MultiTableCommittable correctCheckpointId(MultiTableCommittable committa
     private static class MockCommitRequestImpl<CommT> extends CommitRequestImpl<CommT> {
 
         protected MockCommitRequestImpl(CommT committable) {
-            super(committable);
+            super(
+                    committable,
+                    InternalSinkCommitterMetricGroup.wrap(
+                            UnregisteredMetricsGroup.createOperatorMetricGroup()));
         }
     }
 
@@ -633,5 +640,15 @@ public <IN> TypeSerializer<IN> createInputSerializer() {
         public JobID getJobId() {
             return null;
         }
+
+        @Override
+        public JobInfo getJobInfo() {
+            return null;
+        }
+
+        @Override
+        public TaskInfo getTaskInfo() {
+            return null;
+        }
     }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/pom.xml

@@ -28,7 +28,7 @@ limitations under the License.
     <artifactId>flink-cdc-pipeline-connector-starrocks</artifactId>
 
     <properties>
-        <starrocks.connector.version>1.2.9_flink-${flink.major.version}</starrocks.connector.version>
+        <starrocks.connector.version>1.2.10_flink-${flink.major.version}</starrocks.connector.version>
     </properties>
 
     <dependencies>
@@ -38,15 +38,6 @@ limitations under the License.
             <version>${starrocks.connector.version}</version>
         </dependency>
 
-        <dependency>
-            <!-- TODO connector 1.2.9 depends on this, but not package it, so add this dependency here.
-            This dependency can be removed after upgrading connector to 1.2.10 which will not use
-            commons-compress anymore. -->
-            <groupId>org.apache.commons</groupId>
-            <artifactId>commons-compress</artifactId>
-            <version>1.21</version>
-        </dependency>
-
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-cdc-composer</artifactId>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/com/starrocks/shade/org/apache/commons/compress/utils/Lists.java

@@ -18,6 +18,8 @@
 package org.apache.flink.cdc.connectors.starrocks.sink;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.TaskInfo;
 import org.apache.flink.api.common.operators.MailboxExecutor;
 import org.apache.flink.api.common.operators.ProcessingTimeService;
 import org.apache.flink.api.common.serialization.SerializationSchema;
@@ -353,5 +355,15 @@ public <IN> TypeSerializer<IN> createInputSerializer() {
         public JobID getJobId() {
             throw new UnsupportedOperationException();
         }
+
+        @Override
+        public JobInfo getJobInfo() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public TaskInfo getTaskInfo() {
+            throw new UnsupportedOperationException();
+        }
     }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/EventRecordSerializationSchemaTest.java

@@ -209,7 +209,7 @@ public void testConsumingAllEvents()
                     "spare tire,22.200"
                 };
 
-        List<String> actual = TestValuesTableFactory.getResults("sink");
+        List<String> actual = TestValuesTableFactory.getResultsAsStrings("sink");
         assertThat(actual, containsInAnyOrder(expected));
 
         cancelJobIfRunning(result);
@@ -302,7 +302,7 @@ public void testAllTypes() throws Exception {
                 Arrays.asList(
                         "+I(1,32767,65535,2147483647,5.5,6.6,123.12345,404.4,Hello World,a,abc,2020-07-17T18:00:22.123,2020-07-17,18:00:22,500,2020-07-17T18:00:22.123456789)",
                         "+U(1,0,65535,2147483647,5.5,6.6,123.12345,404.4,Hello World,a,abc,2020-07-17T18:00:22.123,2020-07-17,18:00:22,500,2020-07-17T18:00:22.123456789)");
-        List<String> actual = TestValuesTableFactory.getRawResults("sink");
+        List<String> actual = TestValuesTableFactory.getRawResultsAsStrings("sink");
         Collections.sort(expected);
         Collections.sort(actual);
         assertEquals(expected, actual);
@@ -373,7 +373,7 @@ public void testStartupFromLatestOffset() throws Exception {
         String[] expected =
                 new String[] {"110,jacket,new water resistent white wind breaker,0.500"};
 
-        List<String> actual = TestValuesTableFactory.getResults("sink");
+        List<String> actual = TestValuesTableFactory.getResultsAsStrings("sink");
         assertThat(actual, containsInAnyOrder(expected));
 
         cancelJobIfRunning(result);
@@ -470,7 +470,7 @@ public void testMetadataColumns() throws Throwable {
                         "+U(testdb,DB2INST1,PRODUCTS,111,scooter,Big 2-wheel scooter ,5.170)",
                         "-D(testdb,DB2INST1,PRODUCTS,111,scooter,Big 2-wheel scooter ,5.170)");
 
-        List<String> actual = TestValuesTableFactory.getRawResults("sink");
+        List<String> actual = TestValuesTableFactory.getRawResultsAsStrings("sink");
         Collections.sort(expected);
         Collections.sort(actual);
         assertEquals(expected, actual);
@@ -493,7 +493,7 @@ private static void waitForSinkSize(String sinkName, int expectedSize)
     private static int sinkSize(String sinkName) {
         synchronized (TestValuesTableFactory.class) {
             try {
-                return TestValuesTableFactory.getRawResults(sinkName).size();
+                return TestValuesTableFactory.getRawResultsAsStrings(sinkName).size();
             } catch (IllegalArgumentException e) {
                 // job is not started yet
                 return 0;