You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/10/29 18:28:04 UTC
[flink] branch master updated (d6f8beb -> 980acdb)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from d6f8beb [hotfix][docs] Fix typo in network_mem_tuning.md
new 4335044 [FLINK-24327][connectors/elasticsearch] Integrate unified Elasticsearch 7 sink with Table API
new 980acdb [hotfix][table-common] fix DataType methods in DynamicTableFactory docstring
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../table/ElasticsearchValidationUtils.java | 54 ++++
.../elasticsearch/table/IndexGeneratorFactory.java | 15 ++
.../elasticsearch/table/KeyExtractor.java | 20 +-
...dexGenerator.java => LogicalTypeWithIndex.java} | 18 +-
.../elasticsearch/table/SerializableFunction.java | 9 +-
.../sink/ElasticsearchSinkBuilder.java | 2 +-
.../elasticsearch/sink/ElasticsearchWriter.java | 1 +
.../table/Elasticsearch7Configuration.java | 75 +++++-
.../table/Elasticsearch7ConnectorOptions.java} | 77 ++----
.../table/Elasticsearch7DynamicSink.java | 290 +++++---------------
.../table/Elasticsearch7DynamicSinkFactory.java | 84 +++---
.../table/RowElasticsearchEmitter.java} | 77 ++----
.../Elasticsearch7DynamicSinkFactoryTest.java | 4 +-
.../table/Elasticsearch7DynamicSinkITCase.java | 75 ++----
.../table/Elasticsearch7DynamicSinkTest.java | 294 ---------------------
.../flink/table/factories/DynamicTableFactory.java | 4 +-
16 files changed, 354 insertions(+), 745 deletions(-)
copy flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/{StaticIndexGenerator.java => LogicalTypeWithIndex.java} (70%)
copy flink-core/src/main/java/org/apache/flink/api/common/Archiveable.java => flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/SerializableFunction.java (82%)
copy flink-connectors/{flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchConnectorOptions.java => flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7ConnectorOptions.java} (61%)
copy flink-connectors/{flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java => flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchEmitter.java} (52%)
delete mode 100644 flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkTest.java
[flink] 01/02: [FLINK-24327][connectors/elasticsearch] Integrate
unified Elasticsearch 7 sink with Table API
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4335044245da4d1129e3abababfc7399cecfaf4e
Author: Alexander Preuß <11...@users.noreply.github.com>
AuthorDate: Thu Oct 7 11:00:57 2021 +0200
[FLINK-24327][connectors/elasticsearch] Integrate unified Elasticsearch 7 sink with Table API
---
.../table/ElasticsearchValidationUtils.java | 54 ++++
.../elasticsearch/table/IndexGeneratorFactory.java | 15 ++
.../elasticsearch/table/KeyExtractor.java | 20 +-
.../elasticsearch/table/LogicalTypeWithIndex.java | 31 +++
.../elasticsearch/table/SerializableFunction.java | 24 ++
.../sink/ElasticsearchSinkBuilder.java | 2 +-
.../elasticsearch/sink/ElasticsearchWriter.java | 1 +
.../table/Elasticsearch7Configuration.java | 75 +++++-
.../table/Elasticsearch7ConnectorOptions.java | 124 +++++++++
.../table/Elasticsearch7DynamicSink.java | 290 +++++---------------
.../table/Elasticsearch7DynamicSinkFactory.java | 84 +++---
.../table/RowElasticsearchEmitter.java | 101 +++++++
.../Elasticsearch7DynamicSinkFactoryTest.java | 4 +-
.../table/Elasticsearch7DynamicSinkITCase.java | 75 ++----
.../table/Elasticsearch7DynamicSinkTest.java | 294 ---------------------
15 files changed, 588 insertions(+), 606 deletions(-)
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
index 6452d00..b2ea6ff 100644
--- 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
@@ -21,6 +21,7 @@ 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.DataType;
import org.apache.flink.table.types.logical.DistinctType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeFamily;
@@ -36,6 +37,7 @@ import java.util.stream.Collectors;
class ElasticsearchValidationUtils {
private static final Set<LogicalTypeRoot> ILLEGAL_PRIMARY_KEY_TYPES = new LinkedHashSet<>();
+ private static final Set<LogicalTypeRoot> ALLOWED_PRIMARY_KEY_TYPES = new LinkedHashSet<>();
static {
ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.ARRAY);
@@ -46,6 +48,24 @@ class ElasticsearchValidationUtils {
ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.RAW);
ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.BINARY);
ILLEGAL_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.VARBINARY);
+
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.CHAR);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.VARCHAR);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.BOOLEAN);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.DECIMAL);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.TINYINT);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.SMALLINT);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.INTEGER);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.BIGINT);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.FLOAT);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.DOUBLE);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.DATE);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.TIME_WITHOUT_TIME_ZONE);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.TIMESTAMP_WITH_TIME_ZONE);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.INTERVAL_YEAR_MONTH);
+ ALLOWED_PRIMARY_KEY_TYPES.add(LogicalTypeRoot.INTERVAL_DAY_TIME);
}
/**
@@ -56,6 +76,7 @@ class ElasticsearchValidationUtils {
* The illegal types are mostly {@link LogicalTypeFamily#COLLECTION} types and {@link
* LogicalTypeRoot#RAW} type.
*/
+ @Deprecated
public static void validatePrimaryKey(TableSchema schema) {
schema.getPrimaryKey()
.ifPresent(
@@ -90,5 +111,38 @@ class ElasticsearchValidationUtils {
});
}
+ /**
+ * Checks that the table does not have a 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(DataType primaryKeyDataType) {
+ List<DataType> fieldDataTypes = DataType.getFieldDataTypes(primaryKeyDataType);
+ List<LogicalTypeRoot> illegalTypes =
+ fieldDataTypes.stream()
+ .map(DataType::getLogicalType)
+ .map(
+ logicalType -> {
+ if (logicalType.is(LogicalTypeRoot.DISTINCT_TYPE)) {
+ return ((DistinctType) logicalType)
+ .getSourceType()
+ .getTypeRoot();
+ } else {
+ return logicalType.getTypeRoot();
+ }
+ })
+ .filter(t -> !ALLOWED_PRIMARY_KEY_TYPES.contains(t))
+ .collect(Collectors.toList());
+ if (!illegalTypes.isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "The table has a primary key on columns of illegal types: %s.",
+ illegalTypes));
+ }
+ }
+
private ElasticsearchValidationUtils() {}
}
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
index bf7ac64..a0b8faf 100644
--- 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
@@ -63,6 +63,21 @@ final class IndexGeneratorFactory {
private IndexGeneratorFactory() {}
+ public static IndexGenerator createIndexGenerator(
+ String index, List<String> fieldNames, List<DataType> dataTypes) {
+ final IndexHelper indexHelper = new IndexHelper();
+ if (indexHelper.checkIsDynamicIndex(index)) {
+ return createRuntimeIndexGenerator(
+ index,
+ fieldNames.toArray(new String[0]),
+ dataTypes.toArray(new DataType[0]),
+ indexHelper);
+ } else {
+ return new StaticIndexGenerator(index);
+ }
+ }
+
+ @Deprecated
public static IndexGenerator createIndexGenerator(String index, TableSchema schema) {
final IndexHelper indexHelper = new IndexHelper();
if (indexHelper.checkIsDynamicIndex(index)) {
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
index ae7c522..fb83327 100644
--- 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
@@ -37,7 +37,7 @@ import java.util.function.Function;
/** An extractor for a Elasticsearch key from a {@link RowData}. */
@Internal
-class KeyExtractor implements Function<RowData, String>, Serializable {
+class KeyExtractor implements SerializableFunction<RowData, String> {
private final FieldFormatter[] fieldFormatters;
private final String keyDelimiter;
@@ -81,6 +81,7 @@ class KeyExtractor implements Function<RowData, String>, Serializable {
}
}
+ @Deprecated
public static Function<RowData, String> createKeyExtractor(
TableSchema schema, String keyDelimiter) {
return schema.getPrimaryKey()
@@ -109,6 +110,23 @@ class KeyExtractor implements Function<RowData, String>, Serializable {
.orElseGet(() -> (Function<RowData, String> & Serializable) (row) -> null);
}
+ public static SerializableFunction<RowData, String> createKeyExtractor(
+ List<LogicalTypeWithIndex> primaryKeyTypesWithIndex, String keyDelimiter) {
+ if (!primaryKeyTypesWithIndex.isEmpty()) {
+ FieldFormatter[] formatters =
+ primaryKeyTypesWithIndex.stream()
+ .map(
+ logicalTypeWithIndex ->
+ toFormatter(
+ logicalTypeWithIndex.index,
+ logicalTypeWithIndex.logicalType))
+ .toArray(FieldFormatter[]::new);
+ return new KeyExtractor(formatters, keyDelimiter);
+ } else {
+ return (row) -> null;
+ }
+ }
+
private static FieldFormatter toFormatter(int index, LogicalType type) {
switch (type.getTypeRoot()) {
case DATE:
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/LogicalTypeWithIndex.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/LogicalTypeWithIndex.java
new file mode 100644
index 0000000..81c6347
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/LogicalTypeWithIndex.java
@@ -0,0 +1,31 @@
+/*
+ * 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.types.logical.LogicalType;
+
+class LogicalTypeWithIndex {
+ public final int index;
+ public final LogicalType logicalType;
+
+ LogicalTypeWithIndex(int index, LogicalType logicalType) {
+ this.index = index;
+ this.logicalType = logicalType;
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/SerializableFunction.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/SerializableFunction.java
new file mode 100644
index 0000000..be24313
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/SerializableFunction.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.elasticsearch.table;
+
+import java.io.Serializable;
+import java.util.function.Function;
+
+interface SerializableFunction<T, R> extends Function<T, R>, Serializable {}
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBuilder.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBuilder.java
index 68e30cb..233af3b 100644
--- a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBuilder.java
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBuilder.java
@@ -154,7 +154,7 @@ public class ElasticsearchSinkBuilder<IN> {
}
/**
- * Sets the type of back of to use when flushing bulk requests. The default bulk flush back off
+ * Sets the type of back off to use when flushing bulk requests. The default bulk flush back off
* type is {@link FlushBackoffType#NONE}.
*
* <p>Sets the amount of delay between each backoff attempt when flushing bulk requests, in
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchWriter.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchWriter.java
index b024bb2..c17dc52 100644
--- a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchWriter.java
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchWriter.java
@@ -150,6 +150,7 @@ class ElasticsearchWriter<IN> implements SinkWriter<IN, Void, Void> {
public void close() throws Exception {
closed = true;
emitter.close();
+ bulkProcessor.close();
client.close();
}
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
index 6bd28cf..276f77c 100644
--- 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
@@ -20,20 +20,87 @@ package org.apache.flink.streaming.connectors.elasticsearch.table;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.elasticsearch.sink.FlushBackoffType;
import org.apache.flink.table.api.ValidationException;
import org.apache.http.HttpHost;
+import java.time.Duration;
import java.util.List;
+import java.util.Optional;
import java.util.stream.Collectors;
-import static org.apache.flink.streaming.connectors.elasticsearch.table.ElasticsearchConnectorOptions.HOSTS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_BACKOFF_TYPE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_INTERVAL_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_MAX_SIZE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.CONNECTION_PATH_PREFIX_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.DELIVERY_GUARANTEE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.HOSTS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.INDEX_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.KEY_DELIMITER_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.PASSWORD_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.USERNAME_OPTION;
+import static org.apache.flink.util.Preconditions.checkNotNull;
/** Elasticsearch 7 specific configuration. */
@Internal
-final class Elasticsearch7Configuration extends ElasticsearchConfiguration {
- Elasticsearch7Configuration(ReadableConfig config, ClassLoader classLoader) {
- super(config, classLoader);
+final class Elasticsearch7Configuration {
+ protected final ReadableConfig config;
+
+ Elasticsearch7Configuration(ReadableConfig config) {
+ this.config = checkNotNull(config);
+ }
+
+ public int getBulkFlushMaxActions() {
+ return config.get(BULK_FLUSH_MAX_ACTIONS_OPTION);
+ }
+
+ public long getBulkFlushMaxByteSize() {
+ return config.get(BULK_FLUSH_MAX_SIZE_OPTION).getBytes();
+ }
+
+ public long getBulkFlushInterval() {
+ return config.get(BULK_FLUSH_INTERVAL_OPTION).toMillis();
+ }
+
+ public DeliveryGuarantee getDeliveryGuarantee() {
+ return config.get(DELIVERY_GUARANTEE_OPTION);
+ }
+
+ public Optional<String> getUsername() {
+ return config.getOptional(USERNAME_OPTION);
+ }
+
+ public Optional<String> getPassword() {
+ return config.getOptional(PASSWORD_OPTION);
+ }
+
+ public Optional<FlushBackoffType> getBulkFlushBackoffType() {
+ return config.getOptional(BULK_FLUSH_BACKOFF_TYPE_OPTION);
+ }
+
+ 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 String getIndex() {
+ return config.get(INDEX_OPTION);
+ }
+
+ public String getKeyDelimiter() {
+ return config.get(KEY_DELIMITER_OPTION);
+ }
+
+ public Optional<String> getPathPrefix() {
+ return config.getOptional(CONNECTION_PATH_PREFIX_OPTION);
}
public List<HttpHost> getHosts() {
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7ConnectorOptions.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7ConnectorOptions.java
new file mode 100644
index 0000000..6048e12
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7ConnectorOptions.java
@@ -0,0 +1,124 @@
+/*
+ * 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.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.elasticsearch.sink.FlushBackoffType;
+
+import java.time.Duration;
+import java.util.List;
+
+/** Options for the Elasticsearch connector. */
+@PublicEvolving
+public class Elasticsearch7ConnectorOptions {
+
+ private Elasticsearch7ConnectorOptions() {}
+
+ 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> 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<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_FLUSH_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<FlushBackoffType> BULK_FLUSH_BACKOFF_TYPE_OPTION =
+ ConfigOptions.key("sink.bulk-flush.backoff.strategy")
+ .enumType(FlushBackoffType.class)
+ .noDefaultValue()
+ .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<String> CONNECTION_PATH_PREFIX_OPTION =
+ 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.");
+
+ public static final ConfigOption<DeliveryGuarantee> DELIVERY_GUARANTEE_OPTION =
+ ConfigOptions.key("sink.delivery-guarantee")
+ .enumType(DeliveryGuarantee.class)
+ .defaultValue(DeliveryGuarantee.AT_LEAST_ONCE)
+ .withDescription("Optional delivery guarantee when committing.");
+}
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
index 0fe2dac..e6a03dc 100644
--- 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
@@ -19,89 +19,50 @@
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.connector.elasticsearch.sink.ElasticsearchSink;
+import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSinkBuilder;
+import org.apache.flink.connector.elasticsearch.sink.FlushBackoffType;
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.connector.sink.SinkProvider;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
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.util.List;
import java.util.Objects;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* 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 Elasticsearch7DynamicSink.Elasticsearch7RequestFactory();
private final EncodingFormat<SerializationSchema<RowData>> format;
- private final TableSchema schema;
+ private final DataType physicalRowDataType;
+ private final List<LogicalTypeWithIndex> primaryKeyLogicalTypesWithIndex;
private final Elasticsearch7Configuration config;
- public Elasticsearch7DynamicSink(
- EncodingFormat<SerializationSchema<RowData>> format,
- Elasticsearch7Configuration config,
- TableSchema schema) {
- this(format, config, schema, (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,
- ElasticSearchBuilderProvider builderProvider) {
- this.format = format;
- this.schema = schema;
- this.config = config;
- this.builderProvider = builderProvider;
+ List<LogicalTypeWithIndex> primaryKeyLogicalTypesWithIndex,
+ DataType physicalRowDataType) {
+ this.format = checkNotNull(format);
+ this.physicalRowDataType = checkNotNull(physicalRowDataType);
+ this.primaryKeyLogicalTypesWithIndex = checkNotNull(primaryKeyLogicalTypesWithIndex);
+ this.config = checkNotNull(config);
}
- // --------------------------------------------------------------
- // End of hack to make configuration testing possible
- // --------------------------------------------------------------
-
@Override
public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
ChangelogMode.Builder builder = ChangelogMode.newBuilder();
@@ -114,61 +75,59 @@ final class Elasticsearch7DynamicSink implements DynamicTableSink {
}
@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),
- 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);
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ SerializationSchema<RowData> format =
+ this.format.createRuntimeEncoder(context, physicalRowDataType);
+
+ final RowElasticsearchEmitter rowElasticsearchEmitter =
+ new RowElasticsearchEmitter(
+ IndexGeneratorFactory.createIndexGenerator(
+ config.getIndex(),
+ DataType.getFieldNames(physicalRowDataType),
+ DataType.getFieldDataTypes(physicalRowDataType)),
+ format,
+ XContentType.JSON,
+ KeyExtractor.createKeyExtractor(
+ primaryKeyLogicalTypesWithIndex, config.getKeyDelimiter()));
+
+ final ElasticsearchSinkBuilder<RowData> builder = ElasticsearchSink.builder();
+ builder.setEmitter(rowElasticsearchEmitter);
+ builder.setHosts(config.getHosts().toArray(new HttpHost[0]));
+ builder.setDeliveryGuarantee(config.getDeliveryGuarantee());
+ builder.setBulkFlushMaxActions(config.getBulkFlushMaxActions());
+ builder.setBulkFlushMaxSizeMb((int) (config.getBulkFlushMaxByteSize() >> 20));
+ builder.setBulkFlushInterval(config.getBulkFlushInterval());
+
+ if (config.getBulkFlushBackoffType().isPresent()) {
+ FlushBackoffType backoffType = config.getBulkFlushBackoffType().get();
+ int backoffMaxRetries = config.getBulkFlushBackoffRetries().get();
+ long backoffDelayMs = config.getBulkFlushBackoffDelay().get();
+
+ builder.setBulkFlushBackoffStrategy(backoffType, backoffMaxRetries, backoffDelayMs);
+ }
- // 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)));
- }
+ if (config.getUsername().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getUsername().get())) {
+ builder.setConnectionUsername(config.getUsername().get());
+ }
- final ElasticsearchSink<RowData> sink = builder.build();
+ if (config.getPassword().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getPassword().get())) {
+ builder.setConnectionPassword(config.getPassword().get());
+ }
- if (config.isDisableFlushOnCheckpoint()) {
- sink.disableFlushOnCheckpoint();
- }
+ if (config.getPathPrefix().isPresent()
+ && !StringUtils.isNullOrWhitespaceOnly(config.getPathPrefix().get())) {
+ builder.setConnectionPathPrefix(config.getPathPrefix().get());
+ }
- return sink;
- };
+ return SinkProvider.of(builder.build());
}
@Override
public DynamicTableSink copy() {
- return this;
+ return new Elasticsearch7DynamicSink(
+ format, config, primaryKeyLogicalTypesWithIndex, physicalRowDataType);
}
@Override
@@ -176,126 +135,6 @@ final class Elasticsearch7DynamicSink implements DynamicTableSink {
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) {
@@ -306,13 +145,14 @@ final class Elasticsearch7DynamicSink implements DynamicTableSink {
}
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);
+ && Objects.equals(physicalRowDataType, that.physicalRowDataType)
+ && Objects.equals(
+ primaryKeyLogicalTypesWithIndex, that.primaryKeyLogicalTypesWithIndex)
+ && Objects.equals(config, that.config);
}
@Override
public int hashCode() {
- return Objects.hash(format, schema, config, builderProvider);
+ return Objects.hash(format, physicalRowDataType, primaryKeyLogicalTypesWithIndex, config);
}
}
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
index e5fed88..0646d26 100644
--- 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
@@ -22,38 +22,41 @@ 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.table.api.TableSchema;
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.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.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.util.StringUtils;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
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;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_BACKOFF_DELAY_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_BACKOFF_MAX_RETRIES_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_BACKOFF_TYPE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_INTERVAL_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_MAX_ACTIONS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.BULK_FLUSH_MAX_SIZE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.CONNECTION_PATH_PREFIX_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.DELIVERY_GUARANTEE_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.FORMAT_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.HOSTS_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.INDEX_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.KEY_DELIMITER_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.PASSWORD_OPTION;
+import static org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch7ConnectorOptions.USERNAME_OPTION;
/** A {@link DynamicTableSinkFactory} for discovering {@link Elasticsearch7DynamicSink}. */
@Internal
@@ -63,25 +66,46 @@ public class Elasticsearch7DynamicSinkFactory implements DynamicTableSinkFactory
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_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,
+ CONNECTION_PATH_PREFIX_OPTION,
FORMAT_OPTION,
+ DELIVERY_GUARANTEE_OPTION,
PASSWORD_OPTION,
USERNAME_OPTION)
.collect(Collectors.toSet());
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
- TableSchema tableSchema = context.getCatalogTable().getSchema();
- ElasticsearchValidationUtils.validatePrimaryKey(tableSchema);
+ DataType physicalRowDataType = context.getPhysicalRowDataType();
+ int[] primaryKeyIndexes = context.getPrimaryKeyIndexes();
+ if (primaryKeyIndexes.length != 0) {
+ DataType pkDataType = DataType.projectFields(physicalRowDataType, primaryKeyIndexes);
+
+ ElasticsearchValidationUtils.validatePrimaryKey(pkDataType);
+ }
+
+ ResolvedSchema resolvedSchema = context.getCatalogTable().getResolvedSchema();
+ List<LogicalTypeWithIndex> primaryKeyLogicalTypesWithIndex =
+ Arrays.stream(primaryKeyIndexes)
+ .mapToObj(
+ index -> {
+ Optional<Column> column = resolvedSchema.getColumn(index);
+ if (!column.isPresent()) {
+ throw new IllegalStateException(
+ String.format(
+ "No primary key column found with index '%s'.",
+ index));
+ }
+ LogicalType logicalType =
+ column.get().getDataType().getLogicalType();
+ return new LogicalTypeWithIndex(index, logicalType);
+ })
+ .collect(Collectors.toList());
final FactoryUtil.TableFactoryHelper helper =
FactoryUtil.createTableFactoryHelper(this, context);
@@ -92,17 +116,15 @@ public class Elasticsearch7DynamicSinkFactory implements DynamicTableSinkFactory
helper.validate();
Configuration configuration = new Configuration();
context.getCatalogTable().getOptions().forEach(configuration::setString);
- Elasticsearch7Configuration config =
- new Elasticsearch7Configuration(configuration, context.getClassLoader());
+ Elasticsearch7Configuration config = new Elasticsearch7Configuration(configuration);
validate(config, configuration);
return new Elasticsearch7DynamicSink(
- format, config, TableSchemaUtils.getPhysicalSchema(tableSchema));
+ format, config, primaryKeyLogicalTypesWithIndex, physicalRowDataType);
}
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,
@@ -121,9 +143,9 @@ public class Elasticsearch7DynamicSinkFactory implements DynamicTableSinkFactory
() ->
String.format(
"'%s' must be in MB granularity. Got: %s",
- BULK_FLASH_MAX_SIZE_OPTION.key(),
+ BULK_FLUSH_MAX_SIZE_OPTION.key(),
originalConfiguration
- .get(BULK_FLASH_MAX_SIZE_OPTION)
+ .get(BULK_FLUSH_MAX_SIZE_OPTION)
.toHumanReadableString()));
validate(
config.getBulkFlushBackoffRetries().map(retries -> retries >= 1).orElse(true),
diff --git a/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchEmitter.java b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchEmitter.java
new file mode 100644
index 0000000..8c1f208
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchEmitter.java
@@ -0,0 +1,101 @@
+/*
+ * 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.connector.sink.SinkWriter;
+import org.apache.flink.connector.elasticsearch.sink.ElasticsearchEmitter;
+import org.apache.flink.connector.elasticsearch.sink.RequestIndexer;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+
+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.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Sink function for converting upserts into Elasticsearch {@link ActionRequest}s. */
+class RowElasticsearchEmitter implements ElasticsearchEmitter<RowData> {
+
+ private final IndexGenerator indexGenerator;
+ private final SerializationSchema<RowData> serializationSchema;
+ private final XContentType contentType;
+ private final Function<RowData, String> createKey;
+
+ public RowElasticsearchEmitter(
+ IndexGenerator indexGenerator,
+ SerializationSchema<RowData> serializationSchema,
+ XContentType contentType,
+ Function<RowData, String> createKey) {
+ this.indexGenerator = checkNotNull(indexGenerator);
+ this.serializationSchema = checkNotNull(serializationSchema);
+ this.contentType = checkNotNull(contentType);
+ this.createKey = checkNotNull(createKey);
+ }
+
+ @Override
+ public void open() {
+ indexGenerator.open();
+ }
+
+ @Override
+ public void emit(RowData element, SinkWriter.Context context, 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 =
+ new UpdateRequest(indexGenerator.generate(row), key)
+ .doc(document, contentType)
+ .upsert(document, contentType);
+ indexer.add(updateRequest);
+ } else {
+ final IndexRequest indexRequest =
+ new IndexRequest(indexGenerator.generate(row))
+ .id(key)
+ .source(document, contentType);
+ indexer.add(indexRequest);
+ }
+ }
+
+ private void processDelete(RowData row, RequestIndexer indexer) {
+ final String key = createKey.apply(row);
+ final DeleteRequest deleteRequest = new DeleteRequest(indexGenerator.generate(row), key);
+ indexer.add(deleteRequest);
+ }
+}
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
index 3ff21ed..4f40c1c 100644
--- 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
@@ -159,9 +159,7 @@ public class Elasticsearch7DynamicSinkFactoryTest extends TestLogger {
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].");
+ + "[ARRAY, MAP, MULTISET, ROW, RAW, VARBINARY].");
sinkFactory.createDynamicTableSink(
context()
.withSchema(
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
index fdf7a3a..03d2036 100644
--- 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
@@ -20,8 +20,8 @@ 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.api.connector.sink.Sink;
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;
@@ -29,7 +29,7 @@ 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.connector.sink.SinkProvider;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
@@ -111,35 +111,28 @@ public class Elasticsearch7DynamicSinkITCase extends TestLogger {
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();
+ DynamicTableSink.SinkRuntimeProvider runtimeProvider =
+ sinkFactory
+ .createDynamicTableSink(
+ context()
+ .withSchema(schema)
+ .withOption(
+ Elasticsearch7ConnectorOptions.INDEX_OPTION.key(),
+ index)
+ .withOption(
+ Elasticsearch7ConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ .build())
+ .getSinkRuntimeProvider(new MockContext());
+
+ final SinkProvider sinkProvider = (SinkProvider) runtimeProvider;
+ final Sink<RowData, ?, ?, ?> sink = sinkProvider.createSink();
StreamExecutionEnvironment environment =
StreamExecutionEnvironment.getExecutionEnvironment();
environment.setParallelism(4);
rowData.setRowKind(RowKind.UPDATE_AFTER);
- environment.<RowData>fromElements(rowData).addSink(sinkFunction);
+ environment.<RowData>fromElements(rowData).sinkTo(sink);
environment.execute();
Client client = getClient();
@@ -178,15 +171,11 @@ public class Elasticsearch7DynamicSinkITCase extends TestLogger {
+ 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())
+ Elasticsearch7ConnectorOptions.INDEX_OPTION.key(), index)
+ String.format(
"'%s'='%s'\n",
- ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
- "false")
+ Elasticsearch7ConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ ")");
tableEnvironment
@@ -236,15 +225,11 @@ public class Elasticsearch7DynamicSinkITCase extends TestLogger {
+ 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())
+ Elasticsearch7ConnectorOptions.INDEX_OPTION.key(), index)
+ String.format(
"'%s'='%s'\n",
- ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
- "false")
+ Elasticsearch7ConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ ")");
tableEnvironment
@@ -326,15 +311,11 @@ public class Elasticsearch7DynamicSinkITCase extends TestLogger {
+ 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())
+ Elasticsearch7ConnectorOptions.INDEX_OPTION.key(), index)
+ String.format(
"'%s'='%s'\n",
- ElasticsearchConnectorOptions.FLUSH_ON_CHECKPOINT_OPTION.key(),
- "false")
+ Elasticsearch7ConnectorOptions.HOSTS_OPTION.key(),
+ elasticsearchContainer.getHttpHostAddress())
+ ")");
tableEnvironment
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
deleted file mode 100644
index 6ab6828..0000000
--- a/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkTest.java
+++ /dev/null
@@ -1,294 +0,0 @@
-/*
- * 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.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,
- 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,
- 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,
- 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();
- }
- }
-}
[flink] 02/02: [hotfix][table-common] fix DataType methods in
DynamicTableFactory docstring
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 980acdb665c08a8fa09f02b24755d607a21fc1c8
Author: Alexander Preuß <11...@users.noreply.github.com>
AuthorDate: Thu Oct 7 11:01:39 2021 +0200
[hotfix][table-common] fix DataType methods in DynamicTableFactory docstring
---
.../java/org/apache/flink/table/factories/DynamicTableFactory.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DynamicTableFactory.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DynamicTableFactory.java
index 7fb880a..0490cd7 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DynamicTableFactory.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DynamicTableFactory.java
@@ -95,11 +95,11 @@ public interface DynamicTableFactory extends Factory {
*
* <pre>{@code
* // Project some fields into a new data type
- * DataType projectedDataType = DataType.projectRow(
+ * DataType projectedDataType = DataType.projectFields(
* context.getPhysicalRowDataType(), projectedIndexes);
*
* // Create key data type
- * DataType keyDataType = DataType.projectRow(
+ * DataType keyDataType = DataType.projectFields(
* context.getPhysicalRowDataType(), context.getPrimaryKeyIndexes());
*
* // Create a new data type filtering columns of the original data type