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 2026/07/09 02:37:28 UTC
(flink-connector-elasticsearch) branch main updated: [FLINK-38429] Improve perf for Elasticsearch 8 connector
This is an automated email from the ASF dual-hosted git repository.
reswqa 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 092dfaf [FLINK-38429] Improve perf for Elasticsearch 8 connector
092dfaf is described below
commit 092dfafce014c0cc877c45e7b2fb6a39f309ca10
Author: beliefer <be...@163.com>
AuthorDate: Thu Sep 25 11:02:57 2025 +0800
[FLINK-38429] Improve perf for Elasticsearch 8 connector
---
.../connector/elasticsearch/sink/Elasticsearch8AsyncWriter.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
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 174d4c2..1d8ceaf 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
@@ -71,6 +71,8 @@ public class Elasticsearch8AsyncWriter<InputT> extends AsyncSinkWriter<InputT, O
/** A counter to track the number of bulk requests that are sent to Elasticsearch. */
private final Counter numRequestSubmittedCounter;
+ private final OperationSerializer operationSerializer;
+
private static final FatalExceptionClassifier ELASTICSEARCH_FATAL_EXCEPTION_CLASSIFIER =
FatalExceptionClassifier.createChain(
new FatalExceptionClassifier(
@@ -114,6 +116,7 @@ public class Elasticsearch8AsyncWriter<InputT> extends AsyncSinkWriter<InputT, O
this.numRecordsSendPartialFailureCounter =
metricGroup.counter("numRecordsSendPartialFailure");
this.numRequestSubmittedCounter = metricGroup.counter("numRequestSubmitted");
+ this.operationSerializer = new OperationSerializer();
}
@Override
@@ -194,7 +197,7 @@ public class Elasticsearch8AsyncWriter<InputT> extends AsyncSinkWriter<InputT, O
@Override
protected long getSizeInBytes(Operation requestEntry) {
- return new OperationSerializer().size(requestEntry);
+ return operationSerializer.size(requestEntry);
}
@Override