[FLINK-37517][[Connectors/Elasticsearch] Bump flink to 2.0.0

Bump flink to 2.0

Author: reswqa

.github/workflows/push_pr.yml

@@ -28,8 +28,8 @@ jobs:
   compile_and_test:
     strategy:
       matrix:
-        flink: [ 1.19-SNAPSHOT, 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,11 +30,9 @@ jobs:
     strategy:
       matrix:
         flink_branches: [{
-          flink: 1.19-SNAPSHOT,
-          branch: main
-        }, {
-          flink: 1.20-SNAPSHOT,
-          branch: main
+          flink: 2.1-SNAPSHOT,
+          branch: main,
+          jdk: '11, 17, 21'
         }, {
           flink: 1.19.1,
           branch: v3.0

flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/ElasticsearchApiCallBridge.java

@@ -0,0 +1,68 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.search.SearchRequest;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * An {@link ElasticsearchApiCallBridge} is used to bridge incompatible Elasticsearch Java API calls
+ * across different versions. This includes calls to create Elasticsearch clients, handle failed
+ * item responses, etc. Any incompatible Elasticsearch Java APIs should be bridged using this
+ * interface.
+ *
+ * <p>Implementations are allowed to be stateful. For example, for Elasticsearch 1.x, since
+ * connecting via an embedded node is allowed, the call bridge will hold reference to the created
+ * embedded node. Each instance of the sink will hold exactly one instance of the call bridge, and
+ * state cleanup is performed when the sink is closed.
+ *
+ * @param <C> The Elasticsearch client, that implements {@link AutoCloseable}.
+ */
+@Internal
+public interface ElasticsearchApiCallBridge<C extends AutoCloseable> extends Serializable {
+
+    /**
+     * Creates an Elasticsearch client implementing {@link AutoCloseable}.
+     *
+     * @return The created client.
+     */
+    C createClient(NetworkClientConfig networkClientConfig, List<HttpHost> hosts);
+
+    /**
+     * Executes a search using the Search API.
+     *
+     * @param client the Elasticsearch client.
+     * @param searchRequest A request to execute search against one or more indices (or all).
+     */
+    Tuple2<String, String[]> search(C client, SearchRequest searchRequest) throws IOException;
+
+    /**
+     * Closes this client and releases any system resources associated with it.
+     *
+     * @param client the Elasticsearch client.
+     */
+    void close(C client) throws IOException;
+}

flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/sink/NetworkClientConfig.java renamed to flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/NetworkClientConfig.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.connector.elasticsearch.sink;
+package org.apache.flink.connector.elasticsearch;
 
 import org.apache.flink.util.function.SerializableSupplier;
 
@@ -26,7 +26,8 @@
 
 import java.io.Serializable;
 
-class NetworkClientConfig implements Serializable {
+/** Network config for es client. */
+public class NetworkClientConfig implements Serializable {
 
     @Nullable private final String username;
     @Nullable private final String password;
@@ -37,7 +38,7 @@ class NetworkClientConfig implements Serializable {
     @Nullable private final SerializableSupplier<SSLContext> sslContextSupplier;
     @Nullable private final SerializableSupplier<HostnameVerifier> sslHostnameVerifier;
 
-    NetworkClientConfig(
+    public NetworkClientConfig(
             @Nullable String username,
             @Nullable String password,
             @Nullable String connectionPathPrefix,
@@ -95,4 +96,69 @@ public SerializableSupplier<SSLContext> getSSLContextSupplier() {
     public SerializableSupplier<HostnameVerifier> getSslHostnameVerifier() {
         return sslHostnameVerifier;
     }
+
+    /** Builder for {@link NetworkClientConfig}. */
+    public static class Builder {
+        private String username;
+        private String password;
+        private String connectionPathPrefix;
+        private Integer connectionRequestTimeout;
+        private Integer connectionTimeout;
+        private Integer socketTimeout;
+        private SerializableSupplier<SSLContext> sslContextSupplier;
+        private SerializableSupplier<HostnameVerifier> sslHostnameVerifier;
+
+        public Builder setUsername(String username) {
+            this.username = username;
+            return this;
+        }
+
+        public Builder setPassword(String password) {
+            this.password = password;
+            return this;
+        }
+
+        public Builder setConnectionPathPrefix(String connectionPathPrefix) {
+            this.connectionPathPrefix = connectionPathPrefix;
+            return this;
+        }
+
+        public Builder setConnectionRequestTimeout(Integer connectionRequestTimeout) {
+            this.connectionRequestTimeout = connectionRequestTimeout;
+            return this;
+        }
+
+        public Builder setConnectionTimeout(Integer connectionTimeout) {
+            this.connectionTimeout = connectionTimeout;
+            return this;
+        }
+
+        public Builder setSocketTimeout(Integer socketTimeout) {
+            this.socketTimeout = socketTimeout;
+            return this;
+        }
+
+        public Builder setSslContextSupplier(SerializableSupplier<SSLContext> sslContextSupplier) {
+            this.sslContextSupplier = sslContextSupplier;
+            return this;
+        }
+
+        public Builder setSslHostnameVerifier(
+                SerializableSupplier<HostnameVerifier> sslHostnameVerifier) {
+            this.sslHostnameVerifier = sslHostnameVerifier;
+            return this;
+        }
+
+        public NetworkClientConfig build() {
+            return new NetworkClientConfig(
+                    username,
+                    password,
+                    connectionPathPrefix,
+                    connectionRequestTimeout,
+                    connectionTimeout,
+                    socketTimeout,
+                    sslContextSupplier,
+                    sslHostnameVerifier);
+        }
+    }
 }

flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchRowDataLookupFunction.java renamed to flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/lookup/ElasticsearchRowDataLookupFunction.java

@@ -16,12 +16,13 @@
  * limitations under the License.
  */
 
-package org.apache.flink.streaming.connectors.elasticsearch.table;
+package org.apache.flink.connector.elasticsearch.lookup;
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchApiCallBridge;
+import org.apache.flink.connector.elasticsearch.ElasticsearchApiCallBridge;
+import org.apache.flink.connector.elasticsearch.NetworkClientConfig;
 import org.apache.flink.table.connector.source.LookupTableSource;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.util.DataFormatConverters;
@@ -31,6 +32,7 @@
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.Preconditions;
 
+import org.apache.http.HttpHost;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.index.query.BoolQueryBuilder;
@@ -44,6 +46,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -72,6 +75,8 @@ public class ElasticsearchRowDataLookupFunction<C extends AutoCloseable> extends
     private SearchSourceBuilder searchSourceBuilder;
 
     private final ElasticsearchApiCallBridge<C> callBridge;
+    private final NetworkClientConfig networkClientConfig;
+    private final List<HttpHost> hosts;
 
     private transient C client;
 
@@ -83,13 +88,17 @@ public ElasticsearchRowDataLookupFunction(
             String[] producedNames,
             DataType[] producedTypes,
             String[] lookupKeys,
+            List<HttpHost> hosts,
+            NetworkClientConfig networkClientConfig,
             ElasticsearchApiCallBridge<C> callBridge) {
 
         checkNotNull(deserializationSchema, "No DeserializationSchema supplied.");
         checkNotNull(maxRetryTimes, "No maxRetryTimes supplied.");
         checkNotNull(producedNames, "No fieldNames supplied.");
         checkNotNull(producedTypes, "No fieldTypes supplied.");
         checkNotNull(lookupKeys, "No keyNames supplied.");
+        checkNotNull(hosts, "No hosts supplied.");
+        checkNotNull(networkClientConfig, "No networkClientConfig supplied.");
         checkNotNull(callBridge, "No ElasticsearchApiCallBridge supplied.");
 
         this.deserializationSchema = deserializationSchema;
@@ -110,12 +119,14 @@ public ElasticsearchRowDataLookupFunction(
             converters[i] = DataFormatConverters.getConverterForDataType(producedTypes[position]);
         }
 
+        this.networkClientConfig = networkClientConfig;
+        this.hosts = hosts;
         this.callBridge = callBridge;
     }
 
     @Override
     public void open(FunctionContext context) throws Exception {
-        this.client = callBridge.createClient();
+        this.client = callBridge.createClient(networkClientConfig, hosts);
 
         // Set searchRequest in open method in case of amount of calling in eval method when every
         // record comes.

flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSink.java

@@ -22,7 +22,9 @@
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.connector.sink2.Sink;
 import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
 import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.elasticsearch.NetworkClientConfig;
 import org.apache.flink.connector.elasticsearch.sink.BulkResponseInspector.BulkResponseInspectorFactory;
 
 import org.apache.http.HttpHost;
@@ -82,7 +84,7 @@ public class ElasticsearchSink<IN> implements Sink<IN> {
     }
 
     @Override
-    public SinkWriter<IN> createWriter(InitContext context) throws IOException {
+    public SinkWriter<IN> createWriter(WriterInitContext context) throws IOException {
         return new ElasticsearchWriter<>(
                 hosts,
                 emitter,

flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBuilderBase.java

@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.java.ClosureCleaner;
 import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.elasticsearch.NetworkClientConfig;
 import org.apache.flink.connector.elasticsearch.sink.BulkResponseInspector.BulkResponseInspectorFactory;
 import org.apache.flink.connector.elasticsearch.sink.ElasticsearchWriter.DefaultBulkResponseInspector;
 import org.apache.flink.connector.elasticsearch.sink.ElasticsearchWriter.DefaultFailureHandler;