[FLINK-36651][Connectors/Elasticsearch] Fix IT test not compatible wi…

…th 1.20, drop main branch tests for 1.18
apache · Nov 15, 2024 · 9b545d0 · 9b545d0
1 parent da2ef1f
commit 9b545d0
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 47 deletions.
diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml
@@ -28,13 +28,8 @@ jobs:
   compile_and_test:
     strategy:
       matrix:
-        flink: [ 1.18-SNAPSHOT ]
-        jdk: [ '8, 11, 17' ]
-        include:
-          - flink: 1.19-SNAPSHOT
-            jdk: '8, 11, 17, 21'
-          - flink: 1.20-SNAPSHOT
-            jdk: '8, 11, 17, 21'
+        flink: [ 1.19-SNAPSHOT, 1.20-SNAPSHOT ]
+        jdk: [ '8, 11, 17, 21' ]
     uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
     with:
       flink_version: ${{ matrix.flink }}

diff --git a/.github/workflows/weekly.yml b/.github/workflows/weekly.yml
@@ -30,27 +30,21 @@ jobs:
     strategy:
       matrix:
         flink_branches: [{
-          flink: 1.18-SNAPSHOT,
-          jdk: '8, 11, 17',
-          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,
-          branch: v3.0
         }, {
           flink: 1.19.0,
           branch: v3.0
+        }, {
+          flink: 1.20.0,
+          branch: main
         }]
     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' }}
+      jdk_version: ${{ matrix.flink_branches.jdk || '8, 11, 17, 21' }}
       run_dependency_convergence: false
diff --git a/...rc/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchWriterITCase.java b/...rc/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchWriterITCase.java
@@ -17,7 +17,6 @@
 
 package org.apache.flink.connector.elasticsearch.sink;
 
-import org.apache.flink.api.common.operators.MailboxExecutor;
 import org.apache.flink.api.connector.sink2.SinkWriter;
 import org.apache.flink.api.connector.sink2.SinkWriter.Context;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -31,14 +30,13 @@
 import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
 import org.apache.flink.runtime.metrics.MetricNames;
 import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
 import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
 import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
 import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.TestLoggerExtension;
-import org.apache.flink.util.function.ThrowingRunnable;
 
 import org.apache.http.HttpHost;
 import org.elasticsearch.action.ActionListener;
@@ -329,7 +327,7 @@ private static ElasticsearchWriter<Tuple2<Integer, String>> createWriter(
                 new DefaultBulkResponseInspector(),
                 new NetworkClientConfig(null, null, null, null, null, null, null, null),
                 metricGroup,
-                new TestMailbox());
+                new SyncMailboxExecutor());
     }
 
     private TestingSinkWriterMetricGroup getSinkWriterMetricGroup() {
@@ -481,29 +479,4 @@ GetResponse getResponse(String index, int id) throws IOException {
             return client.get(new GetRequest(index, Integer.toString(id)), RequestOptions.DEFAULT);
         }
     }
-
-    private static class TestMailbox implements MailboxExecutor {
-
-        @Override
-        public void execute(
-                ThrowingRunnable<? extends Exception> command,
-                String descriptionFormat,
-                Object... descriptionArgs) {
-            try {
-                command.run();
-            } catch (Exception e) {
-                throw new RuntimeException("Unexpected error", e);
-            }
-        }
-
-        @Override
-        public void yield() throws InterruptedException, FlinkRuntimeException {
-            Thread.sleep(100);
-        }
-
-        @Override
-        public boolean tryYield() throws FlinkRuntimeException {
-            return false;
-        }
-    }
 }
diff --git a/pom.xml b/pom.xml
@@ -51,7 +51,7 @@ under the License.
 	</modules>
 
 	<properties>
-		<flink.version>1.18.0</flink.version>
+		<flink.version>1.20.0</flink.version>
 
 		<jackson-bom.version>2.15.3</jackson-bom.version>
 		<junit4.version>4.13.2</junit4.version>