From 1dd403fb75393d38f6218cda509c88007207c178 Mon Sep 17 00:00:00 2001
From: beliefer <beliefer@163.com>
Date: Thu, 25 Sep 2025 11:02:57 +0800
Subject: [PATCH] [FLINK-38429] Improve perf for Elasticsearch 8 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 0c20fd1b..e086b003 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
@@ -69,6 +69,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(
@@ -112,6 +114,7 @@ public Elasticsearch8AsyncWriter(
         this.numRecordsSendPartialFailureCounter =
                 metricGroup.counter("numRecordsSendPartialFailure");
         this.numRequestSubmittedCounter = metricGroup.counter("numRequestSubmitted");
+        this.operationSerializer = new OperationSerializer();
     }
 
     @Override
@@ -192,7 +195,7 @@ private boolean isRetryable(Throwable error) {
 
     @Override
     protected long getSizeInBytes(Operation requestEntry) {
-        return new OperationSerializer().size(requestEntry);
+        return operationSerializer.size(requestEntry);
     }
 
     @Override