You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2024/05/22 04:15:09 UTC
(flink-connector-elasticsearch) branch main updated: [FLINK-35287] Builder builds NetworkConfig for Elasticsearch connector 8
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-elasticsearch.git
The following commit(s) were added to refs/heads/main by this push:
new 0b6d9f8 [FLINK-35287] Builder builds NetworkConfig for Elasticsearch connector 8
0b6d9f8 is described below
commit 0b6d9f823d9831816a81d02c0c1f367746363688
Author: Mingliang Liu <mi...@netflix.com>
AuthorDate: Fri May 3 00:20:13 2024 -0700
[FLINK-35287] Builder builds NetworkConfig for Elasticsearch connector 8
---
.../sink/Elasticsearch8AsyncSink.java | 40 ++++------------------
.../sink/Elasticsearch8AsyncSinkBuilder.java | 15 ++++----
.../sink/Elasticsearch8AsyncWriter.java | 12 ++-----
.../elasticsearch/sink/NetworkConfig.java | 7 ++--
.../sink/Elasticsearch8AsyncWriterITCase.java | 12 ++-----
5 files changed, 22 insertions(+), 64 deletions(-)
diff --git a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncSink.java b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncSink.java
index 3d98459..f4d87cf 100644
--- a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncSink.java
+++ b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncSink.java
@@ -21,21 +21,17 @@
package org.apache.flink.connector.elasticsearch.sink;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.connector.base.sink.AsyncSinkBase;
import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.http.Header;
-import org.apache.http.HttpHost;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.Collections;
-import java.util.List;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* Elasticsearch8AsyncSink Apache Flink's Async Sink that submits Operations into an Elasticsearch
@@ -47,15 +43,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
public class Elasticsearch8AsyncSink<InputT> extends AsyncSinkBase<InputT, Operation> {
private static final Logger LOG = LoggerFactory.getLogger(Elasticsearch8AsyncSink.class);
- private final String username;
-
- private final String password;
-
- private final String certificateFingerprint;
-
- private final List<HttpHost> httpHosts;
-
- private final List<Header> headers;
+ @VisibleForTesting protected final NetworkConfig networkConfig;
protected Elasticsearch8AsyncSink(
ElementConverter<InputT, Operation> converter,
@@ -65,11 +53,7 @@ public class Elasticsearch8AsyncSink<InputT> extends AsyncSinkBase<InputT, Opera
long maxBatchSizeInBytes,
long maxTimeInBufferMS,
long maxRecordSizeInByte,
- String username,
- String password,
- String certificateFingerprint,
- List<HttpHost> httpHosts,
- List<Header> headers) {
+ NetworkConfig networkConfig) {
super(
converter,
maxBatchSize,
@@ -79,11 +63,7 @@ public class Elasticsearch8AsyncSink<InputT> extends AsyncSinkBase<InputT, Opera
maxTimeInBufferMS,
maxRecordSizeInByte);
- this.username = username;
- this.password = password;
- this.certificateFingerprint = certificateFingerprint;
- this.httpHosts = checkNotNull(httpHosts, "Hosts must not be null");
- this.headers = headers;
+ this.networkConfig = networkConfig;
}
@Override
@@ -98,11 +78,7 @@ public class Elasticsearch8AsyncSink<InputT> extends AsyncSinkBase<InputT, Opera
getMaxBatchSizeInBytes(),
getMaxTimeInBufferMS(),
getMaxRecordSizeInBytes(),
- username,
- password,
- certificateFingerprint,
- httpHosts,
- headers,
+ networkConfig,
Collections.emptyList());
}
@@ -118,11 +94,7 @@ public class Elasticsearch8AsyncSink<InputT> extends AsyncSinkBase<InputT, Opera
getMaxBatchSizeInBytes(),
getMaxTimeInBufferMS(),
getMaxRecordSizeInBytes(),
- username,
- password,
- certificateFingerprint,
- httpHosts,
- headers,
+ networkConfig,
recoveredState);
}
diff --git a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncSinkBuilder.java b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncSinkBuilder.java
index 391b5fe..b3bd969 100644
--- a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncSinkBuilder.java
+++ b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncSinkBuilder.java
@@ -94,7 +94,7 @@ public class Elasticsearch8AsyncSinkBuilder<InputT>
*/
public Elasticsearch8AsyncSinkBuilder<InputT> setHeaders(Header... headers) {
checkNotNull(hosts);
- checkArgument(headers.length > 0, "Hosts cannot be empty");
+ checkArgument(headers.length > 0, "Headers cannot be empty");
this.headers = Arrays.asList(headers);
return this;
}
@@ -108,7 +108,7 @@ public class Elasticsearch8AsyncSinkBuilder<InputT>
*/
public Elasticsearch8AsyncSinkBuilder<InputT> setCertificateFingerprint(
String certificateFingerprint) {
- checkNotNull(username, "certificateFingerprint must not be null");
+ checkNotNull(certificateFingerprint, "certificateFingerprint must not be null");
this.certificateFingerprint = certificateFingerprint;
return this;
}
@@ -171,11 +171,7 @@ public class Elasticsearch8AsyncSinkBuilder<InputT>
Optional.ofNullable(getMaxBatchSizeInBytes()).orElse(DEFAULT_MAX_BATCH_SIZE_IN_B),
Optional.ofNullable(getMaxTimeInBufferMS()).orElse(DEFAULT_MAX_TIME_IN_BUFFER_MS),
Optional.ofNullable(getMaxRecordSizeInBytes()).orElse(DEFAULT_MAX_RECORD_SIZE_IN_B),
- username,
- password,
- certificateFingerprint,
- hosts,
- headers);
+ buildNetworkConfig());
}
private OperationConverter<InputT> buildOperationConverter(
@@ -183,6 +179,11 @@ public class Elasticsearch8AsyncSinkBuilder<InputT>
return converter != null ? new OperationConverter<>(converter) : null;
}
+ private NetworkConfig buildNetworkConfig() {
+ checkArgument(!hosts.isEmpty(), "Hosts cannot be empty.");
+ return new NetworkConfig(hosts, username, password, headers, certificateFingerprint);
+ }
+
/** A wrapper that evolves the Operation, since a BulkOperationVariant is not Serializable. */
public static class OperationConverter<T> implements ElementConverter<T, Operation> {
private final ElementConverter<T, BulkOperationVariant> converter;
diff --git a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriter.java b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriter.java
index e0f5b42..a3eb8df 100644
--- a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriter.java
+++ b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriter.java
@@ -35,8 +35,6 @@ import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
-import org.apache.http.Header;
-import org.apache.http.HttpHost;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -85,11 +83,7 @@ public class Elasticsearch8AsyncWriter<InputT> extends AsyncSinkWriter<InputT, O
long maxBatchSizeInBytes,
long maxTimeInBufferMS,
long maxRecordSizeInBytes,
- String username,
- String password,
- String certificateFingerprint,
- List<HttpHost> httpHosts,
- List<Header> headers,
+ NetworkConfig networkConfig,
Collection<BufferedRequestState<Operation>> state) {
super(
elementConverter,
@@ -104,9 +98,7 @@ public class Elasticsearch8AsyncWriter<InputT> extends AsyncSinkWriter<InputT, O
.build(),
state);
- this.esClient =
- new NetworkConfig(httpHosts, username, password, headers, certificateFingerprint)
- .create();
+ this.esClient = networkConfig.createEsClient();
final SinkWriterMetricGroup metricGroup = context.metricGroup();
checkNotNull(metricGroup);
diff --git a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/NetworkConfig.java b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/NetworkConfig.java
index ce96a83..fc41fd5 100644
--- a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/NetworkConfig.java
+++ b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/NetworkConfig.java
@@ -34,12 +34,13 @@ import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
+import java.io.Serializable;
import java.util.List;
import static org.apache.flink.util.Preconditions.checkState;
-/** NetworkConfigConfig A factory that creates valid ElasticsearchClient instances. */
-public class NetworkConfig {
+/** A factory that creates valid ElasticsearchClient instances. */
+public class NetworkConfig implements Serializable {
private final List<HttpHost> hosts;
private final List<Header> headers;
@@ -64,7 +65,7 @@ public class NetworkConfig {
this.certificateFingerprint = certificateFingerprint;
}
- public ElasticsearchAsyncClient create() {
+ public ElasticsearchAsyncClient createEsClient() {
return new ElasticsearchAsyncClient(
new RestClientTransport(this.getRestClient(), new JacksonJsonpMapper()));
}
diff --git a/flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java b/flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java
index 891f30b..f30872e 100644
--- a/flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java
+++ b/flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java
@@ -245,11 +245,7 @@ public class Elasticsearch8AsyncWriterITCase extends ElasticsearchSinkBaseITCase
5 * 1024 * 1024,
5000,
1024 * 1024,
- null,
- null,
- null,
- esHost,
- null) {
+ new NetworkConfig(esHost, null, null, null, null)) {
@Override
public StatefulSinkWriter createWriter(InitContext context) {
return new Elasticsearch8AsyncWriter<DummyData>(
@@ -261,11 +257,7 @@ public class Elasticsearch8AsyncWriterITCase extends ElasticsearchSinkBaseITCase
getMaxBatchSizeInBytes(),
getMaxTimeInBufferMS(),
getMaxRecordSizeInBytes(),
- null,
- null,
- null,
- esHost,
- null,
+ networkConfig,
Collections.emptyList()) {
@Override
protected void submitRequestEntries(