[FLINK-36651][Connectors/Elasticsearch] Fix IT test not compatible with 1.20, drop main branch tests for 1.18

Author: vahmed-hamdy

.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 }}

.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

flink-connector-elasticsearch-base/src/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;
-        }
-    }
 }

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>