(flink-connector-elasticsearch) branch main updated: [FLINK-38429] Improve perf for Elasticsearch 8 connector

[guoweijie]

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 <belie...@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