| commit | bfaa761bf7889c27763d1e3b4939f8d9766913d7 | [log] [tgz] |
|---|---|---|
| author | Dawid Wysakowicz <dwysakowicz@apache.org> | Tue Jun 02 17:09:26 2020 +0200 |
| committer | MartijnVisser <martijn@2symbols.com> | Mon Apr 04 15:45:43 2022 +0200 |
| tree | 6d41c7e627da53dc18cce132c45bfaa0d7cd4a79 | |
| parent | 024382389d6bd3907b296a14b9032b80c13c4ade [diff] |
[FLINK-18006] Always overwrite RestClientFactory in ElasticsearchXDynamicSink We always overwrite the RestClientFactory in order to workaround an issue with shading classes in lambdas deserialization method. That way we never use the default lambda from ElasticsearchSink$Builder which cannot be deserialized when used from a flink-sql-connector-elasticsearch module due to shading. This closes #12455
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 index bedfbef..680cb2c 100644 --- 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
@@ -136,8 +136,9 @@ config.getBulkFlushBackoffRetries().ifPresent(builder::setBulkFlushBackoffRetries); config.getBulkFlushBackoffDelay().ifPresent(builder::setBulkFlushBackoffDelay); - config.getPathPrefix() - .ifPresent(pathPrefix -> builder.setRestClientFactory(new DefaultRestClientFactory(pathPrefix))); + // we must overwrite the default factory which is defined with a lambda because of a bug + // in shading lambda serialization shading see FLINK-18006 + builder.setRestClientFactory(new DefaultRestClientFactory(config.getPathPrefix().orElse(null))); final ElasticsearchSink<RowData> sink = builder.build();
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 408673e..7aa52ea 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
@@ -136,8 +136,9 @@ config.getBulkFlushBackoffRetries().ifPresent(builder::setBulkFlushBackoffRetries); config.getBulkFlushBackoffDelay().ifPresent(builder::setBulkFlushBackoffDelay); - config.getPathPrefix() - .ifPresent(pathPrefix -> builder.setRestClientFactory(new DefaultRestClientFactory(pathPrefix))); + // we must overwrite the default factory which is defined with a lambda because of a bug + // in shading lambda serialization shading see FLINK-18006 + builder.setRestClientFactory(new DefaultRestClientFactory(config.getPathPrefix().orElse(null))); final ElasticsearchSink<RowData> sink = builder.build();