[Flink-33859] Support OpenSearch v2

Author: snuyanzin

flink-connector-opensearch-base/archunit-violations/5c4a6228-f9cb-4828-9625-43c57d133967

@@ -1,3 +1,3 @@
 Constructor <org.apache.flink.connector.opensearch.sink.BulkProcessorConfig.<init>(int, int, long, org.apache.flink.connector.opensearch.sink.FlushBackoffType, int, long)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (BulkProcessorConfig.java:44)
 Constructor <org.apache.flink.connector.opensearch.table.OpensearchConfiguration.<init>(org.apache.flink.configuration.ReadableConfig)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (OpensearchConfiguration.java:61)
-Method <org.apache.flink.connector.opensearch.table.IndexGeneratorFactory.createRuntimeIndexGenerator(java.lang.String, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, org.apache.flink.connector.opensearch.table.IndexGeneratorFactory$IndexHelper, java.time.ZoneId)> has parameter of type <[Lorg.apache.flink.table.types.DataType;> in (IndexGeneratorFactory.java:0)
+Method <org.apache.flink.connector.opensearch.table.IndexGeneratorFactory.createRuntimeIndexGenerator(java.lang.String, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, org.apache.flink.connector.opensearch.table.IndexGeneratorFactory$IndexHelper, java.time.ZoneId)> has parameter of type <[Lorg.apache.flink.table.types.DataType;> in (IndexGeneratorFactory.java:0)

flink-connector-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/BulkResponseInspector.java renamed to flink-connector-opensearch-base/src/main/java/org/apache/flink/connector/opensearch/sink/BulkResponseInspector.java

@@ -0,0 +1,14 @@
+package org.apache.flink.connector.opensearch.sink;
+
+import org.apache.flink.util.FlinkRuntimeException;
+
+class DefaultFailureHandler implements FailureHandler {
+
+    @Override
+    public void onFailure(Throwable failure) {
+        if (failure instanceof FlinkRuntimeException) {
+            throw (FlinkRuntimeException) failure;
+        }
+        throw new FlinkRuntimeException(failure);
+    }
+}

flink-connector-opensearch-base/src/main/java/org/apache/flink/connector/opensearch/sink/DefaultFailureHandler.java

@@ -15,4 +15,4 @@ org.apache.flink.connector.opensearch.table.OpensearchDynamicSinkITCase does not
 * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\
 * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
+ or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule

flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/TestingSinkWriterMetricGroup.java renamed to flink-connector-opensearch-base/src/test/java/org/apache/flink/connector/opensearch/sink/TestingSinkWriterMetricGroup.java

@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.opensearch.sink;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.opensearch.action.DocWriteRequest;
+import org.opensearch.action.bulk.BulkItemResponse;
+import org.opensearch.action.bulk.BulkRequest;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.rest.RestStatus;
+
+import static org.apache.flink.util.ExceptionUtils.firstOrSuppressed;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A strict implementation that fails if either the whole bulk request failed or any of its actions.
+ */
+class DefaultBulkResponseInspector implements BulkResponseInspector {
+
+    @VisibleForTesting final FailureHandler failureHandler;
+
+    DefaultBulkResponseInspector() {
+        this(new DefaultFailureHandler());
+    }
+
+    DefaultBulkResponseInspector(FailureHandler failureHandler) {
+        this.failureHandler = checkNotNull(failureHandler);
+    }
+
+    @Override
+    public void inspect(BulkRequest request, BulkResponse response) {
+        if (!response.hasFailures()) {
+            return;
+        }
+
+        Throwable chainedFailures = null;
+        for (int i = 0; i < response.getItems().length; i++) {
+            final BulkItemResponse itemResponse = response.getItems()[i];
+            if (!itemResponse.isFailed()) {
+                continue;
+            }
+            final Throwable failure = itemResponse.getFailure().getCause();
+            if (failure == null) {
+                continue;
+            }
+            final RestStatus restStatus = itemResponse.getFailure().getStatus();
+            final DocWriteRequest<?> actionRequest = request.requests().get(i);
+
+            chainedFailures =
+                    firstOrSuppressed(
+                            wrapException(restStatus, failure, actionRequest), chainedFailures);
+        }
+        if (chainedFailures == null) {
+            return;
+        }
+        failureHandler.onFailure(chainedFailures);
+    }
+
+    private static Throwable wrapException(
+            RestStatus restStatus, Throwable rootFailure, DocWriteRequest<?> actionRequest) {
+        if (restStatus == null) {
+            return new FlinkRuntimeException(
+                    String.format("Single action %s of bulk request failed.", actionRequest),
+                    rootFailure);
+        } else {
+            return new FlinkRuntimeException(
+                    String.format(
+                            "Single action %s of bulk request failed with status %s.",
+                            actionRequest, restStatus.getStatus()),
+                    rootFailure);
+        }
+    }
+}

flink-connector-opensearch/archunit-violations/4382f1f0-807a-45ff-97d8-42f72b6e9484

@@ -21,8 +21,6 @@
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.opensearch.sink.BulkResponseInspector.BulkResponseInspectorFactory;
-import org.apache.flink.connector.opensearch.sink.OpensearchWriter.DefaultBulkResponseInspector;
-import org.apache.flink.connector.opensearch.sink.OpensearchWriter.DefaultFailureHandler;
 import org.apache.flink.util.InstantiationUtil;
 
 import org.apache.http.HttpHost;

flink-connector-opensearch/archunit-violations/5c4a6228-f9cb-4828-9625-43c57d133967

@@ -31,7 +31,6 @@
 import org.opensearch.action.ActionListener;
 import org.opensearch.action.DocWriteRequest;
 import org.opensearch.action.bulk.BackoffPolicy;
-import org.opensearch.action.bulk.BulkItemResponse;
 import org.opensearch.action.bulk.BulkProcessor;
 import org.opensearch.action.bulk.BulkRequest;
 import org.opensearch.action.bulk.BulkResponse;
@@ -53,7 +52,6 @@
 import java.util.List;
 import java.util.function.BiConsumer;
 
-import static org.apache.flink.util.ExceptionUtils.firstOrSuppressed;
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 class OpensearchWriter<IN> implements SinkWriter<IN> {
@@ -325,63 +323,6 @@ public void add(UpdateRequest... updateRequests) {
         }
     }
 
-    /**
-     * A strict implementation that fails if either the whole bulk request failed or any of its
-     * actions.
-     */
-    static class DefaultBulkResponseInspector implements BulkResponseInspector {
-
-        @VisibleForTesting final FailureHandler failureHandler;
-
-        DefaultBulkResponseInspector() {
-            this(new DefaultFailureHandler());
-        }
-
-        DefaultBulkResponseInspector(FailureHandler failureHandler) {
-            this.failureHandler = checkNotNull(failureHandler);
-        }
-
-        @Override
-        public void inspect(BulkRequest request, BulkResponse response) {
-            if (!response.hasFailures()) {
-                return;
-            }
-
-            Throwable chainedFailures = null;
-            for (int i = 0; i < response.getItems().length; i++) {
-                final BulkItemResponse itemResponse = response.getItems()[i];
-                if (!itemResponse.isFailed()) {
-                    continue;
-                }
-                final Throwable failure = itemResponse.getFailure().getCause();
-                if (failure == null) {
-                    continue;
-                }
-                final RestStatus restStatus = itemResponse.getFailure().getStatus();
-                final DocWriteRequest<?> actionRequest = request.requests().get(i);
-
-                chainedFailures =
-                        firstOrSuppressed(
-                                wrapException(restStatus, failure, actionRequest), chainedFailures);
-            }
-            if (chainedFailures == null) {
-                return;
-            }
-            failureHandler.onFailure(chainedFailures);
-        }
-    }
-
-    static class DefaultFailureHandler implements FailureHandler {
-
-        @Override
-        public void onFailure(Throwable failure) {
-            if (failure instanceof FlinkRuntimeException) {
-                throw (FlinkRuntimeException) failure;
-            }
-            throw new FlinkRuntimeException(failure);
-        }
-    }
-
     @Internal
     interface BulkRequestConsumerFactory
             extends BiConsumer<BulkRequest, ActionListener<BulkResponse>> {}

flink-connector-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/DefaultBulkResponseInspector.java

@@ -18,7 +18,6 @@
 
 package org.apache.flink.connector.opensearch.sink;
 
-import org.apache.flink.connector.opensearch.sink.OpensearchWriter.DefaultBulkResponseInspector;
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.TestLoggerExtension;
 

flink-connector-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchSinkBuilder.java

@@ -21,7 +21,6 @@
 import org.apache.flink.api.connector.sink2.Sink.InitContext;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.opensearch.sink.BulkResponseInspector.BulkResponseInspectorFactory;
-import org.apache.flink.connector.opensearch.sink.OpensearchWriter.DefaultBulkResponseInspector;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;

flink-connector-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchWriter.java

@@ -21,8 +21,6 @@
 import org.apache.flink.api.connector.sink2.SinkWriter;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.connector.opensearch.OpensearchUtil;
-import org.apache.flink.connector.opensearch.sink.OpensearchWriter.DefaultBulkResponseInspector;
-import org.apache.flink.connector.opensearch.sink.OpensearchWriter.DefaultFailureHandler;
 import org.apache.flink.connector.opensearch.test.DockerImageVersions;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Gauge;

flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/DefaultBulkInspectorTest.java

@@ -1,18 +1,18 @@
 org.apache.flink.connector.opensearch.sink.Opensearch2SinkITCase does not satisfy: only one of the following predicates match:\
 * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
+* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\
 * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
  or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
 org.apache.flink.connector.opensearch.sink.Opensearch2WriterITCase does not satisfy: only one of the following predicates match:\
 * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
+* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\
 * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
  or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
 org.apache.flink.connector.opensearch.table.Opensearch2DynamicSinkITCase does not satisfy: only one of the following predicates match:\
 * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
+* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\
 * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
+ or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule