[FLINK-37652][Connectors/Opensearch] Move to Flink to 2.0.0

Author: reta

.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: [ 2.1-SNAPSHOT ]
+        jdk: [ '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,23 +30,16 @@ jobs:
     strategy:
       matrix:
         flink_branches: [ {
-          flink: 1.18-SNAPSHOT,
+          flink: 2.1-SNAPSHOT,
+          jdk: '11, 17, 21',
           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.20,
           jdk: '8, 11, 17, 21',
           branch: main
         }, {
-          flink: 1.19.0,
+          flink: 1.19.1,
           jdk: '8, 11, 17, 21',
           branch: v1.2
         }, {

flink-connector-opensearch-base/pom.xml

@@ -26,7 +26,7 @@ under the License.
 	<parent>
 		<groupId>org.apache.flink</groupId>
 		<artifactId>flink-connector-opensearch-parent</artifactId>
-		<version>2.0-SNAPSHOT</version>
+		<version>3.0-SNAPSHOT</version>
 	</parent>
 
 	<artifactId>flink-connector-opensearch-base</artifactId>

flink-connector-opensearch-base/src/test/java/org/apache/flink/connector/opensearch/table/TestContext.java

@@ -53,11 +53,12 @@ DynamicTableFactory.Context build() {
         return new FactoryUtil.DefaultDynamicTableContext(
                 ObjectIdentifier.of("default", "default", "t1"),
                 new ResolvedCatalogTable(
-                        CatalogTable.of(
-                                Schema.newBuilder().fromResolvedSchema(schema).build(),
-                                "mock context",
-                                Collections.emptyList(),
-                                options),
+                        CatalogTable.newBuilder()
+                                .schema(Schema.newBuilder().fromResolvedSchema(schema).build())
+                                .comment("mock context")
+                                .partitionKeys(Collections.emptyList())
+                                .options(options)
+                                .build(),
                         schema),
                 Collections.emptyMap(),
                 new Configuration(),

flink-connector-opensearch-base/src/test/java/org/apache/flink/streaming/connectors/opensearch/testutils/SourceSinkDataTestKit.java

@@ -26,7 +26,7 @@ under the License.
 	<parent>
 		<groupId>org.apache.flink</groupId>
 		<artifactId>flink-connector-opensearch-parent</artifactId>
-		<version>2.0-SNAPSHOT</version>
+		<version>3.0-SNAPSHOT</version>
 	</parent>
 
 	<artifactId>flink-connector-opensearch-e2e-tests</artifactId>

flink-connector-opensearch-e2e-tests/pom.xml

@@ -18,25 +18,20 @@
 package org.apache.flink.streaming.tests;
 
 import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.connector.opensearch.sink.FailureHandler;
+import org.apache.flink.connector.opensearch.sink.OpensearchEmitter;
+import org.apache.flink.connector.opensearch.sink.OpensearchSinkBuilder;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.opensearch.ActionRequestFailureHandler;
-import org.apache.flink.streaming.connectors.opensearch.OpensearchSink;
-import org.apache.flink.streaming.connectors.opensearch.RequestIndexer;
 import org.apache.flink.util.Collector;
-
+import org.apache.flink.util.ParameterTool;
 import org.apache.http.HttpHost;
-import org.opensearch.action.ActionRequest;
 import org.opensearch.action.index.IndexRequest;
 import org.opensearch.action.update.UpdateRequest;
 import org.opensearch.client.Requests;
 
-import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 
 /** End to end test for OpensearchSink. */
@@ -69,31 +64,21 @@ public void flatMap(
                                     }
                                 });
 
-        List<HttpHost> httpHosts = new ArrayList<>();
-        httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
-
-        OpensearchSink.Builder<Tuple2<String, String>> osSinkBuilder =
-                new OpensearchSink.Builder<>(
-                        httpHosts,
-                        (Tuple2<String, String> element,
-                                RuntimeContext ctx,
-                                RequestIndexer indexer) -> {
+        OpensearchSinkBuilder<Tuple2<String, String>> osSinkBuilder =
+            new OpensearchSinkBuilder<>()
+                .setHosts(new HttpHost("127.0.0.1", 9200, "http"))
+                .setEmitter((OpensearchEmitter<Tuple2<String, String>>) (element, writer, indexer) -> {
                             indexer.add(createIndexRequest(element.f1, parameterTool));
                             indexer.add(createUpdateRequest(element, parameterTool));
-                        });
-
-        osSinkBuilder.setFailureHandler(
-                new CustomFailureHandler(parameterTool.getRequired("index")));
-
-        // this instructs the sink to emit after every element, otherwise they would be buffered
-        osSinkBuilder.setBulkFlushMaxActions(1);
-
-        source.addSink(osSinkBuilder.build());
+                        })
+                .setFailureHandler(new CustomFailureHandler(parameterTool.getRequired("index")))
+                .setBulkFlushMaxActions(1);
 
+        source.sinkTo(osSinkBuilder.build());
         env.execute("Opensearch end to end sink test example");
     }
 
-    private static class CustomFailureHandler implements ActionRequestFailureHandler {
+    private static class CustomFailureHandler implements FailureHandler {
 
         private static final long serialVersionUID = 942269087742453482L;
 
@@ -104,21 +89,8 @@ private static class CustomFailureHandler implements ActionRequestFailureHandler
         }
 
         @Override
-        public void onFailure(
-                ActionRequest action, Throwable failure, int restStatusCode, RequestIndexer indexer)
-                throws Throwable {
-            if (action instanceof IndexRequest) {
-                Map<String, Object> json = new HashMap<>();
-                json.put("data", ((IndexRequest) action).source());
-
-                indexer.add(
-                        Requests.indexRequest()
-                                .index(index)
-                                .id(((IndexRequest) action).id())
-                                .source(json));
-            } else {
-                throw new IllegalStateException("unexpected");
-            }
+        public void onFailure(Throwable failure) {
+            throw new IllegalStateException("Unexpected exception for index:" + index);
         }
     }
 

flink-connector-opensearch-e2e-tests/src/main/java/org/apache/flink/streaming/tests/OpensearchSinkExample.java

@@ -26,7 +26,7 @@ under the License.
 	<parent>
 		<groupId>org.apache.flink</groupId>
 		<artifactId>flink-connector-opensearch-parent</artifactId>
-		<version>2.0-SNAPSHOT</version>
+		<version>3.0-SNAPSHOT</version>
 	</parent>
 
 	<artifactId>flink-connector-opensearch</artifactId>