You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/10/14 15:48:11 UTC
[flink] 01/02: [FLINK-13025] Elasticsearch 7.x support
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e2c4213a8f0664285658f25f2169487eb952909e
Author: yanghua <ya...@gmail.com>
AuthorDate: Wed Sep 18 20:05:05 2019 +0800
[FLINK-13025] Elasticsearch 7.x support
---
docs/dev/connectors/elasticsearch.md | 11 +-
docs/dev/connectors/elasticsearch.zh.md | 5 +
.../ElasticsearchUpsertTableSinkBase.java | 3 +
.../table/descriptors/ElasticsearchValidator.java | 4 +-
.../flink-connector-elasticsearch6/pom.xml | 8 -
.../pom.xml | 38 +--
.../Elasticsearch7ApiCallBridge.java | 143 +++++++++++
.../Elasticsearch7BulkProcessorIndexer.java | 85 ++++++
.../elasticsearch7/ElasticsearchSink.java | 238 +++++++++++++++++
.../elasticsearch7/RestClientFactory.java | 40 +++
.../EmbeddedElasticsearchNodeEnvironmentImpl.java | 79 ++++++
.../elasticsearch7/ElasticsearchSinkITCase.java | 100 ++++++++
.../src/test/resources/log4j-test.properties | 24 ++
.../flink-sql-connector-elasticsearch7/pom.xml | 205 +++++++++++++++
.../Elasticsearch7UpsertTableSink.java | 285 +++++++++++++++++++++
.../Elasticsearch7UpsertTableSinkFactory.java | 76 ++++++
.../src/main/resources/META-INF/NOTICE | 45 ++++
.../org.apache.flink.table.factories.TableFactory | 16 ++
.../Elasticsearch7UpsertTableSinkFactoryTest.java | 217 ++++++++++++++++
.../src/test/resources/log4j-test.properties | 24 ++
flink-connectors/pom.xml | 2 +
.../flink-elasticsearch7-test/pom.xml | 92 +++++++
.../streaming/tests/Elasticsearch7SinkExample.java | 150 +++++++++++
flink-end-to-end-tests/pom.xml | 1 +
tools/travis/stage.sh | 2 +
25 files changed, 1851 insertions(+), 42 deletions(-)
diff --git a/docs/dev/connectors/elasticsearch.md b/docs/dev/connectors/elasticsearch.md
index 597b89b..23cdb8e 100644
--- a/docs/dev/connectors/elasticsearch.md
+++ b/docs/dev/connectors/elasticsearch.md
@@ -53,7 +53,12 @@ of the Elasticsearch installation:
<tr>
<td>flink-connector-elasticsearch6{{ site.scala_version_suffix }}</td>
<td>1.6.0</td>
- <td>6 and later versions</td>
+ <td>6.x</td>
+ </tr>
+ <tr>
+ <td>flink-connector-elasticsearch7{{ site.scala_version_suffix }}</td>
+ <td>1.10.0</td>
+ <td>7 and later versions</td>
</tr>
</tbody>
</table>
@@ -123,7 +128,7 @@ input.addSink(new ElasticsearchSink<>(config, transportAddresses, new Elasticsea
}
}));{% endhighlight %}
</div>
-<div data-lang="java, Elasticsearch 6.x" markdown="1">
+<div data-lang="java, Elasticsearch 6.x and above" markdown="1">
{% highlight java %}
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -226,7 +231,7 @@ input.addSink(new ElasticsearchSink(config, transportAddresses, new Elasticsearc
}))
{% endhighlight %}
</div>
-<div data-lang="scala, Elasticsearch 6.x" markdown="1">
+<div data-lang="scala, Elasticsearch 6.x and above" markdown="1">
{% highlight scala %}
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.datastream.DataStream
diff --git a/docs/dev/connectors/elasticsearch.zh.md b/docs/dev/connectors/elasticsearch.zh.md
index 1fba302..e816da8 100644
--- a/docs/dev/connectors/elasticsearch.zh.md
+++ b/docs/dev/connectors/elasticsearch.zh.md
@@ -55,6 +55,11 @@ of the Elasticsearch installation:
<td>1.6.0</td>
<td>6 and later versions</td>
</tr>
+ <tr>
+ <td>flink-connector-elasticsearch7{{ site.scala_version_suffix }}</td>
+ <td>1.10.0</td>
+ <td>7 and later versions</td>
+ </tr>
</tbody>
</table>
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java
index c2c5181..eba7ead 100644
--- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchUpsertTableSinkBase.java
@@ -377,6 +377,7 @@ public abstract class ElasticsearchUpsertTableSinkBase implements UpsertStreamTa
/**
* Creates an update request to be added to a {@link RequestIndexer}.
+ * Note: the type field has been deprecated since Elasticsearch 7.x and it would not take any effort.
*/
UpdateRequest createUpdateRequest(
String index,
@@ -387,6 +388,7 @@ public abstract class ElasticsearchUpsertTableSinkBase implements UpsertStreamTa
/**
* Creates an index request to be added to a {@link RequestIndexer}.
+ * Note: the type field has been deprecated since Elasticsearch 7.x and it would not take any effort.
*/
IndexRequest createIndexRequest(
String index,
@@ -396,6 +398,7 @@ public abstract class ElasticsearchUpsertTableSinkBase implements UpsertStreamTa
/**
* Creates a delete request to be added to a {@link RequestIndexer}.
+ * Note: the type field has been deprecated since Elasticsearch 7.x and it would not take any effort.
*/
DeleteRequest createDeleteRequest(
String index,
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/table/descriptors/ElasticsearchValidator.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/table/descriptors/ElasticsearchValidator.java
index 9d1a926..ec48557 100644
--- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/table/descriptors/ElasticsearchValidator.java
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/table/descriptors/ElasticsearchValidator.java
@@ -21,7 +21,6 @@ package org.apache.flink.table.descriptors;
import org.apache.flink.annotation.Internal;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
@@ -36,6 +35,7 @@ public class ElasticsearchValidator extends ConnectorDescriptorValidator {
public static final String CONNECTOR_TYPE_VALUE_ELASTICSEARCH = "elasticsearch";
public static final String CONNECTOR_VERSION_VALUE_6 = "6";
+ public static final String CONNECTOR_VERSION_VALUE_7 = "7";
public static final String CONNECTOR_HOSTS = "connector.hosts";
public static final String CONNECTOR_HOSTS_HOSTNAME = "hostname";
public static final String CONNECTOR_HOSTS_PORT = "port";
@@ -79,7 +79,7 @@ public class ElasticsearchValidator extends ConnectorDescriptorValidator {
properties.validateEnumValues(
CONNECTOR_VERSION,
false,
- Collections.singletonList(CONNECTOR_VERSION_VALUE_6));
+ Arrays.asList(CONNECTOR_VERSION_VALUE_6, CONNECTOR_VERSION_VALUE_7));
}
private void validateHosts(DescriptorProperties properties) {
diff --git a/flink-connectors/flink-connector-elasticsearch6/pom.xml b/flink-connectors/flink-connector-elasticsearch6/pom.xml
index cc0bbf2..5bdf889 100644
--- a/flink-connectors/flink-connector-elasticsearch6/pom.xml
+++ b/flink-connectors/flink-connector-elasticsearch6/pom.xml
@@ -93,14 +93,6 @@ under the License.
<scope>provided</scope>
<optional>true</optional>
</dependency>
- <!-- A planner dependency won't be necessary once FLIP-32 has been completed. -->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- <optional>true</optional>
- </dependency>
<!-- test dependencies -->
diff --git a/flink-connectors/flink-connector-elasticsearch6/pom.xml b/flink-connectors/flink-connector-elasticsearch7/pom.xml
similarity index 82%
copy from flink-connectors/flink-connector-elasticsearch6/pom.xml
copy to flink-connectors/flink-connector-elasticsearch7/pom.xml
index cc0bbf2..3c68eac 100644
--- a/flink-connectors/flink-connector-elasticsearch6/pom.xml
+++ b/flink-connectors/flink-connector-elasticsearch7/pom.xml
@@ -18,26 +18,26 @@ specific language governing permissions and limitations
under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>org.apache.flink</groupId>
<artifactId>flink-connectors</artifactId>
+ <groupId>org.apache.flink</groupId>
<version>1.10-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
- <artifactId>flink-connector-elasticsearch6_${scala.binary.version}</artifactId>
- <name>flink-connector-elasticsearch6</name>
+ <artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
+ <name>flink-connector-elasticsearch7</name>
<packaging>jar</packaging>
<!-- Allow users to pass custom connector versions -->
<properties>
- <elasticsearch.version>6.3.1</elasticsearch.version>
+ <elasticsearch.version>7.3.2</elasticsearch.version>
</properties>
<dependencies>
@@ -64,26 +64,13 @@ under the License.
</exclusions>
</dependency>
- <!-- Dependency for Elasticsearch 6.x REST Client -->
+ <!-- Dependency for Elasticsearch 7.x REST Client -->
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>${elasticsearch.version}</version>
</dependency>
- <!--
- Elasticsearch 5.x uses Log4j2 and no longer detects logging implementations, making
- Log4j2 a strict dependency. The following is added so that the Log4j2 API in
- Elasticsearch 5.x is routed to SLF4J. This way, user projects can remain flexible
- in the logging implementation preferred.
- -->
-
- <dependency>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-to-slf4j</artifactId>
- <version>2.9.1</version>
- </dependency>
-
<!-- Table ecosystem -->
<!-- Projects depending on this project won't depend on flink-table-*. -->
<dependency>
@@ -93,14 +80,6 @@ under the License.
<scope>provided</scope>
<optional>true</optional>
</dependency>
- <!-- A planner dependency won't be necessary once FLIP-32 has been completed. -->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- <optional>true</optional>
- </dependency>
<!-- test dependencies -->
@@ -134,7 +113,7 @@ under the License.
</dependency>
<!--
- Including elasticsearch transport dependency for tests. Netty3 is not here anymore in 6.x
+ Including elasticsearch transport dependency for tests. Netty3 is not here anymore in 7.x
-->
<dependency>
@@ -195,6 +174,7 @@ under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
+ <version>2.12.2</version>
<configuration>
<systemPropertyVariables>
<jna.nosys>true</jna.nosys>
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7ApiCallBridge.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7ApiCallBridge.java
new file mode 100644
index 0000000..c569f3b
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7ApiCallBridge.java
@@ -0,0 +1,143 @@
+/*
+ * 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.streaming.connectors.elasticsearch7;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchApiCallBridge;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.bulk.BackoffPolicy;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.common.unit.TimeValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Implementation of {@link ElasticsearchApiCallBridge} for Elasticsearch 7 and later versions.
+ */
+@Internal
+public class Elasticsearch7ApiCallBridge implements ElasticsearchApiCallBridge<RestHighLevelClient> {
+
+ private static final long serialVersionUID = -5222683870097809633L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(Elasticsearch7ApiCallBridge.class);
+
+ /**
+ * User-provided HTTP Host.
+ */
+ private final List<HttpHost> httpHosts;
+
+ /**
+ * The factory to configure the rest client.
+ */
+ private final RestClientFactory restClientFactory;
+
+ Elasticsearch7ApiCallBridge(List<HttpHost> httpHosts, RestClientFactory restClientFactory) {
+ Preconditions.checkArgument(httpHosts != null && !httpHosts.isEmpty());
+ this.httpHosts = httpHosts;
+ this.restClientFactory = Preconditions.checkNotNull(restClientFactory);
+ }
+
+ @Override
+ public RestHighLevelClient createClient(Map<String, String> clientConfig) throws IOException {
+ RestClientBuilder builder = RestClient.builder(httpHosts.toArray(new HttpHost[httpHosts.size()]));
+ restClientFactory.configureRestClientBuilder(builder);
+
+ RestHighLevelClient rhlClient = new RestHighLevelClient(builder);
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Pinging Elasticsearch cluster via hosts {} ...", httpHosts);
+ }
+
+ if (!rhlClient.ping(RequestOptions.DEFAULT)) {
+ throw new RuntimeException("There are no reachable Elasticsearch nodes!");
+ }
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Created Elasticsearch RestHighLevelClient connected to {}", httpHosts.toString());
+ }
+
+ return rhlClient;
+ }
+
+ @Override
+ public BulkProcessor.Builder createBulkProcessorBuilder(RestHighLevelClient client, BulkProcessor.Listener listener) {
+ return BulkProcessor.builder((request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener);
+ }
+
+ @Override
+ public Throwable extractFailureCauseFromBulkItemResponse(BulkItemResponse bulkItemResponse) {
+ if (!bulkItemResponse.isFailed()) {
+ return null;
+ } else {
+ return bulkItemResponse.getFailure().getCause();
+ }
+ }
+
+ @Override
+ public void configureBulkProcessorBackoff(
+ BulkProcessor.Builder builder,
+ @Nullable ElasticsearchSinkBase.BulkFlushBackoffPolicy flushBackoffPolicy) {
+
+ BackoffPolicy backoffPolicy;
+ if (flushBackoffPolicy != null) {
+ switch (flushBackoffPolicy.getBackoffType()) {
+ case CONSTANT:
+ backoffPolicy = BackoffPolicy.constantBackoff(
+ new TimeValue(flushBackoffPolicy.getDelayMillis()),
+ flushBackoffPolicy.getMaxRetryCount());
+ break;
+ case EXPONENTIAL:
+ default:
+ backoffPolicy = BackoffPolicy.exponentialBackoff(
+ new TimeValue(flushBackoffPolicy.getDelayMillis()),
+ flushBackoffPolicy.getMaxRetryCount());
+ }
+ } else {
+ backoffPolicy = BackoffPolicy.noBackoff();
+ }
+
+ builder.setBackoffPolicy(backoffPolicy);
+ }
+
+ @Override
+ public RequestIndexer createBulkProcessorIndexer(
+ BulkProcessor bulkProcessor,
+ boolean flushOnCheckpoint,
+ AtomicLong numPendingRequestsRef) {
+ return new Elasticsearch7BulkProcessorIndexer(
+ bulkProcessor,
+ flushOnCheckpoint,
+ numPendingRequestsRef);
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7BulkProcessorIndexer.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7BulkProcessorIndexer.java
new file mode 100644
index 0000000..0b7a437
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7BulkProcessorIndexer.java
@@ -0,0 +1,85 @@
+/*
+ * 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.streaming.connectors.elasticsearch7;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Implementation of a {@link RequestIndexer}, using a {@link BulkProcessor}.
+ * {@link ActionRequest ActionRequests} will be buffered before sending a bulk request to the Elasticsearch cluster.
+ *
+ * <p>Note: This class is binary compatible to Elasticsearch 7.
+ */
+@Internal
+class Elasticsearch7BulkProcessorIndexer implements RequestIndexer {
+
+ private final BulkProcessor bulkProcessor;
+ private final boolean flushOnCheckpoint;
+ private final AtomicLong numPendingRequestsRef;
+
+ Elasticsearch7BulkProcessorIndexer(
+ BulkProcessor bulkProcessor,
+ boolean flushOnCheckpoint,
+ AtomicLong numPendingRequestsRef) {
+ this.bulkProcessor = checkNotNull(bulkProcessor);
+ this.flushOnCheckpoint = flushOnCheckpoint;
+ this.numPendingRequestsRef = checkNotNull(numPendingRequestsRef);
+ }
+
+ @Override
+ public void add(DeleteRequest... deleteRequests) {
+ for (DeleteRequest deleteRequest : deleteRequests) {
+ if (flushOnCheckpoint) {
+ numPendingRequestsRef.getAndIncrement();
+ }
+ this.bulkProcessor.add(deleteRequest);
+ }
+ }
+
+ @Override
+ public void add(IndexRequest... indexRequests) {
+ for (IndexRequest indexRequest : indexRequests) {
+ if (flushOnCheckpoint) {
+ numPendingRequestsRef.getAndIncrement();
+ }
+ this.bulkProcessor.add(indexRequest);
+ }
+ }
+
+ @Override
+ public void add(UpdateRequest... updateRequests) {
+ for (UpdateRequest updateRequest : updateRequests) {
+ if (flushOnCheckpoint) {
+ numPendingRequestsRef.getAndIncrement();
+ }
+ this.bulkProcessor.add(updateRequest);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSink.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSink.java
new file mode 100644
index 0000000..5b874e2
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSink.java
@@ -0,0 +1,238 @@
+/*
+ * 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.streaming.connectors.elasticsearch7;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
+import org.apache.flink.streaming.connectors.elasticsearch.util.NoOpFailureHandler;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.client.RestHighLevelClient;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Elasticsearch 7.x sink that requests multiple {@link ActionRequest ActionRequests}
+ * against a cluster for each incoming element.
+ *
+ * <p>The sink internally uses a {@link RestHighLevelClient} to communicate with an Elasticsearch cluster.
+ * The sink will fail if no cluster can be connected to using the provided transport addresses passed to the constructor.
+ *
+ * <p>Internally, the sink will use a {@link BulkProcessor} to send {@link ActionRequest ActionRequests}.
+ * This will buffer elements before sending a request to the cluster. The behaviour of the
+ * {@code BulkProcessor} can be configured using these config keys:
+ * <ul>
+ * <li> {@code bulk.flush.max.actions}: Maximum amount of elements to buffer
+ * <li> {@code bulk.flush.max.size.mb}: Maximum amount of data (in megabytes) to buffer
+ * <li> {@code bulk.flush.interval.ms}: Interval at which to flush data regardless of the other two
+ * settings in milliseconds
+ * </ul>
+ *
+ * <p>You also have to provide an {@link ElasticsearchSinkFunction}. This is used to create multiple
+ * {@link ActionRequest ActionRequests} for each incoming element. See the class level documentation of
+ * {@link ElasticsearchSinkFunction} for an example.
+ *
+ * @param <T> Type of the elements handled by this sink
+ */
+@PublicEvolving
+public class ElasticsearchSink<T> extends ElasticsearchSinkBase<T, RestHighLevelClient> {
+
+ private static final long serialVersionUID = 1L;
+
+ private ElasticsearchSink(
+ Map<String, String> bulkRequestsConfig,
+ List<HttpHost> httpHosts,
+ ElasticsearchSinkFunction<T> elasticsearchSinkFunction,
+ ActionRequestFailureHandler failureHandler,
+ RestClientFactory restClientFactory) {
+
+ super(new Elasticsearch7ApiCallBridge(httpHosts, restClientFactory), bulkRequestsConfig, elasticsearchSinkFunction, failureHandler);
+ }
+
+ /**
+ * A builder for creating an {@link ElasticsearchSink}.
+ *
+ * @param <T> Type of the elements handled by the sink this builder creates.
+ */
+ @PublicEvolving
+ public static class Builder<T> {
+
+ private final List<HttpHost> httpHosts;
+ private final ElasticsearchSinkFunction<T> elasticsearchSinkFunction;
+
+ private Map<String, String> bulkRequestsConfig = new HashMap<>();
+ private ActionRequestFailureHandler failureHandler = new NoOpFailureHandler();
+ private RestClientFactory restClientFactory = restClientBuilder -> {};
+
+ /**
+ * Creates a new {@code ElasticsearchSink} that connects to the cluster using a {@link RestHighLevelClient}.
+ *
+ * @param httpHosts The list of {@link HttpHost} to which the {@link RestHighLevelClient} connects to.
+ * @param elasticsearchSinkFunction This is used to generate multiple {@link ActionRequest} from the incoming element.
+ */
+ public Builder(List<HttpHost> httpHosts, ElasticsearchSinkFunction<T> elasticsearchSinkFunction) {
+ this.httpHosts = Preconditions.checkNotNull(httpHosts);
+ this.elasticsearchSinkFunction = Preconditions.checkNotNull(elasticsearchSinkFunction);
+ }
+
+ /**
+ * Sets the maximum number of actions to buffer for each bulk request.
+ *
+ * @param numMaxActions the maxinum number of actions to buffer per bulk request.
+ */
+ public void setBulkFlushMaxActions(int numMaxActions) {
+ Preconditions.checkArgument(
+ numMaxActions > 0,
+ "Max number of buffered actions must be larger than 0.");
+
+ this.bulkRequestsConfig.put(CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, String.valueOf(numMaxActions));
+ }
+
+ /**
+ * Sets the maximum size of buffered actions, in mb, per bulk request.
+ *
+ * @param maxSizeMb the maximum size of buffered actions, in mb.
+ */
+ public void setBulkFlushMaxSizeMb(int maxSizeMb) {
+ Preconditions.checkArgument(
+ maxSizeMb > 0,
+ "Max size of buffered actions must be larger than 0.");
+
+ this.bulkRequestsConfig.put(CONFIG_KEY_BULK_FLUSH_MAX_SIZE_MB, String.valueOf(maxSizeMb));
+ }
+
+ /**
+ * Sets the bulk flush interval, in milliseconds.
+ *
+ * @param intervalMillis the bulk flush interval, in milliseconds.
+ */
+ public void setBulkFlushInterval(long intervalMillis) {
+ Preconditions.checkArgument(
+ intervalMillis >= 0,
+ "Interval (in milliseconds) between each flush must be larger than or equal to 0.");
+
+ this.bulkRequestsConfig.put(CONFIG_KEY_BULK_FLUSH_INTERVAL_MS, String.valueOf(intervalMillis));
+ }
+
+ /**
+ * Sets whether or not to enable bulk flush backoff behaviour.
+ *
+ * @param enabled whether or not to enable backoffs.
+ */
+ public void setBulkFlushBackoff(boolean enabled) {
+ this.bulkRequestsConfig.put(CONFIG_KEY_BULK_FLUSH_BACKOFF_ENABLE, String.valueOf(enabled));
+ }
+
+ /**
+ * Sets the type of back of to use when flushing bulk requests.
+ *
+ * @param flushBackoffType the backoff type to use.
+ */
+ public void setBulkFlushBackoffType(FlushBackoffType flushBackoffType) {
+ this.bulkRequestsConfig.put(
+ CONFIG_KEY_BULK_FLUSH_BACKOFF_TYPE,
+ Preconditions.checkNotNull(flushBackoffType).toString());
+ }
+
+ /**
+ * Sets the maximum number of retries for a backoff attempt when flushing bulk requests.
+ *
+ * @param maxRetries the maximum number of retries for a backoff attempt when flushing bulk requests
+ */
+ public void setBulkFlushBackoffRetries(int maxRetries) {
+ Preconditions.checkArgument(
+ maxRetries > 0,
+ "Max number of backoff attempts must be larger than 0.");
+
+ this.bulkRequestsConfig.put(CONFIG_KEY_BULK_FLUSH_BACKOFF_RETRIES, String.valueOf(maxRetries));
+ }
+
+ /**
+ * Sets the amount of delay between each backoff attempt when flushing bulk requests, in milliseconds.
+ *
+ * @param delayMillis the amount of delay between each backoff attempt when flushing bulk requests, in milliseconds.
+ */
+ public void setBulkFlushBackoffDelay(long delayMillis) {
+ Preconditions.checkArgument(
+ delayMillis >= 0,
+ "Delay (in milliseconds) between each backoff attempt must be larger than or equal to 0.");
+ this.bulkRequestsConfig.put(CONFIG_KEY_BULK_FLUSH_BACKOFF_DELAY, String.valueOf(delayMillis));
+ }
+
+ /**
+ * Sets a failure handler for action requests.
+ *
+ * @param failureHandler This is used to handle failed {@link ActionRequest}.
+ */
+ public void setFailureHandler(ActionRequestFailureHandler failureHandler) {
+ this.failureHandler = Preconditions.checkNotNull(failureHandler);
+ }
+
+ /**
+ * Sets a REST client factory for custom client configuration.
+ *
+ * @param restClientFactory the factory that configures the rest client.
+ */
+ public void setRestClientFactory(RestClientFactory restClientFactory) {
+ this.restClientFactory = Preconditions.checkNotNull(restClientFactory);
+ }
+
+ /**
+ * Creates the Elasticsearch sink.
+ *
+ * @return the created Elasticsearch sink.
+ */
+ public ElasticsearchSink<T> build() {
+ return new ElasticsearchSink<>(bulkRequestsConfig, httpHosts, elasticsearchSinkFunction, failureHandler, restClientFactory);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Builder<?> builder = (Builder<?>) o;
+ return Objects.equals(httpHosts, builder.httpHosts) &&
+ Objects.equals(elasticsearchSinkFunction, builder.elasticsearchSinkFunction) &&
+ Objects.equals(bulkRequestsConfig, builder.bulkRequestsConfig) &&
+ Objects.equals(failureHandler, builder.failureHandler) &&
+ Objects.equals(restClientFactory, builder.restClientFactory);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ httpHosts,
+ elasticsearchSinkFunction,
+ bulkRequestsConfig,
+ failureHandler,
+ restClientFactory);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/RestClientFactory.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/RestClientFactory.java
new file mode 100644
index 0000000..6001f43
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/RestClientFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.streaming.connectors.elasticsearch7;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import org.elasticsearch.client.RestClientBuilder;
+
+import java.io.Serializable;
+
+/**
+ * A factory that is used to configure the {@link org.elasticsearch.client.RestHighLevelClient} internally
+ * used in the {@link ElasticsearchSink}.
+ */
+@PublicEvolving
+public interface RestClientFactory extends Serializable {
+
+ /**
+ * Configures the rest client builder.
+ *
+ * @param restClientBuilder the configured rest client builder.
+ */
+ void configureRestClientBuilder(RestClientBuilder restClientBuilder);
+
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/EmbeddedElasticsearchNodeEnvironmentImpl.java b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/EmbeddedElasticsearchNodeEnvironmentImpl.java
new file mode 100644
index 0000000..6f051a3
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/EmbeddedElasticsearchNodeEnvironmentImpl.java
@@ -0,0 +1,79 @@
+/*
+ * 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.streaming.connectors.elasticsearch;
+
+import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSinkITCase;
+
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.node.InternalSettingsPreparer;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.transport.Netty4Plugin;
+
+import java.io.File;
+import java.util.Collections;
+
+/**
+ * Implementation of {@link EmbeddedElasticsearchNodeEnvironment} for Elasticsearch 7.
+ * Will be dynamically loaded in {@link ElasticsearchSinkITCase} for integration tests.
+ */
+public class EmbeddedElasticsearchNodeEnvironmentImpl implements EmbeddedElasticsearchNodeEnvironment {
+
+ private Node node;
+
+ @Override
+ public void start(File tmpDataFolder, String clusterName) throws Exception {
+ if (node == null) {
+ Settings settings = Settings.builder()
+ .put("cluster.name", clusterName)
+ .put("http.cors.enabled", true)
+ .put("path.home", tmpDataFolder.getParent())
+ .put("path.data", tmpDataFolder.getAbsolutePath())
+ .build();
+
+ node = new PluginNode(settings);
+ node.start();
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (node != null && !node.isClosed()) {
+ node.close();
+ node = null;
+ }
+ }
+
+ @Override
+ public Client getClient() {
+ if (node != null && !node.isClosed()) {
+ return node.client();
+ } else {
+ return null;
+ }
+ }
+
+ private static class PluginNode extends Node {
+ public PluginNode(Settings settings) {
+ super(InternalSettingsPreparer.prepareEnvironment(settings, Collections.emptyMap(), null, () -> "node1"), Collections.<Class<? extends Plugin>>singletonList(Netty4Plugin.class), true);
+ }
+ }
+
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSinkITCase.java b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSinkITCase.java
new file mode 100644
index 0000000..c8de447
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSinkITCase.java
@@ -0,0 +1,100 @@
+/*
+ * 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.streaming.connectors.elasticsearch7;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkTestBase;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * IT cases for the {@link ElasticsearchSink}.
+ *
+ * <p>The Elasticsearch ITCases for 7.x CANNOT be executed in the IDE directly, since it is required that the
+ * Log4J-to-SLF4J adapter dependency must be excluded from the test classpath for the Elasticsearch embedded
+ * node used in the tests to work properly.
+ */
+public class ElasticsearchSinkITCase extends ElasticsearchSinkTestBase<RestHighLevelClient, HttpHost> {
+
+ @Test
+ public void testElasticsearchSink() throws Exception {
+ runElasticsearchSinkTest();
+ }
+
+ @Test
+ public void testNullAddresses() throws Exception {
+ runNullAddressesTest();
+ }
+
+ @Test
+ public void testEmptyAddresses() throws Exception {
+ runEmptyAddressesTest();
+ }
+
+ @Test
+ public void testInvalidElasticsearchCluster() throws Exception{
+ runInvalidElasticsearchClusterTest();
+ }
+
+ @Override
+ protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient> createElasticsearchSink(
+ int bulkFlushMaxActions,
+ String clusterName,
+ List<HttpHost> httpHosts,
+ ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction) {
+
+ ElasticsearchSink.Builder<Tuple2<Integer, String>> builder = new ElasticsearchSink.Builder<>(httpHosts, elasticsearchSinkFunction);
+ builder.setBulkFlushMaxActions(bulkFlushMaxActions);
+
+ return builder.build();
+ }
+
+ @Override
+ protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient> createElasticsearchSinkForEmbeddedNode(
+ int bulkFlushMaxActions,
+ String clusterName,
+ ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction) throws Exception {
+
+ return createElasticsearchSinkForNode(
+ bulkFlushMaxActions, clusterName, elasticsearchSinkFunction, "127.0.0.1");
+ }
+
+ @Override
+ protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient> createElasticsearchSinkForNode(
+ int bulkFlushMaxActions,
+ String clusterName,
+ ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction,
+ String ipAddress) throws Exception {
+
+ ArrayList<HttpHost> httpHosts = new ArrayList<>();
+ httpHosts.add(new HttpHost(ipAddress, 9200, "http"));
+
+ ElasticsearchSink.Builder<Tuple2<Integer, String>> builder = new ElasticsearchSink.Builder<>(httpHosts, elasticsearchSinkFunction);
+ builder.setBulkFlushMaxActions(bulkFlushMaxActions);
+
+ return builder.build();
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/test/resources/log4j-test.properties b/flink-connectors/flink-connector-elasticsearch7/src/test/resources/log4j-test.properties
new file mode 100644
index 0000000..fcd8654
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/test/resources/log4j-test.properties
@@ -0,0 +1,24 @@
+################################################################################
+# 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.
+################################################################################
+
+log4j.rootLogger=INFO, testlogger
+
+log4j.appender.testlogger=org.apache.log4j.ConsoleAppender
+log4j.appender.testlogger.target=System.err
+log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout
+log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-connectors/flink-sql-connector-elasticsearch7/pom.xml b/flink-connectors/flink-sql-connector-elasticsearch7/pom.xml
new file mode 100644
index 0000000..d1d8186
--- /dev/null
+++ b/flink-connectors/flink-sql-connector-elasticsearch7/pom.xml
@@ -0,0 +1,205 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connectors</artifactId>
+ <version>1.10-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-sql-connector-elasticsearch7_${scala.binary.version}</artifactId>
+ <name>flink-sql-connector-elasticsearch7</name>
+
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch-base_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <!-- Elasticsearch Java Client has been moved to a different module in 5.x -->
+ <exclusion>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Table ecosystem -->
+ <!-- Projects depending on this project won't depend on flink-table-*. -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch-base_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ </exclusion>
+ </exclusions>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Elasticsearch table sink factory testing -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Elasticsearch table descriptor testing -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadeTestJar>false</shadeTestJar>
+ <artifactSet>
+ <includes>
+ <include>*:*</include>
+ </includes>
+ <excludes>
+ <!-- These dependencies are not required. -->
+ <exclude>com.carrotsearch:hppc</exclude>
+ <exclude>com.tdunning:t-digest</exclude>
+ <exclude>joda-time:joda-time</exclude>
+ <exclude>net.sf.jopt-simple:jopt-simple</exclude>
+ <exclude>org.elasticsearch:jna</exclude>
+ <exclude>org.hdrhistogram:HdrHistogram</exclude>
+ <exclude>org.yaml:snakeyaml</exclude>
+ </excludes>
+ </artifactSet>
+ <filters>
+ <!-- Unless otherwise noticed these filters only serve to reduce the size of the resulting
+ jar by removing unnecessary files -->
+ <filter>
+ <artifact>org.elasticsearch:elasticsearch</artifact>
+ <excludes>
+ <exclude>config/**</exclude>
+ <exclude>modules.txt</exclude>
+ <exclude>plugins.txt</exclude>
+ <exclude>org/joda/**</exclude>
+ </excludes>
+ </filter>
+ <filter>
+ <artifact>org.elasticsearch.client:elasticsearch-rest-high-level-client</artifact>
+ <excludes>
+ <exclude>forbidden/**</exclude>
+ </excludes>
+ </filter>
+ <filter>
+ <artifact>org.apache.httpcomponents:httpclient</artifact>
+ <excludes>
+ <exclude>mozilla/**</exclude>
+ </excludes>
+ </filter>
+ <filter>
+ <artifact>org.apache.lucene:lucene-analyzers-common</artifact>
+ <excludes>
+ <exclude>org/tartarus/**</exclude>
+ </excludes>
+ </filter>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <!-- exclude Java 9 specific classes as otherwise the shade-plugin crashes -->
+ <exclude>META-INF/versions/**</exclude>
+ <exclude>META-INF/services/com.fasterxml.**</exclude>
+ <exclude>META-INF/services/org.apache.lucene.**</exclude>
+ <exclude>META-INF/services/org.elasticsearch.**</exclude>
+ <exclude>META-INF/LICENSE.txt</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <relocations>
+ <!-- Force relocation of all Elasticsearch dependencies. -->
+ <relocation>
+ <pattern>org.apache.commons</pattern>
+ <shadedPattern>org.apache.flink.elasticsearch7.shaded.org.apache.commons</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.http</pattern>
+ <shadedPattern>org.apache.flink.elasticsearch7.shaded.org.apache.http</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.lucene</pattern>
+ <shadedPattern>org.apache.flink.elasticsearch7.shaded.org.apache.lucene</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.elasticsearch</pattern>
+ <shadedPattern>org.apache.flink.elasticsearch7.shaded.org.elasticsearch</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.logging</pattern>
+ <shadedPattern>org.apache.flink.elasticsearch7.shaded.org.apache.logging</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.fasterxml.jackson</pattern>
+ <shadedPattern>org.apache.flink.elasticsearch7.shaded.com.fasterxml.jackson</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/flink-connectors/flink-sql-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSink.java b/flink-connectors/flink-sql-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSink.java
new file mode 100644
index 0000000..b6cf75c
--- /dev/null
+++ b/flink-connectors/flink-sql-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSink.java
@@ -0,0 +1,285 @@
+/*
+ * 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.streaming.sql.connectors.elasticsearch7;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
+import org.apache.flink.streaming.connectors.elasticsearch7.RestClientFactory;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.types.Row;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.common.xcontent.XContentType;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.BULK_FLUSH_BACKOFF_DELAY;
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.BULK_FLUSH_BACKOFF_ENABLED;
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.BULK_FLUSH_BACKOFF_RETRIES;
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.BULK_FLUSH_BACKOFF_TYPE;
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.BULK_FLUSH_INTERVAL;
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.BULK_FLUSH_MAX_ACTIONS;
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.BULK_FLUSH_MAX_SIZE;
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.DISABLE_FLUSH_ON_CHECKPOINT;
+import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption.REST_PATH_PREFIX;
+
+/**
+ * Version-specific upsert table sink for Elasticsearch 7.
+ */
+@Internal
+public class Elasticsearch7UpsertTableSink extends ElasticsearchUpsertTableSinkBase {
+
+ @VisibleForTesting
+ static final RequestFactory UPDATE_REQUEST_FACTORY =
+ new Elasticsearch7RequestFactory();
+
+ public Elasticsearch7UpsertTableSink(
+ boolean isAppendOnly,
+ TableSchema schema,
+ List<Host> hosts,
+ String index,
+ String keyDelimiter,
+ String keyNullLiteral,
+ SerializationSchema<Row> serializationSchema,
+ XContentType contentType,
+ ActionRequestFailureHandler failureHandler,
+ Map<SinkOption, String> sinkOptions) {
+
+ super(
+ isAppendOnly,
+ schema,
+ hosts,
+ index,
+ "",
+ keyDelimiter,
+ keyNullLiteral,
+ serializationSchema,
+ contentType,
+ failureHandler,
+ sinkOptions,
+ UPDATE_REQUEST_FACTORY);
+ }
+
+ @VisibleForTesting
+ Elasticsearch7UpsertTableSink(
+ boolean isAppendOnly,
+ TableSchema schema,
+ List<Host> hosts,
+ String index,
+ String docType,
+ String keyDelimiter,
+ String keyNullLiteral,
+ SerializationSchema<Row> serializationSchema,
+ XContentType contentType,
+ ActionRequestFailureHandler failureHandler,
+ Map<SinkOption, String> sinkOptions) {
+
+ super(
+ isAppendOnly,
+ schema,
+ hosts,
+ index,
+ docType,
+ keyDelimiter,
+ keyNullLiteral,
+ serializationSchema,
+ contentType,
+ failureHandler,
+ sinkOptions,
+ UPDATE_REQUEST_FACTORY);
+ }
+
+ @Override
+ protected ElasticsearchUpsertTableSinkBase copy(
+ boolean isAppendOnly,
+ TableSchema schema,
+ List<Host> hosts,
+ String index,
+ String docType,
+ String keyDelimiter,
+ String keyNullLiteral,
+ SerializationSchema<Row> serializationSchema,
+ XContentType contentType,
+ ActionRequestFailureHandler failureHandler,
+ Map<SinkOption, String> sinkOptions,
+ RequestFactory requestFactory) {
+
+ return new Elasticsearch7UpsertTableSink(
+ isAppendOnly,
+ schema,
+ hosts,
+ index,
+ keyDelimiter,
+ keyNullLiteral,
+ serializationSchema,
+ contentType,
+ failureHandler,
+ sinkOptions);
+ }
+
+ @Override
+ protected SinkFunction<Tuple2<Boolean, Row>> createSinkFunction(
+ List<Host> hosts,
+ ActionRequestFailureHandler failureHandler,
+ Map<SinkOption, String> sinkOptions,
+ ElasticsearchUpsertSinkFunction upsertSinkFunction) {
+
+ final List<HttpHost> httpHosts = hosts.stream()
+ .map((host) -> new HttpHost(host.hostname, host.port, host.protocol))
+ .collect(Collectors.toList());
+
+ final ElasticsearchSink.Builder<Tuple2<Boolean, Row>> builder = createBuilder(upsertSinkFunction, httpHosts);
+
+ builder.setFailureHandler(failureHandler);
+
+ Optional.ofNullable(sinkOptions.get(BULK_FLUSH_MAX_ACTIONS))
+ .ifPresent(v -> builder.setBulkFlushMaxActions(Integer.valueOf(v)));
+
+ Optional.ofNullable(sinkOptions.get(BULK_FLUSH_MAX_SIZE))
+ .ifPresent(v -> builder.setBulkFlushMaxSizeMb(MemorySize.parse(v).getMebiBytes()));
+
+ Optional.ofNullable(sinkOptions.get(BULK_FLUSH_INTERVAL))
+ .ifPresent(v -> builder.setBulkFlushInterval(Long.valueOf(v)));
+
+ Optional.ofNullable(sinkOptions.get(BULK_FLUSH_BACKOFF_ENABLED))
+ .ifPresent(v -> builder.setBulkFlushBackoff(Boolean.valueOf(v)));
+
+ Optional.ofNullable(sinkOptions.get(BULK_FLUSH_BACKOFF_TYPE))
+ .ifPresent(v -> builder.setBulkFlushBackoffType(ElasticsearchSinkBase.FlushBackoffType.valueOf(v)));
+
+ Optional.ofNullable(sinkOptions.get(BULK_FLUSH_BACKOFF_RETRIES))
+ .ifPresent(v -> builder.setBulkFlushBackoffRetries(Integer.valueOf(v)));
+
+ Optional.ofNullable(sinkOptions.get(BULK_FLUSH_BACKOFF_DELAY))
+ .ifPresent(v -> builder.setBulkFlushBackoffDelay(Long.valueOf(v)));
+
+ builder.setRestClientFactory(
+ new DefaultRestClientFactory(sinkOptions.get(REST_PATH_PREFIX)));
+
+ final ElasticsearchSink<Tuple2<Boolean, Row>> sink = builder.build();
+
+ Optional.ofNullable(sinkOptions.get(DISABLE_FLUSH_ON_CHECKPOINT))
+ .ifPresent(v -> {
+ if (Boolean.valueOf(v)) {
+ sink.disableFlushOnCheckpoint();
+ }
+ });
+
+ return sink;
+ }
+
+ @VisibleForTesting
+ ElasticsearchSink.Builder<Tuple2<Boolean, Row>> createBuilder(
+ ElasticsearchUpsertSinkFunction upsertSinkFunction,
+ List<HttpHost> httpHosts) {
+ return new ElasticsearchSink.Builder<>(httpHosts, upsertSinkFunction);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Helper classes
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Serializable {@link RestClientFactory} used by the sink.
+ */
+ @VisibleForTesting
+ static class DefaultRestClientFactory implements RestClientFactory {
+
+ private String pathPrefix;
+
+ public DefaultRestClientFactory(@Nullable String pathPrefix) {
+ this.pathPrefix = pathPrefix;
+ }
+
+ @Override
+ public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
+ if (pathPrefix != null) {
+ restClientBuilder.setPathPrefix(pathPrefix);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DefaultRestClientFactory that = (DefaultRestClientFactory) o;
+ return Objects.equals(pathPrefix, that.pathPrefix);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(pathPrefix);
+ }
+ }
+
+ /**
+ * Version-specific creation of {@link org.elasticsearch.action.ActionRequest}s used by the sink.
+ */
+ private static class Elasticsearch7RequestFactory implements RequestFactory {
+
+ @Override
+ public UpdateRequest createUpdateRequest(
+ String index,
+ String docType,
+ String key,
+ XContentType contentType,
+ byte[] document) {
+ return new UpdateRequest(index, key)
+ .doc(document, contentType)
+ .upsert(document, contentType);
+ }
+
+ @Override
+ public IndexRequest createIndexRequest(
+ String index,
+ String docType,
+ XContentType contentType,
+ byte[] document) {
+ return new IndexRequest(index)
+ .source(document, contentType);
+ }
+
+ @Override
+ public DeleteRequest createDeleteRequest(String index, String docType, String key) {
+ return new DeleteRequest(index, key);
+ }
+ }
+}
diff --git a/flink-connectors/flink-sql-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSinkFactory.java b/flink-connectors/flink-sql-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSinkFactory.java
new file mode 100644
index 0000000..ca2f544
--- /dev/null
+++ b/flink-connectors/flink-sql-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSinkFactory.java
@@ -0,0 +1,76 @@
+/*
+ * 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.streaming.sql.connectors.elasticsearch7;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.Host;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkFactoryBase;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sinks.UpsertStreamTableSink;
+import org.apache.flink.types.Row;
+
+import org.elasticsearch.common.xcontent.XContentType;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.table.descriptors.ElasticsearchValidator.CONNECTOR_VERSION_VALUE_7;
+
+/**
+ * Table factory for creating an {@link UpsertStreamTableSink} for Elasticsearch 7.
+ */
+@Internal
+public class Elasticsearch7UpsertTableSinkFactory extends ElasticsearchUpsertTableSinkFactoryBase {
+
+ @Override
+ protected String elasticsearchVersion() {
+ return CONNECTOR_VERSION_VALUE_7;
+ }
+
+ @Override
+ protected ElasticsearchUpsertTableSinkBase createElasticsearchUpsertTableSink(
+ boolean isAppendOnly,
+ TableSchema schema,
+ List<Host> hosts,
+ String index,
+ String docType,
+ String keyDelimiter,
+ String keyNullLiteral,
+ SerializationSchema<Row> serializationSchema,
+ XContentType contentType,
+ ActionRequestFailureHandler failureHandler,
+ Map<SinkOption, String> sinkOptions) {
+
+ return new Elasticsearch7UpsertTableSink(
+ isAppendOnly,
+ schema,
+ hosts,
+ index,
+ keyDelimiter,
+ keyNullLiteral,
+ serializationSchema,
+ contentType,
+ failureHandler,
+ sinkOptions);
+ }
+}
diff --git a/flink-connectors/flink-sql-connector-elasticsearch7/src/main/resources/META-INF/NOTICE b/flink-connectors/flink-sql-connector-elasticsearch7/src/main/resources/META-INF/NOTICE
new file mode 100644
index 0000000..2c7e1c0
--- /dev/null
+++ b/flink-connectors/flink-sql-connector-elasticsearch7/src/main/resources/META-INF/NOTICE
@@ -0,0 +1,45 @@
+flink-sql-connector-elasticsearch7
+Copyright 2014-2019 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
+
+- com.fasterxml.jackson.core:jackson-core:2.8.10
+- com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.8.10
+- com.fasterxml.jackson.dataformat:jackson-dataformat-smile:2.8.10
+- com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.8.10
+- commons-codec:commons-codec:1.10
+- commons-logging:commons-logging:1.1.3
+- org.apache.httpcomponents:httpasyncclient:4.1.2
+- org.apache.httpcomponents:httpclient:4.5.3
+- org.apache.httpcomponents:httpcore:4.4.6
+- org.apache.httpcomponents:httpcore-nio:4.4.5
+- org.apache.logging.log4j:log4j-api:2.9.1
+- org.apache.logging.log4j:log4j-to-slf4j:2.9.1
+- org.apache.lucene:lucene-analyzers-common:7.3.1
+- org.apache.lucene:lucene-backward-codecs:7.3.1
+- org.apache.lucene:lucene-core:7.3.1
+- org.apache.lucene:lucene-grouping:7.3.1
+- org.apache.lucene:lucene-highlighter:7.3.1
+- org.apache.lucene:lucene-join:7.3.1
+- org.apache.lucene:lucene-memory:7.3.1
+- org.apache.lucene:lucene-misc:7.3.1
+- org.apache.lucene:lucene-queries:7.3.1
+- org.apache.lucene:lucene-queryparser:7.3.1
+- org.apache.lucene:lucene-sandbox:7.3.1
+- org.apache.lucene:lucene-spatial:7.3.1
+- org.apache.lucene:lucene-spatial-extras:7.3.1
+- org.apache.lucene:lucene-spatial3d:7.3.1
+- org.apache.lucene:lucene-suggest:7.3.1
+- org.elasticsearch:elasticsearch:7.3.2
+- org.elasticsearch:elasticsearch-cli:7.3.2
+- org.elasticsearch:elasticsearch-core:7.3.2
+- org.elasticsearch:elasticsearch-secure-sm:7.3.2
+- org.elasticsearch:elasticsearch-x-content:7.3.2
+- org.elasticsearch.client:elasticsearch-rest-client:7.3.2
+- org.elasticsearch.client:elasticsearch-rest-high-level-client:7.3.2
+- org.elasticsearch.plugin:aggs-matrix-stats-client:7.3.2
+- org.elasticsearch.plugin:parent-join-client:7.3.2
+- org.elasticsearch.plugin:rank-eval-client:7.3.2
diff --git a/flink-connectors/flink-sql-connector-elasticsearch7/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory b/flink-connectors/flink-sql-connector-elasticsearch7/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory
new file mode 100644
index 0000000..fe10b08
--- /dev/null
+++ b/flink-connectors/flink-sql-connector-elasticsearch7/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.streaming.sql.connectors.elasticsearch7.Elasticsearch7UpsertTableSinkFactory
diff --git a/flink-connectors/flink-sql-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSinkFactoryTest.java b/flink-connectors/flink-sql-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSinkFactoryTest.java
new file mode 100644
index 0000000..a047b02
--- /dev/null
+++ b/flink-connectors/flink-sql-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/sql/connectors/elasticsearch7/Elasticsearch7UpsertTableSinkFactoryTest.java
@@ -0,0 +1,217 @@
+/*
+ * 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.streaming.sql.connectors.elasticsearch7;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.ElasticsearchUpsertSinkFunction;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.Host;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkBase.SinkOption;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchUpsertTableSinkFactoryTestBase;
+import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.types.Row;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.table.descriptors.ElasticsearchValidator.CONNECTOR_VERSION_VALUE_7;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link Elasticsearch7UpsertTableSink} created by {@link Elasticsearch7UpsertTableSinkFactory}.
+ */
+public class Elasticsearch7UpsertTableSinkFactoryTest extends ElasticsearchUpsertTableSinkFactoryTestBase {
+
+ @Test
+ public void testBuilder() {
+ final TableSchema schema = createTestSchema();
+
+ final TestElasticsearch7UpsertTableSink testSink = new TestElasticsearch7UpsertTableSink(
+ false,
+ schema,
+ Collections.singletonList(new Host(ElasticsearchUpsertTableSinkFactoryTestBase.HOSTNAME, ElasticsearchUpsertTableSinkFactoryTestBase.PORT, ElasticsearchUpsertTableSinkFactoryTestBase.SCHEMA)),
+ ElasticsearchUpsertTableSinkFactoryTestBase.INDEX,
+ ElasticsearchUpsertTableSinkFactoryTestBase.DOC_TYPE,
+ ElasticsearchUpsertTableSinkFactoryTestBase.KEY_DELIMITER,
+ ElasticsearchUpsertTableSinkFactoryTestBase.KEY_NULL_LITERAL,
+ new JsonRowSerializationSchema(schema.toRowType()),
+ XContentType.JSON,
+ new DummyFailureHandler(),
+ createTestSinkOptions());
+
+ final DataStreamMock dataStreamMock = new DataStreamMock(
+ new StreamExecutionEnvironmentMock(),
+ Types.TUPLE(Types.BOOLEAN, schema.toRowType()));
+
+ testSink.emitDataStream(dataStreamMock);
+
+ final ElasticsearchSink.Builder<Tuple2<Boolean, Row>> expectedBuilder = new ElasticsearchSink.Builder<>(
+ Collections.singletonList(new HttpHost(ElasticsearchUpsertTableSinkFactoryTestBase.HOSTNAME, ElasticsearchUpsertTableSinkFactoryTestBase.PORT, ElasticsearchUpsertTableSinkFactoryTestBase.SCHEMA)),
+ new ElasticsearchUpsertSinkFunction(
+ ElasticsearchUpsertTableSinkFactoryTestBase.INDEX,
+ ElasticsearchUpsertTableSinkFactoryTestBase.DOC_TYPE,
+ ElasticsearchUpsertTableSinkFactoryTestBase.KEY_DELIMITER,
+ ElasticsearchUpsertTableSinkFactoryTestBase.KEY_NULL_LITERAL,
+ new JsonRowSerializationSchema(schema.toRowType()),
+ XContentType.JSON,
+ Elasticsearch7UpsertTableSink.UPDATE_REQUEST_FACTORY,
+ new int[0]));
+ expectedBuilder.setFailureHandler(new DummyFailureHandler());
+ expectedBuilder.setBulkFlushBackoff(true);
+ expectedBuilder.setBulkFlushBackoffType(ElasticsearchSinkBase.FlushBackoffType.EXPONENTIAL);
+ expectedBuilder.setBulkFlushBackoffDelay(123);
+ expectedBuilder.setBulkFlushBackoffRetries(3);
+ expectedBuilder.setBulkFlushInterval(100);
+ expectedBuilder.setBulkFlushMaxActions(1000);
+ expectedBuilder.setBulkFlushMaxSizeMb(1);
+ expectedBuilder.setRestClientFactory(new Elasticsearch7UpsertTableSink.DefaultRestClientFactory("/myapp"));
+
+ assertEquals(expectedBuilder, testSink.builder);
+ }
+
+ @Override
+ protected String getElasticsearchVersion() {
+ return CONNECTOR_VERSION_VALUE_7;
+ }
+
+ @Override
+ protected ElasticsearchUpsertTableSinkBase getExpectedTableSink(
+ boolean isAppendOnly,
+ TableSchema schema,
+ List<Host> hosts,
+ String index,
+ String docType,
+ String keyDelimiter,
+ String keyNullLiteral,
+ SerializationSchema<Row> serializationSchema,
+ XContentType contentType,
+ ActionRequestFailureHandler failureHandler,
+ Map<SinkOption, String> sinkOptions) {
+ return new Elasticsearch7UpsertTableSink(
+ isAppendOnly,
+ schema,
+ hosts,
+ index,
+ keyDelimiter,
+ keyNullLiteral,
+ serializationSchema,
+ contentType,
+ failureHandler,
+ sinkOptions);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Helper classes
+ // --------------------------------------------------------------------------------------------
+
+ private static class TestElasticsearch7UpsertTableSink extends Elasticsearch7UpsertTableSink {
+
+ public ElasticsearchSink.Builder<Tuple2<Boolean, Row>> builder;
+
+ public TestElasticsearch7UpsertTableSink(
+ boolean isAppendOnly,
+ TableSchema schema,
+ List<Host> hosts,
+ String index,
+ String docType,
+ String keyDelimiter,
+ String keyNullLiteral,
+ SerializationSchema<Row> serializationSchema,
+ XContentType contentType,
+ ActionRequestFailureHandler failureHandler,
+ Map<SinkOption, String> sinkOptions) {
+
+ super(
+ isAppendOnly,
+ schema,
+ hosts,
+ index,
+ docType,
+ keyDelimiter,
+ keyNullLiteral,
+ serializationSchema,
+ contentType,
+ failureHandler,
+ sinkOptions);
+ }
+
+ @Override
+ protected ElasticsearchSink.Builder<Tuple2<Boolean, Row>> createBuilder(
+ ElasticsearchUpsertSinkFunction upsertSinkFunction,
+ List<HttpHost> httpHosts) {
+ builder = super.createBuilder(upsertSinkFunction, httpHosts);
+ return builder;
+ }
+ }
+
+ private static class StreamExecutionEnvironmentMock extends StreamExecutionEnvironment {
+
+ @Override
+ public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ private static class DataStreamMock extends DataStream<Tuple2<Boolean, Row>> {
+
+ public SinkFunction<?> sinkFunction;
+
+ public DataStreamMock(StreamExecutionEnvironment environment, TypeInformation<Tuple2<Boolean, Row>> outType) {
+ super(environment, new TransformationMock("name", outType, 1));
+ }
+
+ @Override
+ public DataStreamSink<Tuple2<Boolean, Row>> addSink(SinkFunction<Tuple2<Boolean, Row>> sinkFunction) {
+ this.sinkFunction = sinkFunction;
+ return super.addSink(sinkFunction);
+ }
+ }
+
+ private static class TransformationMock extends Transformation<Tuple2<Boolean, Row>> {
+
+ public TransformationMock(String name, TypeInformation<Tuple2<Boolean, Row>> outputType, int parallelism) {
+ super(name, outputType, parallelism);
+ }
+
+ @Override
+ public Collection<Transformation<?>> getTransitivePredecessors() {
+ return null;
+ }
+ }
+}
diff --git a/flink-connectors/flink-sql-connector-elasticsearch7/src/test/resources/log4j-test.properties b/flink-connectors/flink-sql-connector-elasticsearch7/src/test/resources/log4j-test.properties
new file mode 100644
index 0000000..fcd8654
--- /dev/null
+++ b/flink-connectors/flink-sql-connector-elasticsearch7/src/test/resources/log4j-test.properties
@@ -0,0 +1,24 @@
+################################################################################
+# 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.
+################################################################################
+
+log4j.rootLogger=INFO, testlogger
+
+log4j.appender.testlogger=org.apache.log4j.ConsoleAppender
+log4j.appender.testlogger.target=System.err
+log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout
+log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-connectors/pom.xml b/flink-connectors/pom.xml
index 41a5eaa..2c2b053 100644
--- a/flink-connectors/pom.xml
+++ b/flink-connectors/pom.xml
@@ -49,6 +49,7 @@ under the License.
<module>flink-connector-elasticsearch2</module>
<module>flink-connector-elasticsearch5</module>
<module>flink-connector-elasticsearch6</module>
+ <module>flink-connector-elasticsearch7</module>
<module>flink-connector-hive</module>
<module>flink-connector-rabbitmq</module>
<module>flink-connector-twitter</module>
@@ -58,6 +59,7 @@ under the License.
<module>flink-connector-kafka</module>
<module>flink-connector-gcp-pubsub</module>
<module>flink-connector-kinesis</module>
+ <module>flink-sql-connector-elasticsearch7</module>
</modules>
<!-- override these root dependencies as 'provided', so they don't end up
diff --git a/flink-end-to-end-tests/flink-elasticsearch7-test/pom.xml b/flink-end-to-end-tests/flink-elasticsearch7-test/pom.xml
new file mode 100644
index 0000000..76ddc0e
--- /dev/null
+++ b/flink-end-to-end-tests/flink-elasticsearch7-test/pom.xml
@@ -0,0 +1,92 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-end-to-end-tests</artifactId>
+ <version>1.10-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-elasticsearch7-test</artifactId>
+ <name>flink-elasticsearch7-test</name>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>Elasticsearch7SinkExample</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <finalName>Elasticsearch7SinkExample</finalName>
+ <artifactSet>
+ <excludes>
+ <exclude>com.google.code.findbugs:jsr305</exclude>
+ </excludes>
+ </artifactSet>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.flink.streaming.tests.Elasticsearch7SinkExample</mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/flink-end-to-end-tests/flink-elasticsearch7-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch7SinkExample.java b/flink-end-to-end-tests/flink-elasticsearch7-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch7SinkExample.java
new file mode 100644
index 0000000..e7036b5
--- /dev/null
+++ b/flink-end-to-end-tests/flink-elasticsearch7-test/src/main/java/org/apache/flink/streaming/tests/Elasticsearch7SinkExample.java
@@ -0,0 +1,150 @@
+/*
+ * 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.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.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
+import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
+import org.apache.flink.util.Collector;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.client.Requests;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * End to end test for Elasticsearch6Sink.
+ */
+public class Elasticsearch7SinkExample {
+
+ public static void main(String[] args) throws Exception {
+
+ final ParameterTool parameterTool = ParameterTool.fromArgs(args);
+
+ if (parameterTool.getNumberOfParameters() < 2) {
+ System.out.println("Missing parameters!\n" +
+ "Usage: --numRecords <numRecords> --index <index>");
+ return;
+ }
+
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(5000);
+
+ DataStream<Tuple2<String, String>> source = env.generateSequence(0, parameterTool.getInt("numRecords") - 1)
+ .flatMap(new FlatMapFunction<Long, Tuple2<String, String>>() {
+ @Override
+ public void flatMap(Long value, Collector<Tuple2<String, String>> out) {
+ final String key = String.valueOf(value);
+ final String message = "message #" + value;
+ out.collect(Tuple2.of(key, message + "update #1"));
+ out.collect(Tuple2.of(key, message + "update #2"));
+ }
+ });
+
+ List<HttpHost> httpHosts = new ArrayList<>();
+ httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
+
+ ElasticsearchSink.Builder<Tuple2<String, String>> esSinkBuilder = new ElasticsearchSink.Builder<>(
+ httpHosts,
+ (Tuple2<String, String> element, RuntimeContext ctx, RequestIndexer indexer) -> {
+ indexer.add(createIndexRequest(element.f1, parameterTool));
+ indexer.add(createUpdateRequest(element, parameterTool));
+ });
+
+ esSinkBuilder.setFailureHandler(
+ new CustomFailureHandler(parameterTool.getRequired("index")));
+
+ // this instructs the sink to emit after every element, otherwise they would be buffered
+ esSinkBuilder.setBulkFlushMaxActions(1);
+
+ source.addSink(esSinkBuilder.build());
+
+ env.execute("Elasticsearch 7.x end to end sink test example");
+ }
+
+ private static class CustomFailureHandler implements ActionRequestFailureHandler {
+
+ private static final long serialVersionUID = 942269087742453482L;
+
+ private final String index;
+
+ CustomFailureHandler(String index) {
+ this.index = index;
+ }
+
+ @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");
+ }
+ }
+ }
+
+ private static IndexRequest createIndexRequest(String element, ParameterTool parameterTool) {
+ Map<String, Object> json = new HashMap<>();
+ json.put("data", element);
+
+ String index;
+ String type;
+
+ if (element.startsWith("message #15")) {
+ index = ":intentional invalid index:";
+ type = ":intentional invalid type:";
+ } else {
+ index = parameterTool.getRequired("index");
+ }
+
+ return Requests.indexRequest()
+ .index(index)
+ .id(element)
+ .source(json);
+ }
+
+ private static UpdateRequest createUpdateRequest(Tuple2<String, String> element, ParameterTool parameterTool) {
+ Map<String, Object> json = new HashMap<>();
+ json.put("data", element.f1);
+
+ return new UpdateRequest(
+ parameterTool.getRequired("index"),
+ parameterTool.getRequired("type"),
+ element.f0)
+ .doc(json)
+ .upsert(json);
+ }
+}
diff --git a/flink-end-to-end-tests/pom.xml b/flink-end-to-end-tests/pom.xml
index b7ac551..bdf04b3 100644
--- a/flink-end-to-end-tests/pom.xml
+++ b/flink-end-to-end-tests/pom.xml
@@ -71,6 +71,7 @@ under the License.
<module>flink-plugins-test</module>
<module>flink-tpch-test</module>
<module>flink-streaming-kinesis-test</module>
+ <module>flink-elasticsearch7-test</module>
</modules>
<build>
diff --git a/tools/travis/stage.sh b/tools/travis/stage.sh
index 6676b46..b0f4456 100644
--- a/tools/travis/stage.sh
+++ b/tools/travis/stage.sh
@@ -89,7 +89,9 @@ flink-connectors/flink-connector-cassandra,\
flink-connectors/flink-connector-elasticsearch2,\
flink-connectors/flink-connector-elasticsearch5,\
flink-connectors/flink-connector-elasticsearch6,\
+flink-connectors/flink-connector-elasticsearch7,\
flink-connectors/flink-sql-connector-elasticsearch6,\
+flink-connectors/flink-sql-connector-elasticsearch7,\
flink-connectors/flink-connector-elasticsearch-base,\
flink-connectors/flink-connector-filesystem,\
flink-connectors/flink-connector-kafka-0.9,\