You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fp...@apache.org on 2022/03/22 08:35:17 UTC
[flink] 01/02: [FLINK-26638][connectors/elasticsearch] Revert Table-API implementation to SinkFunction-based one
This is an automated email from the ASF dual-hosted git repository.
fpaul pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 317c39fd84be6ee5f391a3b06c43f6b143c82f1f
Author: Alexander Preuß <11...@users.noreply.github.com>
AuthorDate: Thu Mar 17 11:51:22 2022 +0100
[FLINK-26638][connectors/elasticsearch] Revert Table-API implementation to SinkFunction-based one
---
.../table/AbstractTimeIndexGenerator.java | 41 +++
.../table/ElasticsearchConfiguration.java | 169 +++++++++
.../table/ElasticsearchConnectorOptions.java | 171 +++++++++
.../table/ElasticsearchValidationUtils.java | 94 +++++
.../elasticsearch/table/IndexGenerator.java | 39 ++
.../elasticsearch/table/IndexGeneratorBase.java | 52 +++
.../elasticsearch/table/IndexGeneratorFactory.java | 312 ++++++++++++++++
.../elasticsearch/table/KeyExtractor.java | 130 +++++++
.../elasticsearch/table/RequestFactory.java | 54 +++
.../table/RowElasticsearchSinkFunction.java | 140 ++++++++
.../elasticsearch/table/StaticIndexGenerator.java | 35 ++
.../table/IndexGeneratorFactoryTest.java | 282 +++++++++++++++
.../elasticsearch/table/KeyExtractorTest.java | 135 +++++++
.../elasticsearch/table/TestContext.java | 72 ++++
.../table/Elasticsearch6Configuration.java | 79 ++++
.../table/Elasticsearch6DynamicSink.java | 335 +++++++++++++++++
.../table/Elasticsearch6DynamicSinkFactory.java | 186 ++++++++++
.../org.apache.flink.table.factories.Factory | 2 +-
.../Elasticsearch6DynamicSinkFactoryTest.java | 250 +++++++++++++
.../table/Elasticsearch6DynamicSinkITCase.java | 399 +++++++++++++++++++++
.../table/Elasticsearch6DynamicSinkTest.java | 298 +++++++++++++++
.../table/Elasticsearch7Configuration.java | 70 ++++
.../table/Elasticsearch7DynamicSink.java | 335 +++++++++++++++++
.../table/Elasticsearch7DynamicSinkFactory.java | 186 ++++++++++
.../org.apache.flink.table.factories.Factory | 2 +-
.../Elasticsearch7DynamicSinkFactoryTest.java | 234 ++++++++++++
.../table/Elasticsearch7DynamicSinkITCase.java | 376 +++++++++++++++++++
.../table/Elasticsearch7DynamicSinkTest.java | 298 +++++++++++++++
28 files changed, 4774 insertions(+), 2 deletions(-)
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/AbstractTimeIndexGenerator.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/AbstractTimeIndexGenerator.java
new file mode 100644
index 0000000..6c22cf3
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/AbstractTimeIndexGenerator.java
@@ -0,0 +1,41 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+
+import java.time.format.DateTimeFormatter;
+
+/** Abstract class for time related {@link IndexGenerator}. */
+@Internal
+abstract class AbstractTimeIndexGenerator extends IndexGeneratorBase {
+
+ private final String dateTimeFormat;
+ protected transient DateTimeFormatter dateTimeFormatter;
+
+ public AbstractTimeIndexGenerator(String index, String dateTimeFormat) {
+ super(index);
+ this.dateTimeFormat = dateTimeFormat;
+ }
+
+ @Override
+ public void open() {
+ this.dateTimeFormatter = DateTimeFormatter.ofPattern(dateTimeFormat);
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchConfiguration.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchConfiguration.java
new file mode 100644
index 0000000..04c7633
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchConfiguration.java
@@ -0,0 +1,169 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.util.IgnoringFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.util.NoOpFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.util.RetryRejectedExecutionFailureHandler;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_TYPE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_INTERVAL_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.FAILURE_HANDLER_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.PASSWORD_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.USERNAME_OPTION;
+
+/** Accessor methods to elasticsearch options. */
+@Internal
+class ElasticsearchConfiguration {
+ protected final ReadableConfig config;
+ private final ClassLoader classLoader;
+
+ ElasticsearchConfiguration(ReadableConfig config, ClassLoader classLoader) {
+ this.config = config;
+ this.classLoader = classLoader;
+ }
+
+ public ActionRequestFailureHandler getFailureHandler() {
+ final ActionRequestFailureHandler failureHandler;
+ String value = config.get(FAILURE_HANDLER_OPTION);
+ switch (value.toUpperCase()) {
+ case "FAIL":
+ failureHandler = new NoOpFailureHandler();
+ break;
+ case "IGNORE":
+ failureHandler = new IgnoringFailureHandler();
+ break;
+ case "RETRY-REJECTED":
+ failureHandler = new RetryRejectedExecutionFailureHandler();
+ break;
+ default:
+ try {
+ Class<?> failureHandlerClass = Class.forName(value, false, classLoader);
+ failureHandler =
+ (ActionRequestFailureHandler)
+ InstantiationUtil.instantiate(failureHandlerClass);
+ } catch (ClassNotFoundException e) {
+ throw new ValidationException(
+ "Could not instantiate the failure handler class: " + value, e);
+ }
+ break;
+ }
+ return failureHandler;
+ }
+
+ public String getDocumentType() {
+ return config.get(ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION);
+ }
+
+ public int getBulkFlushMaxActions() {
+ int maxActions = config.get(ElasticsearchConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION);
+ // convert 0 to -1, because Elasticsearch client use -1 to disable this configuration.
+ return maxActions == 0 ? -1 : maxActions;
+ }
+
+ public long getBulkFlushMaxByteSize() {
+ long maxSize =
+ config.get(ElasticsearchConnectorOptions.BULK_FLASH_MAX_SIZE_OPTION).getBytes();
+ // convert 0 to -1, because Elasticsearch client use -1 to disable this configuration.
+ return maxSize == 0 ? -1 : maxSize;
+ }
+
+ public long getBulkFlushInterval() {
+ long interval = config.get(BULK_FLUSH_INTERVAL_OPTION).toMillis();
+ // convert 0 to -1, because Elasticsearch client use -1 to disable this configuration.
+ return interval == 0 ? -1 : interval;
+ }
+
+ public Optional<String> getUsername() {
+ return config.getOptional(USERNAME_OPTION);
+ }
+
+ public Optional<String> getPassword() {
+ return config.getOptional(PASSWORD_OPTION);
+ }
+
+ public boolean isBulkFlushBackoffEnabled() {
+ return config.get(BULK_FLUSH_BACKOFF_TYPE_OPTION)
+ != ElasticsearchConnectorOptions.BackOffType.DISABLED;
+ }
+
+ public Optional<ElasticsearchSinkBase.FlushBackoffType> getBulkFlushBackoffType() {
+ switch (config.get(BULK_FLUSH_BACKOFF_TYPE_OPTION)) {
+ case CONSTANT:
+ return Optional.of(ElasticsearchSinkBase.FlushBackoffType.CONSTANT);
+ case EXPONENTIAL:
+ return Optional.of(ElasticsearchSinkBase.FlushBackoffType.EXPONENTIAL);
+ default:
+ return Optional.empty();
+ }
+ }
+
+ public Optional<Integer> getBulkFlushBackoffRetries() {
+ return config.getOptional(BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION);
+ }
+
+ public Optional<Long> getBulkFlushBackoffDelay() {
+ return config.getOptional(BULK_FLUSH_BACKOFF_DELAY_OPTION).map(Duration::toMillis);
+ }
+
+ public boolean isDisableFlushOnCheckpoint() {
+ return !config.get(ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION);
+ }
+
+ public String getIndex() {
+ return config.get(ElasticsearchConnectorOptions.INDEX_OPTION);
+ }
+
+ public String getKeyDelimiter() {
+ return config.get(ElasticsearchConnectorOptions.KEY_DELIMITER_OPTION);
+ }
+
+ public Optional<String> getPathPrefix() {
+ return config.getOptional(ElasticsearchConnectorOptions.CONNECTION_PATH_PREFIX);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ElasticsearchConfiguration that = (ElasticsearchConfiguration) o;
+ return Objects.equals(config, that.config) && Objects.equals(classLoader, that.classLoader);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(config, classLoader);
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchConnectorOptions.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchConnectorOptions.java
new file mode 100644
index 0000000..1b0fc5e
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchConnectorOptions.java
@@ -0,0 +1,171 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Options for the Elasticsearch connector. */
+@PublicEvolving
+public class ElasticsearchConnectorOptions {
+
+ public static final ConfigOption<List<String>> HOSTS_OPTION =
+ ConfigOptions.key("hosts")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription("Elasticsearch hosts to connect to.");
+
+ public static final ConfigOption<String> INDEX_OPTION =
+ ConfigOptions.key("index")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Elasticsearch index for every record.");
+
+ public static final ConfigOption<String> DOCUMENT_TYPE_OPTION =
+ ConfigOptions.key("document-type")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Elasticsearch document type.");
+
+ public static final ConfigOption<String> PASSWORD_OPTION =
+ ConfigOptions.key("password")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Password used to connect to Elasticsearch instance.");
+
+ public static final ConfigOption<String> USERNAME_OPTION =
+ ConfigOptions.key("username")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Username used to connect to Elasticsearch instance.");
+
+ public static final ConfigOption<String> KEY_DELIMITER_OPTION =
+ ConfigOptions.key("document-id.key-delimiter")
+ .stringType()
+ .defaultValue("_")
+ .withDescription(
+ "Delimiter for composite keys e.g., \"$\" would result in IDs \"KEY1$KEY2$KEY3\".");
+
+ public static final ConfigOption<String> FAILURE_HANDLER_OPTION =
+ ConfigOptions.key("failure-handler")
+ .stringType()
+ .defaultValue("fail")
+ .withDescription(
+ Description.builder()
+ .text(
+ "Failure handling strategy in case a request to Elasticsearch fails")
+ .list(
+ text(
+ "\"fail\" (throws an exception if a request fails and thus causes a job failure)"),
+ text(
+ "\"ignore\" (ignores failures and drops the request)"),
+ text(
+ "\"retry-rejected\" (re-adds requests that have failed due to queue capacity saturation)"),
+ text(
+ "\"class name\" for failure handling with a ActionRequestFailureHandler subclass"))
+ .build());
+
+ public static final ConfigOption<Boolean> FLUSH_ON_CHECKPOINT_OPTION =
+ ConfigOptions.key("sink.flush-on-checkpoint")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription("Disables flushing on checkpoint");
+
+ public static final ConfigOption<Integer> BULK_FLUSH_MAX_ACTIONS_OPTION =
+ ConfigOptions.key("sink.bulk-flush.max-actions")
+ .intType()
+ .defaultValue(1000)
+ .withDescription("Maximum number of actions to buffer for each bulk request.");
+
+ public static final ConfigOption<MemorySize> BULK_FLASH_MAX_SIZE_OPTION =
+ ConfigOptions.key("sink.bulk-flush.max-size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("2mb"))
+ .withDescription("Maximum size of buffered actions per bulk request");
+
+ public static final ConfigOption<Duration> BULK_FLUSH_INTERVAL_OPTION =
+ ConfigOptions.key("sink.bulk-flush.interval")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(1))
+ .withDescription("Bulk flush interval");
+
+ public static final ConfigOption<BackOffType> BULK_FLUSH_BACKOFF_TYPE_OPTION =
+ ConfigOptions.key("sink.bulk-flush.backoff.strategy")
+ .enumType(BackOffType.class)
+ .defaultValue(BackOffType.DISABLED)
+ .withDescription("Backoff strategy");
+
+ public static final ConfigOption<Integer> BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION =
+ ConfigOptions.key("sink.bulk-flush.backoff.max-retries")
+ .intType()
+ .noDefaultValue()
+ .withDescription("Maximum number of retries.");
+
+ public static final ConfigOption<Duration> BULK_FLUSH_BACKOFF_DELAY_OPTION =
+ ConfigOptions.key("sink.bulk-flush.backoff.delay")
+ .durationType()
+ .noDefaultValue()
+ .withDescription("Delay between each backoff attempt.");
+
+ public static final ConfigOption<Duration> CONNECTION_MAX_RETRY_TIMEOUT_OPTION =
+ ConfigOptions.key("connection.max-retry-timeout")
+ .durationType()
+ .noDefaultValue()
+ .withDescription("Maximum timeout between retries.");
+
+ public static final ConfigOption<String> CONNECTION_PATH_PREFIX =
+ ConfigOptions.key("connection.path-prefix")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Prefix string to be added to every REST communication.");
+
+ public static final ConfigOption<String> FORMAT_OPTION =
+ ConfigOptions.key("format")
+ .stringType()
+ .defaultValue("json")
+ .withDescription(
+ "The format must produce a valid JSON document. "
+ + "Please refer to the documentation on formats for more details.");
+
+ // --------------------------------------------------------------------------------------------
+ // Enums
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Backoff strategy. Extends {@link ElasticsearchSinkBase.FlushBackoffType} with {@code
+ * DISABLED} option.
+ */
+ public enum BackOffType {
+ DISABLED,
+ CONSTANT,
+ EXPONENTIAL
+ }
+
+ private ElasticsearchConnectorOptions() {}
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchValidationUtils.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchValidationUtils.java
new file mode 100644
index 0000000..6452d00
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchValidationUtils.java
@@ -0,0 +1,94 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Utility methods for validating Elasticsearch properties. */
+@Internal
+class ElasticsearchValidationUtils {
+
+ private static final Set<LogicalTypeRoot> ILLEGAL_PRIMARY_KEY_TYPES = new LinkedHashSet<>();
+
+ static {
+ ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.ARRAY);
+ ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.MAP);
+ ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.MULTISET);
+ ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.STRUCTURED_TYPE);
+ ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.ROW);
+ ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.RAW);
+ ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.BINARY);
+ ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.VARBINARY);
+ }
+
+ /**
+ * Checks that the table does not have primary key defined on illegal types. In Elasticsearch
+ * the primary key is used to calculate the Elasticsearch document id, which is a string of up
+ * to 512 bytes. It cannot have whitespaces. As of now it is calculated by concatenating the
+ * fields. Certain types do not have a good string representation to be used in this scenario.
+ * The illegal types are mostly {@link LogicalTypeFamily#COLLECTION} types and {@link
+ * LogicalTypeRoot#RAW} type.
+ */
+ public static void validatePrimaryKey(TableSchema schema) {
+ schema.getPrimaryKey()
+ .ifPresent(
+ key -> {
+ List<LogicalTypeRoot> illegalTypes =
+ key.getColumns().stream()
+ .map(
+ fieldName -> {
+ LogicalType logicalType =
+ schema.getFieldDataType(fieldName)
+ .get()
+ .getLogicalType();
+ if (logicalType.is(
+ LogicalTypeRoot.DISTINCT_TYPE)) {
+ return ((DistinctType) logicalType)
+ .getSourceType()
+ .getTypeRoot();
+ } else {
+ return logicalType.getTypeRoot();
+ }
+ })
+ .filter(ILLEGAL_PRIMARY_KEY_TYPES::contains)
+ .collect(Collectors.toList());
+
+ if (!illegalTypes.isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "The table has a primary key on columns of illegal types: %s.\n"
+ + " Elasticsearch sink does not support primary keys on columns of types: %s.",
+ illegalTypes, ILLEGAL_PRIMARY_KEY_TYPES));
+ }
+ });
+ }
+
+ private ElasticsearchValidationUtils() {}
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGenerator.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGenerator.java
new file mode 100644
index 0000000..636f340
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGenerator.java
@@ -0,0 +1,39 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.Row;
+
+import java.io.Serializable;
+
+/** This interface is responsible to generate index name from given {@link Row} record. */
+@Internal
+interface IndexGenerator extends Serializable {
+
+ /**
+ * Initialize the index generator, this will be called only once before {@link
+ * #generate(RowData)} is called.
+ */
+ default void open() {}
+
+ /** Generate index name according the the given row. */
+ String generate(RowData row);
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorBase.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorBase.java
new file mode 100644
index 0000000..adfcaa4
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorBase.java
@@ -0,0 +1,52 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.Objects;
+
+/** Base class for {@link IndexGenerator}. */
+@Internal
+public abstract class IndexGeneratorBase implements IndexGenerator {
+
+ private static final long serialVersionUID = 1L;
+ protected final String index;
+
+ public IndexGeneratorBase(String index) {
+ this.index = index;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof IndexGeneratorBase)) {
+ return false;
+ }
+ IndexGeneratorBase that = (IndexGeneratorBase) o;
+ return index.equals(that.index);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(index);
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorFactory.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorFactory.java
new file mode 100644
index 0000000..48f0107
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorFactory.java
@@ -0,0 +1,312 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+
+import javax.annotation.Nonnull;
+
+import java.io.Serializable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Factory of {@link IndexGenerator}.
+ *
+ * <p>Flink supports both static index and dynamic index.
+ *
+ * <p>If you want to have a static index, this option value should be a plain string, e.g.
+ * 'myusers', all the records will be consistently written into "myusers" index.
+ *
+ * <p>If you want to have a dynamic index, you can use '{field_name}' to reference a field value in
+ * the record to dynamically generate a target index. You can also use
+ * '{field_name|date_format_string}' to convert a field value of TIMESTAMP/DATE/TIME type into the
+ * format specified by date_format_string. The date_format_string is compatible with {@link
+ * java.text.SimpleDateFormat}. For example, if the option value is 'myusers_{log_ts|yyyy-MM-dd}',
+ * then a record with log_ts field value 2020-03-27 12:25:55 will be written into
+ * "myusers_2020-03-27" index.
+ */
+@Internal
+final class IndexGeneratorFactory {
+
+ private IndexGeneratorFactory() {}
+
+ public static IndexGenerator createIndexGenerator(String index, TableSchema schema) {
+ return createIndexGenerator(index, schema, ZoneId.systemDefault());
+ }
+
+ public static IndexGenerator createIndexGenerator(
+ String index, TableSchema schema, ZoneId localTimeZoneId) {
+ final IndexHelper indexHelper = new IndexHelper();
+ if (indexHelper.checkIsDynamicIndex(index)) {
+ return createRuntimeIndexGenerator(
+ index,
+ schema.getFieldNames(),
+ schema.getFieldDataTypes(),
+ indexHelper,
+ localTimeZoneId);
+ } else {
+ return new StaticIndexGenerator(index);
+ }
+ }
+
+ interface DynamicFormatter extends Serializable {
+ String format(@Nonnull Object fieldValue, DateTimeFormatter formatter);
+ }
+
+ private static IndexGenerator createRuntimeIndexGenerator(
+ String index,
+ String[] fieldNames,
+ DataType[] fieldTypes,
+ IndexHelper indexHelper,
+ ZoneId localTimeZoneId) {
+ final String dynamicIndexPatternStr = indexHelper.extractDynamicIndexPatternStr(index);
+ final String indexPrefix = index.substring(0, index.indexOf(dynamicIndexPatternStr));
+ final String indexSuffix =
+ index.substring(indexPrefix.length() + dynamicIndexPatternStr.length());
+
+ if (indexHelper.checkIsDynamicIndexWithSystemTimeFormat(index)) {
+ final String dateTimeFormat =
+ indexHelper.extractDateFormat(
+ index, LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE);
+ return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
+ @Override
+ public String generate(RowData row) {
+ return indexPrefix
+ .concat(LocalDateTime.now(localTimeZoneId).format(dateTimeFormatter))
+ .concat(indexSuffix);
+ }
+ };
+ }
+
+ final boolean isDynamicIndexWithFormat = indexHelper.checkIsDynamicIndexWithFormat(index);
+ final int indexFieldPos =
+ indexHelper.extractIndexFieldPos(index, fieldNames, isDynamicIndexWithFormat);
+ final LogicalType indexFieldType = fieldTypes[indexFieldPos].getLogicalType();
+ final LogicalTypeRoot indexFieldLogicalTypeRoot = indexFieldType.getTypeRoot();
+
+ // validate index field type
+ indexHelper.validateIndexFieldType(indexFieldLogicalTypeRoot);
+
+ // time extract dynamic index pattern
+ final RowData.FieldGetter fieldGetter =
+ RowData.createFieldGetter(indexFieldType, indexFieldPos);
+
+ if (isDynamicIndexWithFormat) {
+ final String dateTimeFormat =
+ indexHelper.extractDateFormat(index, indexFieldLogicalTypeRoot);
+ DynamicFormatter formatFunction =
+ createFormatFunction(indexFieldType, indexFieldLogicalTypeRoot);
+
+ return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
+ @Override
+ public String generate(RowData row) {
+ Object fieldOrNull = fieldGetter.getFieldOrNull(row);
+ final String formattedField;
+ // TODO we can possibly optimize it to use the nullability of the field
+ if (fieldOrNull != null) {
+ formattedField = formatFunction.format(fieldOrNull, dateTimeFormatter);
+ } else {
+ formattedField = "null";
+ }
+ return indexPrefix.concat(formattedField).concat(indexSuffix);
+ }
+ };
+ }
+ // general dynamic index pattern
+ return new IndexGeneratorBase(index) {
+ @Override
+ public String generate(RowData row) {
+ Object indexField = fieldGetter.getFieldOrNull(row);
+ return indexPrefix
+ .concat(indexField == null ? "null" : indexField.toString())
+ .concat(indexSuffix);
+ }
+ };
+ }
+
+ private static DynamicFormatter createFormatFunction(
+ LogicalType indexFieldType, LogicalTypeRoot indexFieldLogicalTypeRoot) {
+ switch (indexFieldLogicalTypeRoot) {
+ case DATE:
+ return (value, dateTimeFormatter) -> {
+ Integer indexField = (Integer) value;
+ return LocalDate.ofEpochDay(indexField).format(dateTimeFormatter);
+ };
+ case TIME_WITHOUT_TIME_ZONE:
+ return (value, dateTimeFormatter) -> {
+ Integer indexField = (Integer) value;
+ return LocalTime.ofNanoOfDay(indexField * 1_000_000L).format(dateTimeFormatter);
+ };
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return (value, dateTimeFormatter) -> {
+ TimestampData indexField = (TimestampData) value;
+ return indexField.toLocalDateTime().format(dateTimeFormatter);
+ };
+ case TIMESTAMP_WITH_TIME_ZONE:
+ throw new UnsupportedOperationException(
+ "TIMESTAMP_WITH_TIME_ZONE is not supported yet");
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return (value, dateTimeFormatter) -> {
+ TimestampData indexField = (TimestampData) value;
+ return indexField.toInstant().atZone(ZoneOffset.UTC).format(dateTimeFormatter);
+ };
+ default:
+ throw new TableException(
+ String.format(
+ "Unsupported type '%s' found in Elasticsearch dynamic index field, "
+ + "time-related pattern only support types are: DATE,TIME,TIMESTAMP.",
+ indexFieldType));
+ }
+ }
+
+ /**
+ * Helper class for {@link IndexGeneratorFactory}, this helper can use to validate index field
+ * type ans parse index format from pattern.
+ */
+ static class IndexHelper {
+ private static final Pattern dynamicIndexPattern = Pattern.compile("\\{[^\\{\\}]+\\}?");
+ private static final Pattern dynamicIndexTimeExtractPattern =
+ Pattern.compile(".*\\{.+\\|.*\\}.*");
+ private static final Pattern dynamicIndexSystemTimeExtractPattern =
+ Pattern.compile(
+ ".*\\{\\s*(now\\(\\s*\\)|NOW\\(\\s*\\)|current_timestamp|CURRENT_TIMESTAMP)\\s*\\|.*\\}.*");
+ private static final List<LogicalTypeRoot> supportedTypes = new ArrayList<>();
+ private static final Map<LogicalTypeRoot, String> defaultFormats = new HashMap<>();
+
+ static {
+ // time related types
+ supportedTypes.add(LogicalTypeRoot.DATE);
+ supportedTypes.add(LogicalTypeRoot.TIME_WITHOUT_TIME_ZONE);
+ supportedTypes.add(LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE);
+ supportedTypes.add(LogicalTypeRoot.TIMESTAMP_WITH_TIME_ZONE);
+ supportedTypes.add(LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE);
+ // general types
+ supportedTypes.add(LogicalTypeRoot.VARCHAR);
+ supportedTypes.add(LogicalTypeRoot.CHAR);
+ supportedTypes.add(LogicalTypeRoot.TINYINT);
+ supportedTypes.add(LogicalTypeRoot.INTEGER);
+ supportedTypes.add(LogicalTypeRoot.BIGINT);
+ }
+
+ static {
+ defaultFormats.put(LogicalTypeRoot.DATE, "yyyy_MM_dd");
+ defaultFormats.put(LogicalTypeRoot.TIME_WITHOUT_TIME_ZONE, "HH_mm_ss");
+ defaultFormats.put(LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE, "yyyy_MM_dd_HH_mm_ss");
+ defaultFormats.put(LogicalTypeRoot.TIMESTAMP_WITH_TIME_ZONE, "yyyy_MM_dd_HH_mm_ss");
+ defaultFormats.put(
+ LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE, "yyyy_MM_dd_HH_mm_ssX");
+ }
+
+ /** Validate the index field Type. */
+ void validateIndexFieldType(LogicalTypeRoot logicalType) {
+ if (!supportedTypes.contains(logicalType)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported type %s of index field, " + "Supported types are: %s",
+ logicalType, supportedTypes));
+ }
+ }
+
+ /** Get the default date format. */
+ String getDefaultFormat(LogicalTypeRoot logicalType) {
+ return defaultFormats.get(logicalType);
+ }
+
+ /** Check general dynamic index is enabled or not by index pattern. */
+ boolean checkIsDynamicIndex(String index) {
+ final Matcher matcher = dynamicIndexPattern.matcher(index);
+ int count = 0;
+ while (matcher.find()) {
+ count++;
+ }
+ if (count > 1) {
+ throw new TableException(
+ String.format(
+ "Chaining dynamic index pattern %s is not supported,"
+ + " only support single dynamic index pattern.",
+ index));
+ }
+ return count == 1;
+ }
+
+ /** Check time extract dynamic index is enabled or not by index pattern. */
+ boolean checkIsDynamicIndexWithFormat(String index) {
+ return dynamicIndexTimeExtractPattern.matcher(index).matches();
+ }
+
+ /** Check generate dynamic index is from system time or not. */
+ boolean checkIsDynamicIndexWithSystemTimeFormat(String index) {
+ return dynamicIndexSystemTimeExtractPattern.matcher(index).matches();
+ }
+
+ /** Extract dynamic index pattern string from index pattern string. */
+ String extractDynamicIndexPatternStr(String index) {
+ int start = index.indexOf("{");
+ int end = index.lastIndexOf("}");
+ return index.substring(start, end + 1);
+ }
+
+ /** Extract index field position in a fieldNames, return the field position. */
+ int extractIndexFieldPos(
+ String index, String[] fieldNames, boolean isDynamicIndexWithFormat) {
+ List<String> fieldList = Arrays.asList(fieldNames);
+ String indexFieldName;
+ if (isDynamicIndexWithFormat) {
+ indexFieldName = index.substring(index.indexOf("{") + 1, index.indexOf("|"));
+ } else {
+ indexFieldName = index.substring(index.indexOf("{") + 1, index.indexOf("}"));
+ }
+ if (!fieldList.contains(indexFieldName)) {
+ throw new TableException(
+ String.format(
+ "Unknown field '%s' in index pattern '%s', please check the field name.",
+ indexFieldName, index));
+ }
+ return fieldList.indexOf(indexFieldName);
+ }
+
+ /** Extract dateTime format by the date format that extracted from index pattern string. */
+ private String extractDateFormat(String index, LogicalTypeRoot logicalType) {
+ String format = index.substring(index.indexOf("|") + 1, index.indexOf("}"));
+ if ("".equals(format)) {
+ format = getDefaultFormat(logicalType);
+ }
+ return format;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractor.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractor.java
new file mode 100644
index 0000000..ae7c522
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractor.java
@@ -0,0 +1,130 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableColumn;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.LocalTime;
+import java.time.Period;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+/** An extractor for a Elasticsearch key from a {@link RowData}. */
+@Internal
+class KeyExtractor implements Function<RowData, String>, Serializable {
+ private final FieldFormatter[] fieldFormatters;
+ private final String keyDelimiter;
+
+ private interface FieldFormatter extends Serializable {
+ String format(RowData rowData);
+ }
+
+ private KeyExtractor(FieldFormatter[] fieldFormatters, String keyDelimiter) {
+ this.fieldFormatters = fieldFormatters;
+ this.keyDelimiter = keyDelimiter;
+ }
+
+ @Override
+ public String apply(RowData rowData) {
+ final StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < fieldFormatters.length; i++) {
+ if (i > 0) {
+ builder.append(keyDelimiter);
+ }
+ final String value = fieldFormatters[i].format(rowData);
+ builder.append(value);
+ }
+ return builder.toString();
+ }
+
+ private static class ColumnWithIndex {
+ public TableColumn column;
+ public int index;
+
+ public ColumnWithIndex(TableColumn column, int index) {
+ this.column = column;
+ this.index = index;
+ }
+
+ public LogicalType getType() {
+ return column.getType().getLogicalType();
+ }
+
+ public int getIndex() {
+ return index;
+ }
+ }
+
+ public static Function<RowData, String> createKeyExtractor(
+ TableSchema schema, String keyDelimiter) {
+ return schema.getPrimaryKey()
+ .map(
+ key -> {
+ Map<String, ColumnWithIndex> namesToColumns = new HashMap<>();
+ List<TableColumn> tableColumns = schema.getTableColumns();
+ for (int i = 0; i < schema.getFieldCount(); i++) {
+ TableColumn column = tableColumns.get(i);
+ namesToColumns.put(
+ column.getName(), new ColumnWithIndex(column, i));
+ }
+
+ FieldFormatter[] fieldFormatters =
+ key.getColumns().stream()
+ .map(namesToColumns::get)
+ .map(
+ column ->
+ toFormatter(
+ column.index, column.getType()))
+ .toArray(FieldFormatter[]::new);
+
+ return (Function<RowData, String>)
+ new KeyExtractor(fieldFormatters, keyDelimiter);
+ })
+ .orElseGet(() -> (Function<RowData, String> & Serializable) (row) -> null);
+ }
+
+ private static FieldFormatter toFormatter(int index, LogicalType type) {
+ switch (type.getTypeRoot()) {
+ case DATE:
+ return (row) -> LocalDate.ofEpochDay(row.getInt(index)).toString();
+ case TIME_WITHOUT_TIME_ZONE:
+ return (row) ->
+ LocalTime.ofNanoOfDay((long) row.getInt(index) * 1_000_000L).toString();
+ case INTERVAL_YEAR_MONTH:
+ return (row) -> Period.ofDays(row.getInt(index)).toString();
+ case INTERVAL_DAY_TIME:
+ return (row) -> Duration.ofMillis(row.getLong(index)).toString();
+ case DISTINCT_TYPE:
+ return toFormatter(index, ((DistinctType) type).getSourceType());
+ default:
+ RowData.FieldGetter fieldGetter = RowData.createFieldGetter(type, index);
+ return (row) -> fieldGetter.getFieldOrNull(row).toString();
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RequestFactory.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RequestFactory.java
new file mode 100644
index 0000000..f5b2418
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RequestFactory.java
@@ -0,0 +1,54 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.common.xcontent.XContentType;
+
+import java.io.Serializable;
+
+/** For version-agnostic creating of {@link ActionRequest}s. */
+@Internal
+interface RequestFactory extends Serializable {
+ /**
+ * 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, String docType, String key, XContentType contentType, byte[] document);
+
+ /**
+ * 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, String docType, String key, XContentType contentType, byte[] document);
+
+ /**
+ * 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, String docType, String key);
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java
new file mode 100644
index 0000000..7fb1e4d
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java
@@ -0,0 +1,140 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
+import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Preconditions;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.common.xcontent.XContentType;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+import java.util.function.Function;
+
+/** Sink function for converting upserts into Elasticsearch {@link ActionRequest}s. */
+@Internal
+class RowElasticsearchSinkFunction implements ElasticsearchSinkFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final IndexGenerator indexGenerator;
+ private final String docType;
+ private final SerializationSchema<RowData> serializationSchema;
+ private final XContentType contentType;
+ private final RequestFactory requestFactory;
+ private final Function<RowData, String> createKey;
+
+ public RowElasticsearchSinkFunction(
+ IndexGenerator indexGenerator,
+ @Nullable String docType, // this is deprecated in es 7+
+ SerializationSchema<RowData> serializationSchema,
+ XContentType contentType,
+ RequestFactory requestFactory,
+ Function<RowData, String> createKey) {
+ this.indexGenerator = Preconditions.checkNotNull(indexGenerator);
+ this.docType = docType;
+ this.serializationSchema = Preconditions.checkNotNull(serializationSchema);
+ this.contentType = Preconditions.checkNotNull(contentType);
+ this.requestFactory = Preconditions.checkNotNull(requestFactory);
+ this.createKey = Preconditions.checkNotNull(createKey);
+ }
+
+ @Override
+ public void open() {
+ indexGenerator.open();
+ }
+
+ @Override
+ public void process(RowData element, RuntimeContext ctx, RequestIndexer indexer) {
+ switch (element.getRowKind()) {
+ case INSERT:
+ case UPDATE_AFTER:
+ processUpsert(element, indexer);
+ break;
+ case UPDATE_BEFORE:
+ case DELETE:
+ processDelete(element, indexer);
+ break;
+ default:
+ throw new TableException("Unsupported message kind: " + element.getRowKind());
+ }
+ }
+
+ private void processUpsert(RowData row, RequestIndexer indexer) {
+ final byte[] document = serializationSchema.serialize(row);
+ final String key = createKey.apply(row);
+ if (key != null) {
+ final UpdateRequest updateRequest =
+ requestFactory.createUpdateRequest(
+ indexGenerator.generate(row), docType, key, contentType, document);
+ indexer.add(updateRequest);
+ } else {
+ final IndexRequest indexRequest =
+ requestFactory.createIndexRequest(
+ indexGenerator.generate(row), docType, key, contentType, document);
+ indexer.add(indexRequest);
+ }
+ }
+
+ private void processDelete(RowData row, RequestIndexer indexer) {
+ final String key = createKey.apply(row);
+ final DeleteRequest deleteRequest =
+ requestFactory.createDeleteRequest(indexGenerator.generate(row), docType, key);
+ indexer.add(deleteRequest);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ RowElasticsearchSinkFunction that = (RowElasticsearchSinkFunction) o;
+ return Objects.equals(indexGenerator, that.indexGenerator)
+ && Objects.equals(docType, that.docType)
+ && Objects.equals(serializationSchema, that.serializationSchema)
+ && contentType == that.contentType
+ && Objects.equals(requestFactory, that.requestFactory)
+ && Objects.equals(createKey, that.createKey);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ indexGenerator,
+ docType,
+ serializationSchema,
+ contentType,
+ requestFactory,
+ createKey);
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/StaticIndexGenerator.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/StaticIndexGenerator.java
new file mode 100644
index 0000000..1ffcac4
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/StaticIndexGenerator.java
@@ -0,0 +1,35 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.RowData;
+
+/** A static {@link IndexGenerator} which generate fixed index name. */
+@Internal
+final class StaticIndexGenerator extends IndexGeneratorBase {
+
+ public StaticIndexGenerator(String index) {
+ super(index);
+ }
+
+ public String generate(RowData row) {
+ return index;
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorFactoryTest.java b/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorFactoryTest.java
new file mode 100644
index 0000000..a5f7759
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorFactoryTest.java
@@ -0,0 +1,282 @@
+/*
+ * 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.table;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.UnsupportedTemporalTypeException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/** Tests for {@link IndexGeneratorFactory}. */
+public class IndexGeneratorFactoryTest extends TestLogger {
+
+ private TableSchema schema;
+ private List<RowData> rows;
+
+ @Before
+ public void prepareData() {
+ schema =
+ new TableSchema.Builder()
+ .field("id", DataTypes.INT())
+ .field("item", DataTypes.STRING())
+ .field("log_ts", DataTypes.BIGINT())
+ .field("log_date", DataTypes.DATE())
+ .field("log_time", DataTypes.TIME())
+ .field("order_timestamp", DataTypes.TIMESTAMP())
+ .field("local_timestamp", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .field("status", DataTypes.BOOLEAN())
+ .build();
+
+ rows = new ArrayList<>();
+ rows.add(
+ GenericRowData.of(
+ 1,
+ StringData.fromString("apple"),
+ Timestamp.valueOf("2020-03-18 12:12:14").getTime(),
+ (int) LocalDate.parse("2020-03-18").toEpochDay(),
+ (int) (LocalTime.parse("12:12:14").toNanoOfDay() / 1_000_000L),
+ TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-03-18T12:12:14")),
+ TimestampData.fromInstant(Instant.parse("2020-03-18T12:12:14Z")),
+ true));
+ rows.add(
+ GenericRowData.of(
+ 2,
+ StringData.fromString("peanut"),
+ Timestamp.valueOf("2020-03-19 12:12:14").getTime(),
+ (int) LocalDate.parse("2020-03-19").toEpochDay(),
+ (int) (LocalTime.parse("12:22:21").toNanoOfDay() / 1_000_000L),
+ TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-03-19T12:22:14")),
+ TimestampData.fromInstant(Instant.parse("2020-03-19T12:12:14Z")),
+ false));
+ }
+
+ @Test
+ public void testDynamicIndexFromTimestamp() {
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator(
+ "{order_timestamp|yyyy_MM_dd_HH-ss}_index", schema);
+ indexGenerator.open();
+ Assert.assertEquals("2020_03_18_12-14_index", indexGenerator.generate(rows.get(0)));
+ IndexGenerator indexGenerator1 =
+ IndexGeneratorFactory.createIndexGenerator(
+ "{order_timestamp|yyyy_MM_dd_HH_mm}_index", schema);
+ indexGenerator1.open();
+ Assert.assertEquals("2020_03_19_12_22_index", indexGenerator1.generate(rows.get(1)));
+ }
+
+ @Test
+ public void testDynamicIndexFromDate() {
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator(
+ "my-index-{log_date|yyyy/MM/dd}", schema);
+ indexGenerator.open();
+ Assert.assertEquals("my-index-2020/03/18", indexGenerator.generate(rows.get(0)));
+ Assert.assertEquals("my-index-2020/03/19", indexGenerator.generate(rows.get(1)));
+ }
+
+ @Test
+ public void testDynamicIndexFromTime() {
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator("my-index-{log_time|HH-mm}", schema);
+ indexGenerator.open();
+ Assert.assertEquals("my-index-12-12", indexGenerator.generate(rows.get(0)));
+ Assert.assertEquals("my-index-12-22", indexGenerator.generate(rows.get(1)));
+ }
+
+ @Test
+ public void testDynamicIndexDefaultFormat() {
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator("my-index-{log_time|}", schema);
+ indexGenerator.open();
+ Assert.assertEquals("my-index-12_12_14", indexGenerator.generate(rows.get(0)));
+ Assert.assertEquals("my-index-12_22_21", indexGenerator.generate(rows.get(1)));
+ }
+
+ @Test
+ public void testDynamicIndexFromSystemTime() {
+ List<String> supportedUseCases =
+ Arrays.asList(
+ "now()",
+ "NOW()",
+ "now( )",
+ "NOW(\t)",
+ "\t NOW( ) \t",
+ "current_timestamp",
+ "CURRENT_TIMESTAMP",
+ "\tcurrent_timestamp\t",
+ " current_timestamp ");
+
+ supportedUseCases.stream()
+ .forEach(
+ f -> {
+ DateTimeFormatter dateTimeFormatter =
+ DateTimeFormatter.ofPattern("yyyy_MM_dd");
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator(
+ String.format("my-index-{%s|yyyy_MM_dd}", f), schema);
+ indexGenerator.open();
+ // The date may change during the running of the unit test.
+ // Generate expected index-name based on the current time
+ // before and after calling the generate method.
+ String expectedIndex1 =
+ "my-index-" + LocalDateTime.now().format(dateTimeFormatter);
+ String actualIndex = indexGenerator.generate(rows.get(1));
+ String expectedIndex2 =
+ "my-index-" + LocalDateTime.now().format(dateTimeFormatter);
+ Assert.assertTrue(
+ actualIndex.equals(expectedIndex1)
+ || actualIndex.equals(expectedIndex2));
+ });
+
+ List<String> invalidUseCases =
+ Arrays.asList(
+ "now",
+ "now(",
+ "NOW",
+ "NOW)",
+ "current_timestamp()",
+ "CURRENT_TIMESTAMP()",
+ "CURRENT_timestamp");
+ invalidUseCases.stream()
+ .forEach(
+ f -> {
+ String expectedExceptionMsg =
+ String.format(
+ "Unknown field '%s' in index pattern 'my-index-{%s|yyyy_MM_dd}',"
+ + " please check the field name.",
+ f, f);
+ try {
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator(
+ String.format("my-index-{%s|yyyy_MM_dd}", f),
+ schema);
+ indexGenerator.open();
+ } catch (TableException e) {
+ Assert.assertEquals(expectedExceptionMsg, e.getMessage());
+ }
+ });
+ }
+
+ @Test
+ public void testDynamicIndexDefaultFormatTimestampWithLocalTimeZone() {
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator("my-index-{local_timestamp|}", schema);
+ indexGenerator.open();
+ Assert.assertEquals("my-index-2020_03_18_12_12_14Z", indexGenerator.generate(rows.get(0)));
+ Assert.assertEquals("my-index-2020_03_19_12_12_14Z", indexGenerator.generate(rows.get(1)));
+ }
+
+ @Test
+ public void testGeneralDynamicIndex() {
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator("index_{item}", schema);
+ indexGenerator.open();
+ Assert.assertEquals("index_apple", indexGenerator.generate(rows.get(0)));
+ Assert.assertEquals("index_peanut", indexGenerator.generate(rows.get(1)));
+ }
+
+ @Test
+ public void testStaticIndex() {
+ IndexGenerator indexGenerator =
+ IndexGeneratorFactory.createIndexGenerator("my-index", schema);
+ indexGenerator.open();
+ Assert.assertEquals("my-index", indexGenerator.generate(rows.get(0)));
+ Assert.assertEquals("my-index", indexGenerator.generate(rows.get(1)));
+ }
+
+ @Test
+ public void testUnknownField() {
+ String expectedExceptionMsg =
+ "Unknown field 'unknown_ts' in index pattern 'my-index-{unknown_ts|yyyy-MM-dd}',"
+ + " please check the field name.";
+ try {
+ IndexGeneratorFactory.createIndexGenerator("my-index-{unknown_ts|yyyy-MM-dd}", schema);
+ } catch (TableException e) {
+ Assert.assertEquals(e.getMessage(), expectedExceptionMsg);
+ }
+ }
+
+ @Test
+ public void testUnsupportedTimeType() {
+ String expectedExceptionMsg =
+ "Unsupported type 'INT' found in Elasticsearch dynamic index field, "
+ + "time-related pattern only support types are: DATE,TIME,TIMESTAMP.";
+ try {
+ IndexGeneratorFactory.createIndexGenerator("my-index-{id|yyyy-MM-dd}", schema);
+ } catch (TableException e) {
+ Assert.assertEquals(expectedExceptionMsg, e.getMessage());
+ }
+ }
+
+ @Test
+ public void testUnsupportedMultiParametersType() {
+ String expectedExceptionMsg =
+ "Chaining dynamic index pattern my-index-{local_date}-{local_time} is not supported,"
+ + " only support single dynamic index pattern.";
+ try {
+ IndexGeneratorFactory.createIndexGenerator(
+ "my-index-{local_date}-{local_time}", schema);
+ } catch (TableException e) {
+ Assert.assertEquals(expectedExceptionMsg, e.getMessage());
+ }
+ }
+
+ @Test
+ public void testDynamicIndexUnsupportedFormat() {
+ String expectedExceptionMsg = "Unsupported field: HourOfDay";
+ try {
+ IndexGeneratorFactory.createIndexGenerator(
+ "my-index-{log_date|yyyy/MM/dd HH:mm}", schema);
+ } catch (UnsupportedTemporalTypeException e) {
+ Assert.assertEquals(expectedExceptionMsg, e.getMessage());
+ }
+ }
+
+ @Test
+ public void testUnsupportedIndexFieldType() {
+ String expectedExceptionMsg =
+ "Unsupported type BOOLEAN of index field, Supported types are:"
+ + " [DATE, TIME_WITHOUT_TIME_ZONE, TIMESTAMP_WITHOUT_TIME_ZONE, TIMESTAMP_WITH_TIME_ZONE,"
+ + " TIMESTAMP_WITH_LOCAL_TIME_ZONE, VARCHAR, CHAR, TINYINT, INTEGER, BIGINT]";
+ try {
+ IndexGeneratorFactory.createIndexGenerator("index_{status}", schema);
+ } catch (IllegalArgumentException e) {
+ Assert.assertEquals(expectedExceptionMsg, e.getMessage());
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractorTest.java b/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractorTest.java
new file mode 100644
index 0000000..bcfb68d
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractorTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.table;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+
+import org.junit.Test;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.function.Function;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.junit.Assert.assertThat;
+
+/** Tests for {@link KeyExtractor}. */
+public class KeyExtractorTest {
+ @Test
+ public void testSimpleKey() {
+ TableSchema schema =
+ TableSchema.builder()
+ .field("a", DataTypes.BIGINT().notNull())
+ .field("b", DataTypes.STRING())
+ .primaryKey("a")
+ .build();
+
+ Function<RowData, String> keyExtractor = KeyExtractor.createKeyExtractor(schema, "_");
+
+ String key = keyExtractor.apply(GenericRowData.of(12L, StringData.fromString("ABCD")));
+ assertThat(key, equalTo("12"));
+ }
+
+ @Test
+ public void testNoPrimaryKey() {
+ TableSchema schema =
+ TableSchema.builder()
+ .field("a", DataTypes.BIGINT().notNull())
+ .field("b", DataTypes.STRING())
+ .build();
+
+ Function<RowData, String> keyExtractor = KeyExtractor.createKeyExtractor(schema, "_");
+
+ String key = keyExtractor.apply(GenericRowData.of(12L, StringData.fromString("ABCD")));
+ assertThat(key, nullValue());
+ }
+
+ @Test
+ public void testTwoFieldsKey() {
+ TableSchema schema =
+ TableSchema.builder()
+ .field("a", DataTypes.BIGINT().notNull())
+ .field("b", DataTypes.STRING())
+ .field("c", DataTypes.TIMESTAMP().notNull())
+ .primaryKey("a", "c")
+ .build();
+
+ Function<RowData, String> keyExtractor = KeyExtractor.createKeyExtractor(schema, "_");
+
+ String key =
+ keyExtractor.apply(
+ GenericRowData.of(
+ 12L,
+ StringData.fromString("ABCD"),
+ TimestampData.fromLocalDateTime(
+ LocalDateTime.parse("2012-12-12T12:12:12"))));
+ assertThat(key, equalTo("12_2012-12-12T12:12:12"));
+ }
+
+ @Test
+ public void testAllTypesKey() {
+ TableSchema schema =
+ TableSchema.builder()
+ .field("a", DataTypes.TINYINT().notNull())
+ .field("b", DataTypes.SMALLINT().notNull())
+ .field("c", DataTypes.INT().notNull())
+ .field("d", DataTypes.BIGINT().notNull())
+ .field("e", DataTypes.BOOLEAN().notNull())
+ .field("f", DataTypes.FLOAT().notNull())
+ .field("g", DataTypes.DOUBLE().notNull())
+ .field("h", DataTypes.STRING().notNull())
+ .field("i", DataTypes.TIMESTAMP().notNull())
+ .field("j", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE().notNull())
+ .field("k", DataTypes.TIME().notNull())
+ .field("l", DataTypes.DATE().notNull())
+ .primaryKey("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l")
+ .build();
+
+ Function<RowData, String> keyExtractor = KeyExtractor.createKeyExtractor(schema, "_");
+
+ String key =
+ keyExtractor.apply(
+ GenericRowData.of(
+ (byte) 1,
+ (short) 2,
+ 3,
+ (long) 4,
+ true,
+ 1.0f,
+ 2.0d,
+ StringData.fromString("ABCD"),
+ TimestampData.fromLocalDateTime(
+ LocalDateTime.parse("2012-12-12T12:12:12")),
+ TimestampData.fromInstant(Instant.parse("2013-01-13T13:13:13Z")),
+ (int) (LocalTime.parse("14:14:14").toNanoOfDay() / 1_000_000),
+ (int) LocalDate.parse("2015-05-15").toEpochDay()));
+ assertThat(
+ key,
+ equalTo(
+ "1_2_3_4_true_1.0_2.0_ABCD_2012-12-12T12:12:12_2013-01-13T13:13:13_14:14:14_2015-05-15"));
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/TestContext.java b/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/TestContext.java
new file mode 100644
index 0000000..4040514
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/TestContext.java
@@ -0,0 +1,72 @@
+/*
+ * 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.table;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/** A utility class for mocking {@link DynamicTableFactory.Context}. */
+class TestContext {
+
+ private ResolvedSchema schema = ResolvedSchema.of(Column.physical("a", DataTypes.TIME()));
+
+ private final Map<String, String> options = new HashMap<>();
+
+ public static TestContext context() {
+ return new TestContext();
+ }
+
+ public TestContext withSchema(ResolvedSchema schema) {
+ this.schema = schema;
+ return this;
+ }
+
+ DynamicTableFactory.Context build() {
+ return new FactoryUtil.DefaultDynamicTableContext(
+ ObjectIdentifier.of("default", "default", "t1"),
+ new ResolvedCatalogTable(
+ CatalogTable.of(
+ Schema.newBuilder().fromResolvedSchema(schema).build(),
+ "mock context",
+ Collections.emptyList(),
+ options),
+ schema),
+ Collections.emptyMap(),
+ new Configuration(),
+ TestContext.class.getClassLoader(),
+ false);
+ }
+
+ public TestContext withOption(String key, String value) {
+ options.put(key, value);
+ return this;
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6Configuration.java b/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6Configuration.java
new file mode 100644
index 0000000..8b83321
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6Configuration.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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.ValidationException;
+
+import org.apache.http.HttpHost;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.HOSTS_OPTION;
+
+/** Elasticsearch 6 specific configuration. */
+@Internal
+final class Elasticsearch6Configuration extends ElasticsearchConfiguration {
+ Elasticsearch6Configuration(ReadableConfig config, ClassLoader classLoader) {
+ super(config, classLoader);
+ }
+
+ public List<HttpHost> getHosts() {
+ return config.get(HOSTS_OPTION).stream()
+ .map(Elasticsearch6Configuration::validateAndParseHostsString)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Parse Hosts String to list.
+ *
+ * <p>Hosts String format was given as following:
+ *
+ * <pre>
+ * connector.hosts = http://host_name:9092;http://host_name:9093
+ * </pre>
+ */
+ private static HttpHost validateAndParseHostsString(String host) {
+ try {
+ HttpHost httpHost = HttpHost.create(host);
+ if (httpHost.getPort() < 0) {
+ throw new ValidationException(
+ String.format(
+ "Could not parse host '%s' in option '%s'. It should follow the format 'http://host_name:port'. Missing port.",
+ host, HOSTS_OPTION.key()));
+ }
+
+ if (httpHost.getSchemeName() == null) {
+ throw new ValidationException(
+ String.format(
+ "Could not parse host '%s' in option '%s'. It should follow the format 'http://host_name:port'. Missing scheme.",
+ host, HOSTS_OPTION.key()));
+ }
+ return httpHost;
+ } catch (Exception e) {
+ throw new ValidationException(
+ String.format(
+ "Could not parse host '%s' in option '%s'. It should follow the format 'http://host_name:port'.",
+ host, HOSTS_OPTION.key()),
+ e);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSink.java b/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSink.java
new file mode 100644
index 0000000..0c7c756
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSink.java
@@ -0,0 +1,335 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
+import org.apache.flink.streaming.connectors.elasticsearch6.RestClientFactory;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.StringUtils;
+
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+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.time.ZoneId;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A {@link DynamicTableSink} that describes how to create a {@link ElasticsearchSink} from a
+ * logical description.
+ */
+@PublicEvolving
+final class Elasticsearch6DynamicSink implements DynamicTableSink {
+ @VisibleForTesting
+ static final Elasticsearch6RequestFactory REQUEST_FACTORY = new Elasticsearch6RequestFactory();
+
+ private final EncodingFormat<SerializationSchema<RowData>> format;
+ private final TableSchema schema;
+ private final Elasticsearch6Configuration config;
+ private final ZoneId localTimeZoneId;
+ private final boolean isDynamicIndexWithSystemTime;
+
+ public Elasticsearch6DynamicSink(
+ EncodingFormat<SerializationSchema<RowData>> format,
+ Elasticsearch6Configuration config,
+ TableSchema schema,
+ ZoneId localTimeZoneId) {
+ this(format, config, schema, localTimeZoneId, (ElasticsearchSink.Builder::new));
+ }
+
+ // --------------------------------------------------------------
+ // Hack to make configuration testing possible.
+ //
+ // The code in this block should never be used outside of tests.
+ // Having a way to inject a builder we can assert the builder in
+ // the test. We can not assert everything though, e.g. it is not
+ // possible to assert flushing on checkpoint, as it is configured
+ // on the sink itself.
+ // --------------------------------------------------------------
+
+ private final ElasticSearchBuilderProvider builderProvider;
+
+ @FunctionalInterface
+ interface ElasticSearchBuilderProvider {
+ ElasticsearchSink.Builder<RowData> createBuilder(
+ List<HttpHost> httpHosts, RowElasticsearchSinkFunction upsertSinkFunction);
+ }
+
+ Elasticsearch6DynamicSink(
+ EncodingFormat<SerializationSchema<RowData>> format,
+ Elasticsearch6Configuration config,
+ TableSchema schema,
+ ZoneId localTimeZoneId,
+ ElasticSearchBuilderProvider builderProvider) {
+ this.format = format;
+ this.schema = schema;
+ this.config = config;
+ this.localTimeZoneId = localTimeZoneId;
+ this.isDynamicIndexWithSystemTime = isDynamicIndexWithSystemTime();
+ this.builderProvider = builderProvider;
+ }
+
+ // --------------------------------------------------------------
+ // End of hack to make configuration testing possible
+ // --------------------------------------------------------------
+
+ public boolean isDynamicIndexWithSystemTime() {
+ IndexGeneratorFactory.IndexHelper indexHelper = new IndexGeneratorFactory.IndexHelper();
+ return indexHelper.checkIsDynamicIndexWithSystemTimeFormat(config.getIndex());
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ ChangelogMode.Builder builder = ChangelogMode.newBuilder();
+ for (RowKind kind : requestedMode.getContainedKinds()) {
+ if (kind != RowKind.UPDATE_BEFORE) {
+ builder.addContainedKind(kind);
+ }
+ }
+ if (isDynamicIndexWithSystemTime && !requestedMode.containsOnly(RowKind.INSERT)) {
+ throw new ValidationException(
+ "Dynamic indexing based on system time only works on append only stream.");
+ }
+ return builder.build();
+ }
+
+ @Override
+ public SinkFunctionProvider getSinkRuntimeProvider(Context context) {
+ return () -> {
+ SerializationSchema<RowData> format =
+ this.format.createRuntimeEncoder(context, schema.toRowDataType());
+
+ final RowElasticsearchSinkFunction upsertFunction =
+ new RowElasticsearchSinkFunction(
+ IndexGeneratorFactory.createIndexGenerator(
+ config.getIndex(), schema, localTimeZoneId),
+ config.getDocumentType(),
+ format,
+ XContentType.JSON,
+ REQUEST_FACTORY,
+ KeyExtractor.createKeyExtractor(schema, config.getKeyDelimiter()));
+
+ final ElasticsearchSink.Builder<RowData> builder =
+ builderProvider.createBuilder(config.getHosts(), upsertFunction);
+
+ builder.setFailureHandler(config.getFailureHandler());
+ builder.setBulkFlushMaxActions(config.getBulkFlushMaxActions());
+ builder.setBulkFlushMaxSizeMb((int) (config.getBulkFlushMaxByteSize() >> 20));
+ builder.setBulkFlushInterval(config.getBulkFlushInterval());
+ builder.setBulkFlushBackoff(config.isBulkFlushBackoffEnabled());
+ config.getBulkFlushBackoffType().ifPresent(builder::setBulkFlushBackoffType);
+ config.getBulkFlushBackoffRetries().ifPresent(builder::setBulkFlushBackoffRetries);
+ config.getBulkFlushBackoffDelay().ifPresent(builder::setBulkFlushBackoffDelay);
+
+ // we must overwrite the default factory which is defined with a lambda because of a bug
+ // in shading lambda serialization shading see FLINK-18006
+ if (config.getUsername().isPresent()
+ && config.getPassword().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getUsername().get())
+ && !StringUtils.isNullOrWhitespaceOnly(config.getPassword().get())) {
+ builder.setRestClientFactory(
+ new AuthRestClientFactory(
+ config.getPathPrefix().orElse(null),
+ config.getUsername().get(),
+ config.getPassword().get()));
+ } else {
+ builder.setRestClientFactory(
+ new DefaultRestClientFactory(config.getPathPrefix().orElse(null)));
+ }
+
+ final ElasticsearchSink<RowData> sink = builder.build();
+
+ if (config.isDisableFlushOnCheckpoint()) {
+ sink.disableFlushOnCheckpoint();
+ }
+
+ return sink;
+ };
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return this;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "Elasticsearch6";
+ }
+
+ /** Serializable {@link RestClientFactory} used by the sink. */
+ @VisibleForTesting
+ static class DefaultRestClientFactory implements RestClientFactory {
+
+ private final 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);
+ }
+ }
+
+ /** Serializable {@link RestClientFactory} used by the sink which enable authentication. */
+ @VisibleForTesting
+ static class AuthRestClientFactory implements RestClientFactory {
+
+ private final String pathPrefix;
+ private final String username;
+ private final String password;
+ private transient CredentialsProvider credentialsProvider;
+
+ public AuthRestClientFactory(
+ @Nullable String pathPrefix, String username, String password) {
+ this.pathPrefix = pathPrefix;
+ this.password = password;
+ this.username = username;
+ }
+
+ @Override
+ public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
+ if (pathPrefix != null) {
+ restClientBuilder.setPathPrefix(pathPrefix);
+ }
+ if (credentialsProvider == null) {
+ credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider.setCredentials(
+ AuthScope.ANY, new UsernamePasswordCredentials(username, password));
+ }
+ restClientBuilder.setHttpClientConfigCallback(
+ httpAsyncClientBuilder ->
+ httpAsyncClientBuilder.setDefaultCredentialsProvider(
+ credentialsProvider));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ AuthRestClientFactory that = (AuthRestClientFactory) o;
+ return Objects.equals(pathPrefix, that.pathPrefix)
+ && Objects.equals(username, that.username)
+ && Objects.equals(password, that.password);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(pathPrefix, username, password);
+ }
+ }
+
+ /**
+ * Version-specific creation of {@link org.elasticsearch.action.ActionRequest}s used by the
+ * sink.
+ */
+ private static class Elasticsearch6RequestFactory implements RequestFactory {
+ @Override
+ public UpdateRequest createUpdateRequest(
+ String index,
+ String docType,
+ String key,
+ XContentType contentType,
+ byte[] document) {
+ return new UpdateRequest(index, docType, key)
+ .doc(document, contentType)
+ .upsert(document, contentType);
+ }
+
+ @Override
+ public IndexRequest createIndexRequest(
+ String index,
+ String docType,
+ String key,
+ XContentType contentType,
+ byte[] document) {
+ return new IndexRequest(index, docType, key).source(document, contentType);
+ }
+
+ @Override
+ public DeleteRequest createDeleteRequest(String index, String docType, String key) {
+ return new DeleteRequest(index, docType, key);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Elasticsearch6DynamicSink that = (Elasticsearch6DynamicSink) o;
+ return Objects.equals(format, that.format)
+ && Objects.equals(schema, that.schema)
+ && Objects.equals(config, that.config)
+ && Objects.equals(builderProvider, that.builderProvider);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(format, schema, config, builderProvider);
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkFactory.java b/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkFactory.java
new file mode 100644
index 0000000..ff600a7
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkFactory.java
@@ -0,0 +1,186 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.StringUtils;
+
+import java.time.ZoneId;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLASH_MAX_SIZE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_TYPE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_INTERVAL_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.CONNECTION_MAX_RETRY_TIMEOUT_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.CONNECTION_PATH_PREFIX;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.FAILURE_HANDLER_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.FORMAT_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.HOSTS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.INDEX_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.KEY_DELIMITER_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.PASSWORD_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.USERNAME_OPTION;
+
+/** A {@link DynamicTableSinkFactory} for discovering {@link Elasticsearch6DynamicSink}. */
+@Internal
+public class Elasticsearch6DynamicSinkFactory implements DynamicTableSinkFactory {
+ private static final Set<ConfigOption<?>> requiredOptions =
+ Stream.of(HOSTS_OPTION, INDEX_OPTION, DOCUMENT_TYPE_OPTION).collect(Collectors.toSet());
+ private static final Set<ConfigOption<?>> optionalOptions =
+ Stream.of(
+ KEY_DELIMITER_OPTION,
+ FAILURE_HANDLER_OPTION,
+ FLUSH_ON_CHECKPOINT_OPTION,
+ BULK_FLASH_MAX_SIZE_OPTION,
+ BULK_FLUSH_MAX_ACTIONS_OPTION,
+ BULK_FLUSH_INTERVAL_OPTION,
+ BULK_FLUSH_BACKOFF_TYPE_OPTION,
+ BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION,
+ BULK_FLUSH_BACKOFF_DELAY_OPTION,
+ CONNECTION_MAX_RETRY_TIMEOUT_OPTION,
+ CONNECTION_PATH_PREFIX,
+ FORMAT_OPTION,
+ PASSWORD_OPTION,
+ USERNAME_OPTION)
+ .collect(Collectors.toSet());
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ TableSchema tableSchema = context.getCatalogTable().getSchema();
+ ElasticsearchValidationUtils.validatePrimaryKey(tableSchema);
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ final EncodingFormat<SerializationSchema<RowData>> format =
+ helper.discoverEncodingFormat(SerializationFormatFactory.class, FORMAT_OPTION);
+
+ helper.validate();
+ Configuration configuration = new Configuration();
+ context.getCatalogTable().getOptions().forEach(configuration::setString);
+ Elasticsearch6Configuration config =
+ new Elasticsearch6Configuration(configuration, context.getClassLoader());
+
+ validate(config, configuration);
+
+ return new Elasticsearch6DynamicSink(
+ format,
+ config,
+ TableSchemaUtils.getPhysicalSchema(tableSchema),
+ getLocalTimeZoneId(context.getConfiguration()));
+ }
+
+ ZoneId getLocalTimeZoneId(ReadableConfig readableConfig) {
+ final String zone = readableConfig.get(TableConfigOptions.LOCAL_TIME_ZONE);
+ final ZoneId zoneId =
+ TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)
+ ? ZoneId.systemDefault()
+ : ZoneId.of(zone);
+
+ return zoneId;
+ }
+
+ private void validate(Elasticsearch6Configuration config, Configuration originalConfiguration) {
+ config.getFailureHandler(); // checks if we can instantiate the custom failure handler
+ config.getHosts(); // validate hosts
+ validate(
+ config.getIndex().length() >= 1,
+ () -> String.format("'%s' must not be empty", INDEX_OPTION.key()));
+ int maxActions = config.getBulkFlushMaxActions();
+ validate(
+ maxActions == -1 || maxActions >= 1,
+ () ->
+ String.format(
+ "'%s' must be at least 1. Got: %s",
+ BULK_FLUSH_MAX_ACTIONS_OPTION.key(), maxActions));
+ long maxSize = config.getBulkFlushMaxByteSize();
+ long mb1 = 1024 * 1024;
+ validate(
+ maxSize == -1 || (maxSize >= mb1 && maxSize % mb1 == 0),
+ () ->
+ String.format(
+ "'%s' must be in MB granularity. Got: %s",
+ BULK_FLASH_MAX_SIZE_OPTION.key(),
+ originalConfiguration
+ .get(BULK_FLASH_MAX_SIZE_OPTION)
+ .toHumanReadableString()));
+ validate(
+ config.getBulkFlushBackoffRetries().map(retries -> retries >= 1).orElse(true),
+ () ->
+ String.format(
+ "'%s' must be at least 1. Got: %s",
+ BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION.key(),
+ config.getBulkFlushBackoffRetries().get()));
+ if (config.getUsername().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getUsername().get())) {
+ validate(
+ config.getPassword().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getPassword().get()),
+ () ->
+ String.format(
+ "'%s' and '%s' must be set at the same time. Got: username '%s' and password '%s'",
+ USERNAME_OPTION.key(),
+ PASSWORD_OPTION.key(),
+ config.getUsername().get(),
+ config.getPassword().orElse("")));
+ }
+ }
+
+ private static void validate(boolean condition, Supplier<String> message) {
+ if (!condition) {
+ throw new ValidationException(message.get());
+ }
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return "elasticsearch-6";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return requiredOptions;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return optionalOptions;
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch6/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-connectors/flink-connector-elasticsearch6/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
index bb5a894..29a8593 100644
--- a/flink-connectors/flink-connector-elasticsearch6/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++ b/flink-connectors/flink-connector-elasticsearch6/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -13,4 +13,4 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-org.apache.flink.connector.elasticsearch.table.Elasticsearch6DynamicSinkFactory
+org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch6DynamicSinkFactory
diff --git a/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkFactoryTest.java b/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkFactoryTest.java
new file mode 100644
index 0000000..e99abbe
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkFactoryTest.java
@@ -0,0 +1,250 @@
+/*
+ * 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.table;
+
+import org.apache.flink.api.common.typeutils.base.VoidSerializer;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.TestContext.context;
+
+/** Tests for validation in {@link Elasticsearch6DynamicSinkFactory}. */
+public class Elasticsearch6DynamicSinkFactoryTest extends TestLogger {
+ @Rule public ExpectedException thrown = ExpectedException.none();
+
+ @Test
+ public void validateEmptyConfiguration() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "One or more required options are missing.\n"
+ + "\n"
+ + "Missing required options are:\n"
+ + "\n"
+ + "document-type\n"
+ + "hosts\n"
+ + "index");
+ sinkFactory.createDynamicTableSink(context().build());
+ }
+
+ @Test
+ public void validateWrongIndex() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage("'index' must not be empty");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption("index", "")
+ .withOption("document-type", "MyType")
+ .withOption("hosts", "http://localhost:12345")
+ .build());
+ }
+
+ @Test
+ public void validateWrongHosts() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "Could not parse host 'wrong-host' in option 'hosts'. It should follow the format 'http://host_name:port'.");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption("index", "MyIndex")
+ .withOption("document-type", "MyType")
+ .withOption("hosts", "wrong-host")
+ .build());
+ }
+
+ @Test
+ public void validateWrongFlushSize() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "'sink.bulk-flush.max-size' must be in MB granularity. Got: 1024 bytes");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), "MyType")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLASH_MAX_SIZE_OPTION.key(),
+ "1kb")
+ .build());
+ }
+
+ @Test
+ public void validateWrongRetries() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage("'sink.bulk-flush.backoff.max-retries' must be at least 1. Got: 0");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), "MyType")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION
+ .key(),
+ "0")
+ .build());
+ }
+
+ @Test
+ public void validateWrongMaxActions() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage("'sink.bulk-flush.max-actions' must be at least 1. Got: -2");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), "MyType")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION.key(),
+ "-2")
+ .build());
+ }
+
+ @Test
+ public void validateWrongBackoffDelay() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage("Invalid value for option 'sink.bulk-flush.backoff.delay'.");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), "MyType")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION.key(),
+ "-1s")
+ .build());
+ }
+
+ @Test
+ public void validatePrimaryKeyOnIllegalColumn() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "The table has a primary key on columns of illegal types: "
+ + "[ARRAY, MAP, MULTISET, ROW, RAW, VARBINARY].\n"
+ + " Elasticsearch sink does not support primary keys on columns of types: "
+ + "[ARRAY, MAP, MULTISET, STRUCTURED_TYPE, ROW, RAW, BINARY, VARBINARY].");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withSchema(
+ new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("a", DataTypes.BIGINT().notNull()),
+ Column.physical(
+ "b",
+ DataTypes.ARRAY(
+ DataTypes.BIGINT()
+ .notNull())
+ .notNull()),
+ Column.physical(
+ "c",
+ DataTypes.MAP(
+ DataTypes.BIGINT(),
+ DataTypes.STRING())
+ .notNull()),
+ Column.physical(
+ "d",
+ DataTypes.MULTISET(
+ DataTypes.BIGINT()
+ .notNull())
+ .notNull()),
+ Column.physical(
+ "e",
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "a",
+ DataTypes.BIGINT()))
+ .notNull()),
+ Column.physical(
+ "f",
+ DataTypes.RAW(
+ Void.class,
+ VoidSerializer.INSTANCE)
+ .notNull()),
+ Column.physical("g", DataTypes.BYTES().notNull())),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey(
+ "name",
+ Arrays.asList("a", "b", "c", "d", "e", "f", "g"))))
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION.key(),
+ "1s")
+ .build());
+ }
+
+ @Test
+ public void validateWrongCredential() {
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "'username' and 'password' must be set at the same time. Got: username 'username' and password ''");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), "MyType")
+ .withOption(ElasticsearchConnectorOptions.USERNAME_OPTION.key(), "username")
+ .withOption(ElasticsearchConnectorOptions.PASSWORD_OPTION.key(), "")
+ .build());
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java b/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java
new file mode 100644
index 0000000..0ebc52b
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java
@@ -0,0 +1,399 @@
+/*
+ * 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.table;
+
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.TestLogger;
+
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.search.SearchHits;
+import org.elasticsearch.transport.client.PreBuiltTransportClient;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.TestContext.context;
+import static org.apache.flink.table.api.Expressions.row;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/** IT tests for {@link Elasticsearch6DynamicSink}. */
+public class Elasticsearch6DynamicSinkITCase extends TestLogger {
+
+ @ClassRule
+ public static ElasticsearchContainer elasticsearchContainer =
+ new ElasticsearchContainer(DockerImageName.parse(DockerImageVersions.ELASTICSEARCH_6));
+
+ @SuppressWarnings("deprecation")
+ protected final Client getClient() {
+ TransportAddress transportAddress =
+ new TransportAddress(elasticsearchContainer.getTcpHost());
+ String expectedClusterName = "docker-cluster";
+ Settings settings = Settings.builder().put("cluster.name", expectedClusterName).build();
+ return new PreBuiltTransportClient(settings).addTransportAddress(transportAddress);
+ }
+
+ @Test
+ public void testWritingDocuments() throws Exception {
+ ResolvedSchema schema =
+ new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("a", DataTypes.BIGINT().notNull()),
+ Column.physical("b", DataTypes.TIME()),
+ Column.physical("c", DataTypes.STRING().notNull()),
+ Column.physical("d", DataTypes.FLOAT()),
+ Column.physical("e", DataTypes.TINYINT().notNull()),
+ Column.physical("f", DataTypes.DATE()),
+ Column.physical("g", DataTypes.TIMESTAMP().notNull())),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey("name", Arrays.asList("a", "g")));
+ GenericRowData rowData =
+ GenericRowData.of(
+ 1L,
+ 12345,
+ StringData.fromString("ABCDE"),
+ 12.12f,
+ (byte) 2,
+ 12345,
+ TimestampData.fromLocalDateTime(
+ LocalDateTime.parse("2012-12-12T12:12:12")));
+
+ String index = "writing-documents";
+ String myType = "MyType";
+ Elasticsearch6DynamicSinkFactory sinkFactory = new Elasticsearch6DynamicSinkFactory();
+
+ SinkFunctionProvider sinkRuntimeProvider =
+ (SinkFunctionProvider)
+ sinkFactory
+ .createDynamicTableSink(
+ context()
+ .withSchema(schema)
+ .withOption(
+ ElasticsearchConnectorOptions.INDEX_OPTION
+ .key(),
+ index)
+ .withOption(
+ ElasticsearchConnectorOptions
+ .DOCUMENT_TYPE_OPTION
+ .key(),
+ myType)
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION
+ .key(),
+ elasticsearchContainer.getHttpHostAddress())
+ .withOption(
+ ElasticsearchConnectorOptions
+ .FLUSH_ON_CHECKPOINT_OPTION
+ .key(),
+ "false")
+ .build())
+ .getSinkRuntimeProvider(new MockContext());
+
+ SinkFunction<RowData> sinkFunction = sinkRuntimeProvider.createSinkFunction();
+ StreamExecutionEnvironment environment =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+ environment.setParallelism(4);
+
+ rowData.setRowKind(RowKind.UPDATE_AFTER);
+ environment.<RowData>fromElements(rowData).addSink(sinkFunction);
+ environment.execute();
+
+ Client client = getClient();
+ Map<String, Object> response =
+ client.get(new GetRequest(index, myType, "1_2012-12-12T12:12:12"))
+ .actionGet()
+ .getSource();
+ Map<Object, Object> expectedMap = new HashMap<>();
+ expectedMap.put("a", 1);
+ expectedMap.put("b", "00:00:12");
+ expectedMap.put("c", "ABCDE");
+ expectedMap.put("d", 12.12d);
+ expectedMap.put("e", 2);
+ expectedMap.put("f", "2003-10-20");
+ expectedMap.put("g", "2012-12-12 12:12:12");
+ assertThat(response, equalTo(expectedMap));
+ }
+
+ @Test
+ public void testWritingDocumentsFromTableApi() throws Exception {
+ TableEnvironment tableEnvironment =
+ TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+ String index = "table-api";
+ String myType = "MyType";
+ tableEnvironment.executeSql(
+ "CREATE TABLE esTable ("
+ + "a BIGINT NOT NULL,\n"
+ + "b TIME,\n"
+ + "c STRING NOT NULL,\n"
+ + "d FLOAT,\n"
+ + "e TINYINT NOT NULL,\n"
+ + "f DATE,\n"
+ + "g TIMESTAMP NOT NULL,\n"
+ + "h as a + 2,\n"
+ + "PRIMARY KEY (a, g) NOT ENFORCED\n"
+ + ")\n"
+ + "WITH (\n"
+ + String.format("'%s'='%s',\n", "connector", "elasticsearch-6")
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.INDEX_OPTION.key(), index)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), myType)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ + String.format(
+ "'%s'='%s'\n",
+ ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
+ "false")
+ + ")");
+
+ tableEnvironment
+ .fromValues(
+ row(
+ 1L,
+ LocalTime.ofNanoOfDay(12345L * 1_000_000L),
+ "ABCDE",
+ 12.12f,
+ (byte) 2,
+ LocalDate.ofEpochDay(12345),
+ LocalDateTime.parse("2012-12-12T12:12:12")))
+ .executeInsert("esTable")
+ .await();
+
+ Client client = getClient();
+ Map<String, Object> response =
+ client.get(new GetRequest(index, myType, "1_2012-12-12T12:12:12"))
+ .actionGet()
+ .getSource();
+ Map<Object, Object> expectedMap = new HashMap<>();
+ expectedMap.put("a", 1);
+ expectedMap.put("b", "00:00:12");
+ expectedMap.put("c", "ABCDE");
+ expectedMap.put("d", 12.12d);
+ expectedMap.put("e", 2);
+ expectedMap.put("f", "2003-10-20");
+ expectedMap.put("g", "2012-12-12 12:12:12");
+ assertThat(response, equalTo(expectedMap));
+ }
+
+ @Test
+ public void testWritingDocumentsNoPrimaryKey() throws Exception {
+ TableEnvironment tableEnvironment =
+ TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+ String index = "no-primary-key";
+ String myType = "MyType";
+ tableEnvironment.executeSql(
+ "CREATE TABLE esTable ("
+ + "a BIGINT NOT NULL,\n"
+ + "b TIME,\n"
+ + "c STRING NOT NULL,\n"
+ + "d FLOAT,\n"
+ + "e TINYINT NOT NULL,\n"
+ + "f DATE,\n"
+ + "g TIMESTAMP NOT NULL\n"
+ + ")\n"
+ + "WITH (\n"
+ + String.format("'%s'='%s',\n", "connector", "elasticsearch-6")
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.INDEX_OPTION.key(), index)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), myType)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ + String.format(
+ "'%s'='%s'\n",
+ ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
+ "false")
+ + ")");
+
+ tableEnvironment
+ .fromValues(
+ row(
+ 1L,
+ LocalTime.ofNanoOfDay(12345L * 1_000_000L),
+ "ABCDE",
+ 12.12f,
+ (byte) 2,
+ LocalDate.ofEpochDay(12345),
+ LocalDateTime.parse("2012-12-12T12:12:12")),
+ row(
+ 2L,
+ LocalTime.ofNanoOfDay(12345L * 1_000_000L),
+ "FGHIJK",
+ 13.13f,
+ (byte) 4,
+ LocalDate.ofEpochDay(12345),
+ LocalDateTime.parse("2013-12-12T13:13:13")))
+ .executeInsert("esTable")
+ .await();
+
+ Client client = getClient();
+
+ // search API does not return documents that were not indexed, we might need to query
+ // the index a few times
+ Deadline deadline = Deadline.fromNow(Duration.ofSeconds(30));
+ SearchHits hits;
+ do {
+ hits = client.prepareSearch(index).execute().actionGet().getHits();
+ if (hits.getTotalHits() < 2) {
+ Thread.sleep(200);
+ }
+ } while (hits.getTotalHits() < 2 && deadline.hasTimeLeft());
+
+ if (hits.getTotalHits() < 2) {
+ throw new AssertionError("Could not retrieve results from Elasticsearch.");
+ }
+
+ HashSet<Map<String, Object>> resultSet = new HashSet<>();
+ resultSet.add(hits.getAt(0).getSourceAsMap());
+ resultSet.add(hits.getAt(1).getSourceAsMap());
+ Map<Object, Object> expectedMap1 = new HashMap<>();
+ expectedMap1.put("a", 1);
+ expectedMap1.put("b", "00:00:12");
+ expectedMap1.put("c", "ABCDE");
+ expectedMap1.put("d", 12.12d);
+ expectedMap1.put("e", 2);
+ expectedMap1.put("f", "2003-10-20");
+ expectedMap1.put("g", "2012-12-12 12:12:12");
+ Map<Object, Object> expectedMap2 = new HashMap<>();
+ expectedMap2.put("a", 2);
+ expectedMap2.put("b", "00:00:12");
+ expectedMap2.put("c", "FGHIJK");
+ expectedMap2.put("d", 13.13d);
+ expectedMap2.put("e", 4);
+ expectedMap2.put("f", "2003-10-20");
+ expectedMap2.put("g", "2013-12-12 13:13:13");
+ HashSet<Map<Object, Object>> expectedSet = new HashSet<>();
+ expectedSet.add(expectedMap1);
+ expectedSet.add(expectedMap2);
+ assertThat(resultSet, equalTo(expectedSet));
+ }
+
+ @Test
+ public void testWritingDocumentsWithDynamicIndex() throws Exception {
+ TableEnvironment tableEnvironment =
+ TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+ String index = "dynamic-index-{b|yyyy-MM-dd}";
+ String myType = "MyType";
+ tableEnvironment.executeSql(
+ "CREATE TABLE esTable ("
+ + "a BIGINT NOT NULL,\n"
+ + "b TIMESTAMP NOT NULL,\n"
+ + "PRIMARY KEY (a) NOT ENFORCED\n"
+ + ")\n"
+ + "WITH (\n"
+ + String.format("'%s'='%s',\n", "connector", "elasticsearch-6")
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.INDEX_OPTION.key(), index)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), myType)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ + String.format(
+ "'%s'='%s'\n",
+ ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
+ "false")
+ + ")");
+
+ tableEnvironment
+ .fromValues(row(1L, LocalDateTime.parse("2012-12-12T12:12:12")))
+ .executeInsert("esTable")
+ .await();
+
+ Client client = getClient();
+ Map<String, Object> response =
+ client.get(new GetRequest("dynamic-index-2012-12-12", myType, "1"))
+ .actionGet()
+ .getSource();
+ Map<Object, Object> expectedMap = new HashMap<>();
+ expectedMap.put("a", 1);
+ expectedMap.put("b", "2012-12-12 12:12:12");
+ assertThat(response, equalTo(expectedMap));
+ }
+
+ private static class MockContext implements DynamicTableSink.Context {
+ @Override
+ public boolean isBounded() {
+ return false;
+ }
+
+ @Override
+ public TypeInformation<?> createTypeInformation(DataType consumedDataType) {
+ return null;
+ }
+
+ @Override
+ public TypeInformation<?> createTypeInformation(LogicalType consumedLogicalType) {
+ return null;
+ }
+
+ @Override
+ public DynamicTableSink.DataStructureConverter createDataStructureConverter(
+ DataType consumedDataType) {
+ return null;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkTest.java b/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkTest.java
new file mode 100644
index 0000000..f8ab0ab
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkTest.java
@@ -0,0 +1,298 @@
+/*
+ * 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.table;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
+import org.apache.flink.streaming.connectors.elasticsearch.util.NoOpFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.ActionRequest;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.time.ZoneId;
+import java.util.List;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link Elasticsearch6DynamicSink} parameters. */
+public class Elasticsearch6DynamicSinkTest extends TestLogger {
+
+ private static final String FIELD_KEY = "key";
+ private static final String FIELD_FRUIT_NAME = "fruit_name";
+ private static final String FIELD_COUNT = "count";
+ private static final String FIELD_TS = "ts";
+
+ private static final String HOSTNAME = "host1";
+ private static final int PORT = 1234;
+ private static final String SCHEMA = "https";
+ private static final String INDEX = "MyIndex";
+ private static final String DOC_TYPE = "MyType";
+ private static final String USERNAME = "username";
+ private static final String PASSWORD = "password";
+
+ @Test
+ public void testBuilder() {
+ final TableSchema schema = createTestSchema();
+
+ BuilderProvider provider = new BuilderProvider();
+ final Elasticsearch6DynamicSink testSink =
+ new Elasticsearch6DynamicSink(
+ new DummyEncodingFormat(),
+ new Elasticsearch6Configuration(
+ getConfig(), this.getClass().getClassLoader()),
+ schema,
+ ZoneId.systemDefault(),
+ provider);
+
+ testSink.getSinkRuntimeProvider(new MockSinkContext()).createSinkFunction();
+
+ verify(provider.builderSpy).setFailureHandler(new DummyFailureHandler());
+ verify(provider.builderSpy).setBulkFlushBackoff(true);
+ verify(provider.builderSpy)
+ .setBulkFlushBackoffType(ElasticsearchSinkBase.FlushBackoffType.EXPONENTIAL);
+ verify(provider.builderSpy).setBulkFlushBackoffDelay(123);
+ verify(provider.builderSpy).setBulkFlushBackoffRetries(3);
+ verify(provider.builderSpy).setBulkFlushInterval(100);
+ verify(provider.builderSpy).setBulkFlushMaxActions(1000);
+ verify(provider.builderSpy).setBulkFlushMaxSizeMb(1);
+ verify(provider.builderSpy)
+ .setRestClientFactory(
+ new Elasticsearch6DynamicSink.DefaultRestClientFactory("/myapp"));
+ verify(provider.sinkSpy).disableFlushOnCheckpoint();
+ }
+
+ @Test
+ public void testDefaultConfig() {
+ final TableSchema schema = createTestSchema();
+ Configuration configuration = new Configuration();
+ configuration.setString(ElasticsearchConnectorOptions.INDEX_OPTION.key(), INDEX);
+ configuration.setString(ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), DOC_TYPE);
+ configuration.setString(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ SCHEMA + "://" + HOSTNAME + ":" + PORT);
+
+ BuilderProvider provider = new BuilderProvider();
+ final Elasticsearch6DynamicSink testSink =
+ new Elasticsearch6DynamicSink(
+ new DummyEncodingFormat(),
+ new Elasticsearch6Configuration(
+ configuration, this.getClass().getClassLoader()),
+ schema,
+ ZoneId.systemDefault(),
+ provider);
+
+ testSink.getSinkRuntimeProvider(new MockSinkContext()).createSinkFunction();
+
+ verify(provider.builderSpy).setFailureHandler(new NoOpFailureHandler());
+ verify(provider.builderSpy).setBulkFlushBackoff(false);
+ verify(provider.builderSpy).setBulkFlushInterval(1000);
+ verify(provider.builderSpy).setBulkFlushMaxActions(1000);
+ verify(provider.builderSpy).setBulkFlushMaxSizeMb(2);
+ verify(provider.builderSpy)
+ .setRestClientFactory(new Elasticsearch6DynamicSink.DefaultRestClientFactory(null));
+ verify(provider.sinkSpy, never()).disableFlushOnCheckpoint();
+ }
+
+ @Test
+ public void testAuthConfig() {
+ final TableSchema schema = createTestSchema();
+ Configuration configuration = new Configuration();
+ configuration.setString(ElasticsearchConnectorOptions.INDEX_OPTION.key(), INDEX);
+ configuration.setString(ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), DOC_TYPE);
+ configuration.setString(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ SCHEMA + "://" + HOSTNAME + ":" + PORT);
+ configuration.setString(ElasticsearchConnectorOptions.USERNAME_OPTION.key(), USERNAME);
+ configuration.setString(ElasticsearchConnectorOptions.PASSWORD_OPTION.key(), PASSWORD);
+
+ BuilderProvider provider = new BuilderProvider();
+ final Elasticsearch6DynamicSink testSink =
+ new Elasticsearch6DynamicSink(
+ new DummyEncodingFormat(),
+ new Elasticsearch6Configuration(
+ configuration, this.getClass().getClassLoader()),
+ schema,
+ ZoneId.systemDefault(),
+ provider);
+
+ testSink.getSinkRuntimeProvider(new MockSinkContext()).createSinkFunction();
+
+ verify(provider.builderSpy).setFailureHandler(new NoOpFailureHandler());
+ verify(provider.builderSpy).setBulkFlushBackoff(false);
+ verify(provider.builderSpy).setBulkFlushInterval(1000);
+ verify(provider.builderSpy).setBulkFlushMaxActions(1000);
+ verify(provider.builderSpy).setBulkFlushMaxSizeMb(2);
+ verify(provider.builderSpy)
+ .setRestClientFactory(
+ new Elasticsearch6DynamicSink.AuthRestClientFactory(
+ null, USERNAME, PASSWORD));
+ verify(provider.sinkSpy, never()).disableFlushOnCheckpoint();
+ }
+
+ private Configuration getConfig() {
+ Configuration configuration = new Configuration();
+ configuration.setString(ElasticsearchConnectorOptions.INDEX_OPTION.key(), INDEX);
+ configuration.setString(ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), DOC_TYPE);
+ configuration.setString(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ SCHEMA + "://" + HOSTNAME + ":" + PORT);
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_TYPE_OPTION.key(), "exponential");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION.key(), "123");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION.key(), "3");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_INTERVAL_OPTION.key(), "100");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION.key(), "1000");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLASH_MAX_SIZE_OPTION.key(), "1mb");
+ configuration.setString(
+ ElasticsearchConnectorOptions.CONNECTION_PATH_PREFIX.key(), "/myapp");
+ configuration.setString(
+ ElasticsearchConnectorOptions.FAILURE_HANDLER_OPTION.key(),
+ DummyFailureHandler.class.getName());
+ configuration.setString(
+ ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(), "false");
+ return configuration;
+ }
+
+ private static class BuilderProvider
+ implements Elasticsearch6DynamicSink.ElasticSearchBuilderProvider {
+ public ElasticsearchSink.Builder<RowData> builderSpy;
+ public ElasticsearchSink<RowData> sinkSpy;
+
+ @Override
+ public ElasticsearchSink.Builder<RowData> createBuilder(
+ List<HttpHost> httpHosts, RowElasticsearchSinkFunction upsertSinkFunction) {
+ builderSpy =
+ Mockito.spy(new ElasticsearchSink.Builder<>(httpHosts, upsertSinkFunction));
+ doAnswer(
+ invocation -> {
+ sinkSpy =
+ Mockito.spy(
+ (ElasticsearchSink<RowData>)
+ invocation.callRealMethod());
+ return sinkSpy;
+ })
+ .when(builderSpy)
+ .build();
+
+ return builderSpy;
+ }
+ }
+
+ private TableSchema createTestSchema() {
+ return TableSchema.builder()
+ .field(FIELD_KEY, DataTypes.BIGINT())
+ .field(FIELD_FRUIT_NAME, DataTypes.STRING())
+ .field(FIELD_COUNT, DataTypes.DECIMAL(10, 4))
+ .field(FIELD_TS, DataTypes.TIMESTAMP(3))
+ .build();
+ }
+
+ private static class DummySerializationSchema implements SerializationSchema<RowData> {
+
+ private static final DummySerializationSchema INSTANCE = new DummySerializationSchema();
+
+ @Override
+ public byte[] serialize(RowData element) {
+ return new byte[0];
+ }
+ }
+
+ private static class DummyEncodingFormat
+ implements EncodingFormat<SerializationSchema<RowData>> {
+ @Override
+ public SerializationSchema<RowData> createRuntimeEncoder(
+ DynamicTableSink.Context context, DataType consumedDataType) {
+ return DummySerializationSchema.INSTANCE;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return null;
+ }
+ }
+
+ private static class MockSinkContext implements DynamicTableSink.Context {
+ @Override
+ public boolean isBounded() {
+ return false;
+ }
+
+ @Override
+ public TypeInformation<?> createTypeInformation(DataType consumedDataType) {
+ return null;
+ }
+
+ @Override
+ public TypeInformation<?> createTypeInformation(LogicalType consumedLogicalType) {
+ return null;
+ }
+
+ @Override
+ public DynamicTableSink.DataStructureConverter createDataStructureConverter(
+ DataType consumedDataType) {
+ return null;
+ }
+ }
+
+ /** Custom failure handler for testing. */
+ public static class DummyFailureHandler implements ActionRequestFailureHandler {
+
+ @Override
+ public void onFailure(
+ ActionRequest action,
+ Throwable failure,
+ int restStatusCode,
+ RequestIndexer indexer) {
+ // do nothing
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof DummyFailureHandler;
+ }
+
+ @Override
+ public int hashCode() {
+ return DummyFailureHandler.class.hashCode();
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7Configuration.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7Configuration.java
new file mode 100644
index 0000000..6bd28cf
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7Configuration.java
@@ -0,0 +1,70 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.ValidationException;
+
+import org.apache.http.HttpHost;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.HOSTS_OPTION;
+
+/** Elasticsearch 7 specific configuration. */
+@Internal
+final class Elasticsearch7Configuration extends ElasticsearchConfiguration {
+ Elasticsearch7Configuration(ReadableConfig config, ClassLoader classLoader) {
+ super(config, classLoader);
+ }
+
+ public List<HttpHost> getHosts() {
+ return config.get(HOSTS_OPTION).stream()
+ .map(Elasticsearch7Configuration::validateAndParseHostsString)
+ .collect(Collectors.toList());
+ }
+
+ private static HttpHost validateAndParseHostsString(String host) {
+ try {
+ HttpHost httpHost = HttpHost.create(host);
+ if (httpHost.getPort() < 0) {
+ throw new ValidationException(
+ String.format(
+ "Could not parse host '%s' in option '%s'. It should follow the format 'http://host_name:port'. Missing port.",
+ host, HOSTS_OPTION.key()));
+ }
+
+ if (httpHost.getSchemeName() == null) {
+ throw new ValidationException(
+ String.format(
+ "Could not parse host '%s' in option '%s'. It should follow the format 'http://host_name:port'. Missing scheme.",
+ host, HOSTS_OPTION.key()));
+ }
+ return httpHost;
+ } catch (Exception e) {
+ throw new ValidationException(
+ String.format(
+ "Could not parse host '%s' in option '%s'. It should follow the format 'http://host_name:port'.",
+ host, HOSTS_OPTION.key()),
+ e);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSink.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSink.java
new file mode 100644
index 0000000..1926e44
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSink.java
@@ -0,0 +1,335 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+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.table.api.ValidationException;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.StringUtils;
+
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+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.time.ZoneId;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A {@link DynamicTableSink} that describes how to create a {@link ElasticsearchSink} from a
+ * logical description.
+ */
+@Internal
+final class Elasticsearch7DynamicSink implements DynamicTableSink {
+ @VisibleForTesting
+ static final Elasticsearch7RequestFactory REQUEST_FACTORY = new Elasticsearch7RequestFactory();
+
+ private final EncodingFormat<SerializationSchema<RowData>> format;
+ private final TableSchema schema;
+ private final Elasticsearch7Configuration config;
+ private final ZoneId localTimeZoneId;
+ private final boolean isDynamicIndexWithSystemTime;
+
+ public Elasticsearch7DynamicSink(
+ EncodingFormat<SerializationSchema<RowData>> format,
+ Elasticsearch7Configuration config,
+ TableSchema schema,
+ ZoneId localTimeZoneId) {
+ this(format, config, schema, localTimeZoneId, (ElasticsearchSink.Builder::new));
+ }
+
+ // --------------------------------------------------------------
+ // Hack to make configuration testing possible.
+ //
+ // The code in this block should never be used outside of tests.
+ // Having a way to inject a builder we can assert the builder in
+ // the test. We can not assert everything though, e.g. it is not
+ // possible to assert flushing on checkpoint, as it is configured
+ // on the sink itself.
+ // --------------------------------------------------------------
+
+ private final ElasticSearchBuilderProvider builderProvider;
+
+ @FunctionalInterface
+ interface ElasticSearchBuilderProvider {
+ ElasticsearchSink.Builder<RowData> createBuilder(
+ List<HttpHost> httpHosts, RowElasticsearchSinkFunction upsertSinkFunction);
+ }
+
+ Elasticsearch7DynamicSink(
+ EncodingFormat<SerializationSchema<RowData>> format,
+ Elasticsearch7Configuration config,
+ TableSchema schema,
+ ZoneId localTimeZoneId,
+ ElasticSearchBuilderProvider builderProvider) {
+ this.format = format;
+ this.schema = schema;
+ this.config = config;
+ this.localTimeZoneId = localTimeZoneId;
+ this.isDynamicIndexWithSystemTime = isDynamicIndexWithSystemTime();
+ this.builderProvider = builderProvider;
+ }
+
+ // --------------------------------------------------------------
+ // End of hack to make configuration testing possible
+ // --------------------------------------------------------------
+
+ public boolean isDynamicIndexWithSystemTime() {
+ IndexGeneratorFactory.IndexHelper indexHelper = new IndexGeneratorFactory.IndexHelper();
+ return indexHelper.checkIsDynamicIndexWithSystemTimeFormat(config.getIndex());
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ ChangelogMode.Builder builder = ChangelogMode.newBuilder();
+ for (RowKind kind : requestedMode.getContainedKinds()) {
+ if (kind != RowKind.UPDATE_BEFORE) {
+ builder.addContainedKind(kind);
+ }
+ }
+ if (isDynamicIndexWithSystemTime && !requestedMode.containsOnly(RowKind.INSERT)) {
+ throw new ValidationException(
+ "Dynamic indexing based on system time only works on append only stream.");
+ }
+ return builder.build();
+ }
+
+ @Override
+ public SinkFunctionProvider getSinkRuntimeProvider(Context context) {
+ return () -> {
+ SerializationSchema<RowData> format =
+ this.format.createRuntimeEncoder(context, schema.toRowDataType());
+
+ final RowElasticsearchSinkFunction upsertFunction =
+ new RowElasticsearchSinkFunction(
+ IndexGeneratorFactory.createIndexGenerator(
+ config.getIndex(), schema, localTimeZoneId),
+ null, // this is deprecated in es 7+
+ format,
+ XContentType.JSON,
+ REQUEST_FACTORY,
+ KeyExtractor.createKeyExtractor(schema, config.getKeyDelimiter()));
+
+ final ElasticsearchSink.Builder<RowData> builder =
+ builderProvider.createBuilder(config.getHosts(), upsertFunction);
+
+ builder.setFailureHandler(config.getFailureHandler());
+ builder.setBulkFlushMaxActions(config.getBulkFlushMaxActions());
+ builder.setBulkFlushMaxSizeMb((int) (config.getBulkFlushMaxByteSize() >> 20));
+ builder.setBulkFlushInterval(config.getBulkFlushInterval());
+ builder.setBulkFlushBackoff(config.isBulkFlushBackoffEnabled());
+ config.getBulkFlushBackoffType().ifPresent(builder::setBulkFlushBackoffType);
+ config.getBulkFlushBackoffRetries().ifPresent(builder::setBulkFlushBackoffRetries);
+ config.getBulkFlushBackoffDelay().ifPresent(builder::setBulkFlushBackoffDelay);
+
+ // we must overwrite the default factory which is defined with a lambda because of a bug
+ // in shading lambda serialization shading see FLINK-18006
+ if (config.getUsername().isPresent()
+ && config.getPassword().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getUsername().get())
+ && !StringUtils.isNullOrWhitespaceOnly(config.getPassword().get())) {
+ builder.setRestClientFactory(
+ new AuthRestClientFactory(
+ config.getPathPrefix().orElse(null),
+ config.getUsername().get(),
+ config.getPassword().get()));
+ } else {
+ builder.setRestClientFactory(
+ new DefaultRestClientFactory(config.getPathPrefix().orElse(null)));
+ }
+
+ final ElasticsearchSink<RowData> sink = builder.build();
+
+ if (config.isDisableFlushOnCheckpoint()) {
+ sink.disableFlushOnCheckpoint();
+ }
+
+ return sink;
+ };
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return this;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "Elasticsearch7";
+ }
+
+ /** Serializable {@link RestClientFactory} used by the sink. */
+ @VisibleForTesting
+ static class DefaultRestClientFactory implements RestClientFactory {
+
+ private final 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);
+ }
+ }
+
+ /** Serializable {@link RestClientFactory} used by the sink which enable authentication. */
+ @VisibleForTesting
+ static class AuthRestClientFactory implements RestClientFactory {
+
+ private final String pathPrefix;
+ private final String username;
+ private final String password;
+ private transient CredentialsProvider credentialsProvider;
+
+ public AuthRestClientFactory(
+ @Nullable String pathPrefix, String username, String password) {
+ this.pathPrefix = pathPrefix;
+ this.password = password;
+ this.username = username;
+ }
+
+ @Override
+ public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
+ if (pathPrefix != null) {
+ restClientBuilder.setPathPrefix(pathPrefix);
+ }
+ if (credentialsProvider == null) {
+ credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider.setCredentials(
+ AuthScope.ANY, new UsernamePasswordCredentials(username, password));
+ }
+ restClientBuilder.setHttpClientConfigCallback(
+ httpAsyncClientBuilder ->
+ httpAsyncClientBuilder.setDefaultCredentialsProvider(
+ credentialsProvider));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ AuthRestClientFactory that = (AuthRestClientFactory) o;
+ return Objects.equals(pathPrefix, that.pathPrefix)
+ && Objects.equals(username, that.username)
+ && Objects.equals(password, that.password);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(pathPrefix, password, username);
+ }
+ }
+
+ /**
+ * 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,
+ String key,
+ XContentType contentType,
+ byte[] document) {
+ return new IndexRequest(index).id(key).source(document, contentType);
+ }
+
+ @Override
+ public DeleteRequest createDeleteRequest(String index, String docType, String key) {
+ return new DeleteRequest(index, key);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Elasticsearch7DynamicSink that = (Elasticsearch7DynamicSink) o;
+ return Objects.equals(format, that.format)
+ && Objects.equals(schema, that.schema)
+ && Objects.equals(config, that.config)
+ && Objects.equals(builderProvider, that.builderProvider);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(format, schema, config, builderProvider);
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkFactory.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkFactory.java
new file mode 100644
index 0000000..50c37f0
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkFactory.java
@@ -0,0 +1,186 @@
+/*
+ * 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.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.StringUtils;
+
+import java.time.ZoneId;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLASH_MAX_SIZE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_TYPE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_INTERVAL_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.CONNECTION_MAX_RETRY_TIMEOUT_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.CONNECTION_PATH_PREFIX;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.FAILURE_HANDLER_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.FORMAT_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.HOSTS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.INDEX_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.KEY_DELIMITER_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.PASSWORD_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.USERNAME_OPTION;
+
+/** A {@link DynamicTableSinkFactory} for discovering {@link Elasticsearch7DynamicSink}. */
+@Internal
+public class Elasticsearch7DynamicSinkFactory implements DynamicTableSinkFactory {
+ private static final Set<ConfigOption<?>> requiredOptions =
+ Stream.of(HOSTS_OPTION, INDEX_OPTION).collect(Collectors.toSet());
+ private static final Set<ConfigOption<?>> optionalOptions =
+ Stream.of(
+ KEY_DELIMITER_OPTION,
+ FAILURE_HANDLER_OPTION,
+ FLUSH_ON_CHECKPOINT_OPTION,
+ BULK_FLASH_MAX_SIZE_OPTION,
+ BULK_FLUSH_MAX_ACTIONS_OPTION,
+ BULK_FLUSH_INTERVAL_OPTION,
+ BULK_FLUSH_BACKOFF_TYPE_OPTION,
+ BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION,
+ BULK_FLUSH_BACKOFF_DELAY_OPTION,
+ CONNECTION_MAX_RETRY_TIMEOUT_OPTION,
+ CONNECTION_PATH_PREFIX,
+ FORMAT_OPTION,
+ PASSWORD_OPTION,
+ USERNAME_OPTION)
+ .collect(Collectors.toSet());
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ TableSchema tableSchema = context.getCatalogTable().getSchema();
+ ElasticsearchValidationUtils.validatePrimaryKey(tableSchema);
+
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+
+ final EncodingFormat<SerializationSchema<RowData>> format =
+ helper.discoverEncodingFormat(SerializationFormatFactory.class, FORMAT_OPTION);
+
+ helper.validate();
+ Configuration configuration = new Configuration();
+ context.getCatalogTable().getOptions().forEach(configuration::setString);
+ Elasticsearch7Configuration config =
+ new Elasticsearch7Configuration(configuration, context.getClassLoader());
+
+ validate(config, configuration);
+
+ return new Elasticsearch7DynamicSink(
+ format,
+ config,
+ TableSchemaUtils.getPhysicalSchema(tableSchema),
+ getLocalTimeZoneId(context.getConfiguration()));
+ }
+
+ ZoneId getLocalTimeZoneId(ReadableConfig readableConfig) {
+ final String zone = readableConfig.get(TableConfigOptions.LOCAL_TIME_ZONE);
+ final ZoneId zoneId =
+ TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)
+ ? ZoneId.systemDefault()
+ : ZoneId.of(zone);
+
+ return zoneId;
+ }
+
+ private void validate(Elasticsearch7Configuration config, Configuration originalConfiguration) {
+ config.getFailureHandler(); // checks if we can instantiate the custom failure handler
+ config.getHosts(); // validate hosts
+ validate(
+ config.getIndex().length() >= 1,
+ () -> String.format("'%s' must not be empty", INDEX_OPTION.key()));
+ int maxActions = config.getBulkFlushMaxActions();
+ validate(
+ maxActions == -1 || maxActions >= 1,
+ () ->
+ String.format(
+ "'%s' must be at least 1. Got: %s",
+ BULK_FLUSH_MAX_ACTIONS_OPTION.key(), maxActions));
+ long maxSize = config.getBulkFlushMaxByteSize();
+ long mb1 = 1024 * 1024;
+ validate(
+ maxSize == -1 || (maxSize >= mb1 && maxSize % mb1 == 0),
+ () ->
+ String.format(
+ "'%s' must be in MB granularity. Got: %s",
+ BULK_FLASH_MAX_SIZE_OPTION.key(),
+ originalConfiguration
+ .get(BULK_FLASH_MAX_SIZE_OPTION)
+ .toHumanReadableString()));
+ validate(
+ config.getBulkFlushBackoffRetries().map(retries -> retries >= 1).orElse(true),
+ () ->
+ String.format(
+ "'%s' must be at least 1. Got: %s",
+ BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION.key(),
+ config.getBulkFlushBackoffRetries().get()));
+ if (config.getUsername().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getUsername().get())) {
+ validate(
+ config.getPassword().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getPassword().get()),
+ () ->
+ String.format(
+ "'%s' and '%s' must be set at the same time. Got: username '%s' and password '%s'",
+ USERNAME_OPTION.key(),
+ PASSWORD_OPTION.key(),
+ config.getUsername().get(),
+ config.getPassword().orElse("")));
+ }
+ }
+
+ private static void validate(boolean condition, Supplier<String> message) {
+ if (!condition) {
+ throw new ValidationException(message.get());
+ }
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return "elasticsearch-7";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return requiredOptions;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return optionalOptions;
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-connectors/flink-connector-elasticsearch7/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
index 9e189e3..10e4846 100644
--- a/flink-connectors/flink-connector-elasticsearch7/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -13,4 +13,4 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-org.apache.flink.connector.elasticsearch.table.Elasticsearch7DynamicSinkFactory
+org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7DynamicSinkFactory
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkFactoryTest.java b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkFactoryTest.java
new file mode 100644
index 0000000..3ff21ed
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkFactoryTest.java
@@ -0,0 +1,234 @@
+/*
+ * 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.table;
+
+import org.apache.flink.api.common.typeutils.base.VoidSerializer;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.TestContext.context;
+
+/** Tests for validation in {@link Elasticsearch7DynamicSinkFactory}. */
+public class Elasticsearch7DynamicSinkFactoryTest extends TestLogger {
+ @Rule public ExpectedException thrown = ExpectedException.none();
+
+ @Test
+ public void validateEmptyConfiguration() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "One or more required options are missing.\n"
+ + "\n"
+ + "Missing required options are:\n"
+ + "\n"
+ + "hosts\n"
+ + "index");
+ sinkFactory.createDynamicTableSink(context().build());
+ }
+
+ @Test
+ public void validateWrongIndex() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage("'index' must not be empty");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption("index", "")
+ .withOption("hosts", "http://localhost:12345")
+ .build());
+ }
+
+ @Test
+ public void validateWrongHosts() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "Could not parse host 'wrong-host' in option 'hosts'. It should follow the format 'http://host_name:port'.");
+ sinkFactory.createDynamicTableSink(
+ context().withOption("index", "MyIndex").withOption("hosts", "wrong-host").build());
+ }
+
+ @Test
+ public void validateWrongFlushSize() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "'sink.bulk-flush.max-size' must be in MB granularity. Got: 1024 bytes");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLASH_MAX_SIZE_OPTION.key(),
+ "1kb")
+ .build());
+ }
+
+ @Test
+ public void validateWrongRetries() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage("'sink.bulk-flush.backoff.max-retries' must be at least 1. Got: 0");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION
+ .key(),
+ "0")
+ .build());
+ }
+
+ @Test
+ public void validateWrongMaxActions() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage("'sink.bulk-flush.max-actions' must be at least 1. Got: -2");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION.key(),
+ "-2")
+ .build());
+ }
+
+ @Test
+ public void validateWrongBackoffDelay() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage("Invalid value for option 'sink.bulk-flush.backoff.delay'.");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION.key(),
+ "-1s")
+ .build());
+ }
+
+ @Test
+ public void validatePrimaryKeyOnIllegalColumn() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "The table has a primary key on columns of illegal types: "
+ + "[ARRAY, MAP, MULTISET, ROW, RAW, VARBINARY].\n"
+ + " Elasticsearch sink does not support primary keys on columns of types: "
+ + "[ARRAY, MAP, MULTISET, STRUCTURED_TYPE, ROW, RAW, BINARY, VARBINARY].");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withSchema(
+ new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("a", DataTypes.BIGINT().notNull()),
+ Column.physical(
+ "b",
+ DataTypes.ARRAY(
+ DataTypes.BIGINT()
+ .notNull())
+ .notNull()),
+ Column.physical(
+ "c",
+ DataTypes.MAP(
+ DataTypes.BIGINT(),
+ DataTypes.STRING())
+ .notNull()),
+ Column.physical(
+ "d",
+ DataTypes.MULTISET(
+ DataTypes.BIGINT()
+ .notNull())
+ .notNull()),
+ Column.physical(
+ "e",
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "a",
+ DataTypes.BIGINT()))
+ .notNull()),
+ Column.physical(
+ "f",
+ DataTypes.RAW(
+ Void.class,
+ VoidSerializer.INSTANCE)
+ .notNull()),
+ Column.physical("g", DataTypes.BYTES().notNull())),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey(
+ "name",
+ Arrays.asList("a", "b", "c", "d", "e", "f", "g"))))
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION.key(),
+ "1s")
+ .build());
+ }
+
+ @Test
+ public void validateWrongCredential() {
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ thrown.expect(ValidationException.class);
+ thrown.expectMessage(
+ "'username' and 'password' must be set at the same time. Got: username 'username' and password ''");
+ sinkFactory.createDynamicTableSink(
+ context()
+ .withOption(ElasticsearchConnectorOptions.INDEX_OPTION.key(), "MyIndex")
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ "http://localhost:1234")
+ .withOption(ElasticsearchConnectorOptions.USERNAME_OPTION.key(), "username")
+ .withOption(ElasticsearchConnectorOptions.PASSWORD_OPTION.key(), "")
+ .build());
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java
new file mode 100644
index 0000000..fdf7a3a
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java
@@ -0,0 +1,376 @@
+/*
+ * 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.table;
+
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.TestLogger;
+
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.search.SearchHits;
+import org.elasticsearch.transport.client.PreBuiltTransportClient;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import static org.apache.flink.streaming.connectors.elasticsearch.table.TestContext.context;
+import static org.apache.flink.table.api.Expressions.row;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/** IT tests for {@link Elasticsearch7DynamicSink}. */
+public class Elasticsearch7DynamicSinkITCase extends TestLogger {
+
+ @ClassRule
+ public static ElasticsearchContainer elasticsearchContainer =
+ new ElasticsearchContainer(DockerImageName.parse(DockerImageVersions.ELASTICSEARCH_7));
+
+ @SuppressWarnings("deprecation")
+ protected final Client getClient() {
+ TransportAddress transportAddress =
+ new TransportAddress(elasticsearchContainer.getTcpHost());
+ String expectedClusterName = "docker-cluster";
+ Settings settings = Settings.builder().put("cluster.name", expectedClusterName).build();
+ return new PreBuiltTransportClient(settings).addTransportAddress(transportAddress);
+ }
+
+ @Test
+ public void testWritingDocuments() throws Exception {
+ ResolvedSchema schema =
+ new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("a", DataTypes.BIGINT().notNull()),
+ Column.physical("b", DataTypes.TIME()),
+ Column.physical("c", DataTypes.STRING().notNull()),
+ Column.physical("d", DataTypes.FLOAT()),
+ Column.physical("e", DataTypes.TINYINT().notNull()),
+ Column.physical("f", DataTypes.DATE()),
+ Column.physical("g", DataTypes.TIMESTAMP().notNull())),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey("name", Arrays.asList("a", "g")));
+
+ GenericRowData rowData =
+ GenericRowData.of(
+ 1L,
+ 12345,
+ StringData.fromString("ABCDE"),
+ 12.12f,
+ (byte) 2,
+ 12345,
+ TimestampData.fromLocalDateTime(
+ LocalDateTime.parse("2012-12-12T12:12:12")));
+
+ String index = "writing-documents";
+ Elasticsearch7DynamicSinkFactory sinkFactory = new Elasticsearch7DynamicSinkFactory();
+
+ SinkFunctionProvider sinkRuntimeProvider =
+ (SinkFunctionProvider)
+ sinkFactory
+ .createDynamicTableSink(
+ context()
+ .withSchema(schema)
+ .withOption(
+ ElasticsearchConnectorOptions.INDEX_OPTION
+ .key(),
+ index)
+ .withOption(
+ ElasticsearchConnectorOptions.HOSTS_OPTION
+ .key(),
+ elasticsearchContainer.getHttpHostAddress())
+ .withOption(
+ ElasticsearchConnectorOptions
+ .FLUSH_ON_CHECKPOINT_OPTION
+ .key(),
+ "false")
+ .build())
+ .getSinkRuntimeProvider(new MockContext());
+
+ SinkFunction<RowData> sinkFunction = sinkRuntimeProvider.createSinkFunction();
+ StreamExecutionEnvironment environment =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+ environment.setParallelism(4);
+
+ rowData.setRowKind(RowKind.UPDATE_AFTER);
+ environment.<RowData>fromElements(rowData).addSink(sinkFunction);
+ environment.execute();
+
+ Client client = getClient();
+ Map<String, Object> response =
+ client.get(new GetRequest(index, "1_2012-12-12T12:12:12")).actionGet().getSource();
+ Map<Object, Object> expectedMap = new HashMap<>();
+ expectedMap.put("a", 1);
+ expectedMap.put("b", "00:00:12");
+ expectedMap.put("c", "ABCDE");
+ expectedMap.put("d", 12.12d);
+ expectedMap.put("e", 2);
+ expectedMap.put("f", "2003-10-20");
+ expectedMap.put("g", "2012-12-12 12:12:12");
+ assertThat(response, equalTo(expectedMap));
+ }
+
+ @Test
+ public void testWritingDocumentsFromTableApi() throws Exception {
+ TableEnvironment tableEnvironment =
+ TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+ String index = "table-api";
+ tableEnvironment.executeSql(
+ "CREATE TABLE esTable ("
+ + "a BIGINT NOT NULL,\n"
+ + "b TIME,\n"
+ + "c STRING NOT NULL,\n"
+ + "d FLOAT,\n"
+ + "e TINYINT NOT NULL,\n"
+ + "f DATE,\n"
+ + "g TIMESTAMP NOT NULL,"
+ + "h as a + 2,\n"
+ + "PRIMARY KEY (a, g) NOT ENFORCED\n"
+ + ")\n"
+ + "WITH (\n"
+ + String.format("'%s'='%s',\n", "connector", "elasticsearch-7")
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.INDEX_OPTION.key(), index)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ + String.format(
+ "'%s'='%s'\n",
+ ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
+ "false")
+ + ")");
+
+ tableEnvironment
+ .fromValues(
+ row(
+ 1L,
+ LocalTime.ofNanoOfDay(12345L * 1_000_000L),
+ "ABCDE",
+ 12.12f,
+ (byte) 2,
+ LocalDate.ofEpochDay(12345),
+ LocalDateTime.parse("2012-12-12T12:12:12")))
+ .executeInsert("esTable")
+ .await();
+
+ Client client = getClient();
+ Map<String, Object> response =
+ client.get(new GetRequest(index, "1_2012-12-12T12:12:12")).actionGet().getSource();
+ Map<Object, Object> expectedMap = new HashMap<>();
+ expectedMap.put("a", 1);
+ expectedMap.put("b", "00:00:12");
+ expectedMap.put("c", "ABCDE");
+ expectedMap.put("d", 12.12d);
+ expectedMap.put("e", 2);
+ expectedMap.put("f", "2003-10-20");
+ expectedMap.put("g", "2012-12-12 12:12:12");
+ assertThat(response, equalTo(expectedMap));
+ }
+
+ @Test
+ public void testWritingDocumentsNoPrimaryKey() throws Exception {
+ TableEnvironment tableEnvironment =
+ TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+ String index = "no-primary-key";
+ tableEnvironment.executeSql(
+ "CREATE TABLE esTable ("
+ + "a BIGINT NOT NULL,\n"
+ + "b TIME,\n"
+ + "c STRING NOT NULL,\n"
+ + "d FLOAT,\n"
+ + "e TINYINT NOT NULL,\n"
+ + "f DATE,\n"
+ + "g TIMESTAMP NOT NULL\n"
+ + ")\n"
+ + "WITH (\n"
+ + String.format("'%s'='%s',\n", "connector", "elasticsearch-7")
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.INDEX_OPTION.key(), index)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ + String.format(
+ "'%s'='%s'\n",
+ ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
+ "false")
+ + ")");
+
+ tableEnvironment
+ .fromValues(
+ row(
+ 1L,
+ LocalTime.ofNanoOfDay(12345L * 1_000_000L),
+ "ABCDE",
+ 12.12f,
+ (byte) 2,
+ LocalDate.ofEpochDay(12345),
+ LocalDateTime.parse("2012-12-12T12:12:12")),
+ row(
+ 2L,
+ LocalTime.ofNanoOfDay(12345L * 1_000_000L),
+ "FGHIJK",
+ 13.13f,
+ (byte) 4,
+ LocalDate.ofEpochDay(12345),
+ LocalDateTime.parse("2013-12-12T13:13:13")))
+ .executeInsert("esTable")
+ .await();
+
+ Client client = getClient();
+
+ // search API does not return documents that were not indexed, we might need to query
+ // the index a few times
+ Deadline deadline = Deadline.fromNow(Duration.ofSeconds(30));
+ SearchHits hits;
+ do {
+ hits = client.prepareSearch(index).execute().actionGet().getHits();
+ if (hits.getTotalHits().value < 2) {
+ Thread.sleep(200);
+ }
+ } while (hits.getTotalHits().value < 2 && deadline.hasTimeLeft());
+
+ if (hits.getTotalHits().value < 2) {
+ throw new AssertionError("Could not retrieve results from Elasticsearch.");
+ }
+
+ HashSet<Map<String, Object>> resultSet = new HashSet<>();
+ resultSet.add(hits.getAt(0).getSourceAsMap());
+ resultSet.add(hits.getAt(1).getSourceAsMap());
+ Map<Object, Object> expectedMap1 = new HashMap<>();
+ expectedMap1.put("a", 1);
+ expectedMap1.put("b", "00:00:12");
+ expectedMap1.put("c", "ABCDE");
+ expectedMap1.put("d", 12.12d);
+ expectedMap1.put("e", 2);
+ expectedMap1.put("f", "2003-10-20");
+ expectedMap1.put("g", "2012-12-12 12:12:12");
+ Map<Object, Object> expectedMap2 = new HashMap<>();
+ expectedMap2.put("a", 2);
+ expectedMap2.put("b", "00:00:12");
+ expectedMap2.put("c", "FGHIJK");
+ expectedMap2.put("d", 13.13d);
+ expectedMap2.put("e", 4);
+ expectedMap2.put("f", "2003-10-20");
+ expectedMap2.put("g", "2013-12-12 13:13:13");
+ HashSet<Map<Object, Object>> expectedSet = new HashSet<>();
+ expectedSet.add(expectedMap1);
+ expectedSet.add(expectedMap2);
+ assertThat(resultSet, equalTo(expectedSet));
+ }
+
+ @Test
+ public void testWritingDocumentsWithDynamicIndex() throws Exception {
+ TableEnvironment tableEnvironment =
+ TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+
+ String index = "dynamic-index-{b|yyyy-MM-dd}";
+ tableEnvironment.executeSql(
+ "CREATE TABLE esTable ("
+ + "a BIGINT NOT NULL,\n"
+ + "b TIMESTAMP NOT NULL,\n"
+ + "PRIMARY KEY (a) NOT ENFORCED\n"
+ + ")\n"
+ + "WITH (\n"
+ + String.format("'%s'='%s',\n", "connector", "elasticsearch-7")
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.INDEX_OPTION.key(), index)
+ + String.format(
+ "'%s'='%s',\n",
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ + String.format(
+ "'%s'='%s'\n",
+ ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
+ "false")
+ + ")");
+
+ tableEnvironment
+ .fromValues(row(1L, LocalDateTime.parse("2012-12-12T12:12:12")))
+ .executeInsert("esTable")
+ .await();
+
+ Client client = getClient();
+ Map<String, Object> response =
+ client.get(new GetRequest("dynamic-index-2012-12-12", "1")).actionGet().getSource();
+ Map<Object, Object> expectedMap = new HashMap<>();
+ expectedMap.put("a", 1);
+ expectedMap.put("b", "2012-12-12 12:12:12");
+ assertThat(response, equalTo(expectedMap));
+ }
+
+ private static class MockContext implements DynamicTableSink.Context {
+ @Override
+ public boolean isBounded() {
+ return false;
+ }
+
+ @Override
+ public TypeInformation<?> createTypeInformation(DataType consumedDataType) {
+ return null;
+ }
+
+ @Override
+ public TypeInformation<?> createTypeInformation(LogicalType consumedLogicalType) {
+ return null;
+ }
+
+ @Override
+ public DynamicTableSink.DataStructureConverter createDataStructureConverter(
+ DataType consumedDataType) {
+ return null;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkTest.java b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkTest.java
new file mode 100644
index 0000000..2928c8d
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkTest.java
@@ -0,0 +1,298 @@
+/*
+ * 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.table;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
+import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
+import org.apache.flink.streaming.connectors.elasticsearch.util.NoOpFailureHandler;
+import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.ActionRequest;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.time.ZoneId;
+import java.util.List;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link Elasticsearch7DynamicSink} parameters. */
+public class Elasticsearch7DynamicSinkTest extends TestLogger {
+
+ private static final String FIELD_KEY = "key";
+ private static final String FIELD_FRUIT_NAME = "fruit_name";
+ private static final String FIELD_COUNT = "count";
+ private static final String FIELD_TS = "ts";
+
+ private static final String HOSTNAME = "host1";
+ private static final int PORT = 1234;
+ private static final String SCHEMA = "https";
+ private static final String INDEX = "MyIndex";
+ private static final String DOC_TYPE = "MyType";
+ private static final String USERNAME = "username";
+ private static final String PASSWORD = "password";
+
+ @Test
+ public void testBuilder() {
+ final TableSchema schema = createTestSchema();
+
+ BuilderProvider provider = new BuilderProvider();
+ final Elasticsearch7DynamicSink testSink =
+ new Elasticsearch7DynamicSink(
+ new DummyEncodingFormat(),
+ new Elasticsearch7Configuration(
+ getConfig(), this.getClass().getClassLoader()),
+ schema,
+ ZoneId.systemDefault(),
+ provider);
+
+ testSink.getSinkRuntimeProvider(new MockSinkContext()).createSinkFunction();
+
+ verify(provider.builderSpy).setFailureHandler(new DummyFailureHandler());
+ verify(provider.builderSpy).setBulkFlushBackoff(true);
+ verify(provider.builderSpy)
+ .setBulkFlushBackoffType(ElasticsearchSinkBase.FlushBackoffType.EXPONENTIAL);
+ verify(provider.builderSpy).setBulkFlushBackoffDelay(123);
+ verify(provider.builderSpy).setBulkFlushBackoffRetries(3);
+ verify(provider.builderSpy).setBulkFlushInterval(100);
+ verify(provider.builderSpy).setBulkFlushMaxActions(1000);
+ verify(provider.builderSpy).setBulkFlushMaxSizeMb(1);
+ verify(provider.builderSpy)
+ .setRestClientFactory(
+ new Elasticsearch7DynamicSink.DefaultRestClientFactory("/myapp"));
+ verify(provider.sinkSpy).disableFlushOnCheckpoint();
+ }
+
+ @Test
+ public void testDefaultConfig() {
+ final TableSchema schema = createTestSchema();
+ Configuration configuration = new Configuration();
+ configuration.setString(ElasticsearchConnectorOptions.INDEX_OPTION.key(), INDEX);
+ configuration.setString(ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), DOC_TYPE);
+ configuration.setString(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ SCHEMA + "://" + HOSTNAME + ":" + PORT);
+
+ BuilderProvider provider = new BuilderProvider();
+ final Elasticsearch7DynamicSink testSink =
+ new Elasticsearch7DynamicSink(
+ new DummyEncodingFormat(),
+ new Elasticsearch7Configuration(
+ configuration, this.getClass().getClassLoader()),
+ schema,
+ ZoneId.systemDefault(),
+ provider);
+
+ testSink.getSinkRuntimeProvider(new MockSinkContext()).createSinkFunction();
+
+ verify(provider.builderSpy).setFailureHandler(new NoOpFailureHandler());
+ verify(provider.builderSpy).setBulkFlushBackoff(false);
+ verify(provider.builderSpy).setBulkFlushInterval(1000);
+ verify(provider.builderSpy).setBulkFlushMaxActions(1000);
+ verify(provider.builderSpy).setBulkFlushMaxSizeMb(2);
+ verify(provider.builderSpy)
+ .setRestClientFactory(new Elasticsearch7DynamicSink.DefaultRestClientFactory(null));
+ verify(provider.sinkSpy, never()).disableFlushOnCheckpoint();
+ }
+
+ @Test
+ public void testAuthConfig() {
+ final TableSchema schema = createTestSchema();
+ Configuration configuration = new Configuration();
+ configuration.setString(ElasticsearchConnectorOptions.INDEX_OPTION.key(), INDEX);
+ configuration.setString(ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), DOC_TYPE);
+ configuration.setString(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ SCHEMA + "://" + HOSTNAME + ":" + PORT);
+ configuration.setString(ElasticsearchConnectorOptions.USERNAME_OPTION.key(), USERNAME);
+ configuration.setString(ElasticsearchConnectorOptions.PASSWORD_OPTION.key(), PASSWORD);
+
+ BuilderProvider provider = new BuilderProvider();
+ final Elasticsearch7DynamicSink testSink =
+ new Elasticsearch7DynamicSink(
+ new DummyEncodingFormat(),
+ new Elasticsearch7Configuration(
+ configuration, this.getClass().getClassLoader()),
+ schema,
+ ZoneId.systemDefault(),
+ provider);
+
+ testSink.getSinkRuntimeProvider(new MockSinkContext()).createSinkFunction();
+
+ verify(provider.builderSpy).setFailureHandler(new NoOpFailureHandler());
+ verify(provider.builderSpy).setBulkFlushBackoff(false);
+ verify(provider.builderSpy).setBulkFlushInterval(1000);
+ verify(provider.builderSpy).setBulkFlushMaxActions(1000);
+ verify(provider.builderSpy).setBulkFlushMaxSizeMb(2);
+ verify(provider.builderSpy)
+ .setRestClientFactory(
+ new Elasticsearch7DynamicSink.AuthRestClientFactory(
+ null, USERNAME, PASSWORD));
+ verify(provider.sinkSpy, never()).disableFlushOnCheckpoint();
+ }
+
+ private Configuration getConfig() {
+ Configuration configuration = new Configuration();
+ configuration.setString(ElasticsearchConnectorOptions.INDEX_OPTION.key(), INDEX);
+ configuration.setString(ElasticsearchConnectorOptions.DOCUMENT_TYPE_OPTION.key(), DOC_TYPE);
+ configuration.setString(
+ ElasticsearchConnectorOptions.HOSTS_OPTION.key(),
+ SCHEMA + "://" + HOSTNAME + ":" + PORT);
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_TYPE_OPTION.key(), "exponential");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION.key(), "123");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION.key(), "3");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_INTERVAL_OPTION.key(), "100");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION.key(), "1000");
+ configuration.setString(
+ ElasticsearchConnectorOptions.BULK_FLASH_MAX_SIZE_OPTION.key(), "1mb");
+ configuration.setString(
+ ElasticsearchConnectorOptions.CONNECTION_PATH_PREFIX.key(), "/myapp");
+ configuration.setString(
+ ElasticsearchConnectorOptions.FAILURE_HANDLER_OPTION.key(),
+ DummyFailureHandler.class.getName());
+ configuration.setString(
+ ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(), "false");
+ return configuration;
+ }
+
+ private static class BuilderProvider
+ implements Elasticsearch7DynamicSink.ElasticSearchBuilderProvider {
+ public ElasticsearchSink.Builder<RowData> builderSpy;
+ public ElasticsearchSink<RowData> sinkSpy;
+
+ @Override
+ public ElasticsearchSink.Builder<RowData> createBuilder(
+ List<HttpHost> httpHosts, RowElasticsearchSinkFunction upsertSinkFunction) {
+ builderSpy =
+ Mockito.spy(new ElasticsearchSink.Builder<>(httpHosts, upsertSinkFunction));
+ doAnswer(
+ invocation -> {
+ sinkSpy =
+ Mockito.spy(
+ (ElasticsearchSink<RowData>)
+ invocation.callRealMethod());
+ return sinkSpy;
+ })
+ .when(builderSpy)
+ .build();
+
+ return builderSpy;
+ }
+ }
+
+ private TableSchema createTestSchema() {
+ return TableSchema.builder()
+ .field(FIELD_KEY, DataTypes.BIGINT())
+ .field(FIELD_FRUIT_NAME, DataTypes.STRING())
+ .field(FIELD_COUNT, DataTypes.DECIMAL(10, 4))
+ .field(FIELD_TS, DataTypes.TIMESTAMP(3))
+ .build();
+ }
+
+ private static class DummySerializationSchema implements SerializationSchema<RowData> {
+
+ private static final DummySerializationSchema INSTANCE = new DummySerializationSchema();
+
+ @Override
+ public byte[] serialize(RowData element) {
+ return new byte[0];
+ }
+ }
+
+ private static class DummyEncodingFormat
+ implements EncodingFormat<SerializationSchema<RowData>> {
+ @Override
+ public SerializationSchema<RowData> createRuntimeEncoder(
+ DynamicTableSink.Context context, DataType consumedDataType) {
+ return DummySerializationSchema.INSTANCE;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return null;
+ }
+ }
+
+ private static class MockSinkContext implements DynamicTableSink.Context {
+ @Override
+ public boolean isBounded() {
+ return false;
+ }
+
+ @Override
+ public TypeInformation<?> createTypeInformation(DataType consumedDataType) {
+ return null;
+ }
+
+ @Override
+ public TypeInformation<?> createTypeInformation(LogicalType consumedLogicalType) {
+ return null;
+ }
+
+ @Override
+ public DynamicTableSink.DataStructureConverter createDataStructureConverter(
+ DataType consumedDataType) {
+ return null;
+ }
+ }
+
+ /** Custom failure handler for testing. */
+ public static class DummyFailureHandler implements ActionRequestFailureHandler {
+
+ @Override
+ public void onFailure(
+ ActionRequest action,
+ Throwable failure,
+ int restStatusCode,
+ RequestIndexer indexer) {
+ // do nothing
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof DummyFailureHandler;
+ }
+
+ @Override
+ public int hashCode() {
+ return DummyFailureHandler.class.hashCode();
+ }
+ }
+}