(flink-connector-elasticsearch) branch main updated: [FLINK-38054] Fix OperationSerializer issue for es8 sink

[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 ad7464e  [FLINK-38054] Fix OperationSerializer issue for es8 sink
ad7464e is described below

commit ad7464ed8fa1a16b071d344a0a484980eee29d10
Author: reswqa <res...@163.com>
AuthorDate: Tue Jul 7 11:28:19 2026 +0800

    [FLINK-38054] Fix OperationSerializer issue for es8 sink
    
    Co-authored-by: Andrey Starostin <andrey.staros...@choco.com>
---
 .../elasticsearch/sink/OperationSerializer.java    |  8 +++++-
 .../sink/OperationSerializerTest.java              | 33 ++++++++++++++++++++++
 2 files changed, 40 insertions(+), 1 deletion(-)

diff --git 
a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java
 
b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java
index 1c2cb81..d9cb555 100644
--- 
a/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java
+++ 
b/flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java
@@ -24,6 +24,7 @@ package org.apache.flink.connector.elasticsearch.sink;
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
 import org.objenesis.strategy.StdInstantiatorStrategy;
 
 import java.io.ByteArrayOutputStream;
@@ -36,7 +37,12 @@ public class OperationSerializer {
 
     public OperationSerializer() {
         kryo.setRegistrationRequired(false);
-        kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+        DefaultInstantiatorStrategy instantiatorStrategy = new 
DefaultInstantiatorStrategy();
+        instantiatorStrategy.setFallbackInstantiatorStrategy(new 
StdInstantiatorStrategy());
+        kryo.setInstantiatorStrategy(instantiatorStrategy);
+        // Use TCCL so Kryo can resolve classes that live in the user-code
+        // ClassLoader, not the system AppClassLoader.
+        kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
     }
 
     public void serialize(Operation request, DataOutputStream out) {
diff --git 
a/flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializerTest.java
 
b/flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializerTest.java
index 5c45d7c..a62cc1e 100644
--- 
a/flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializerTest.java
+++ 
b/flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializerTest.java
@@ -25,7 +25,11 @@ import 
co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant;
 import co.elastic.clients.elasticsearch.core.bulk.DeleteOperation;
 import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
 import co.elastic.clients.elasticsearch.core.bulk.UpdateOperation;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import 
org.assertj.core.api.recursive.comparison.RecursiveComparisonConfiguration;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -82,4 +86,33 @@ public class OperationSerializerTest {
     private int getRequestSize(Operation requestEntry) {
         return new OperationSerializer().size(requestEntry);
     }
+
+    @Test
+    public void testSerializeUpdateOperationWithJacksonNodes() throws 
Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        ObjectNode doc = mapper.createObjectNode();
+        doc.put("field", "value");
+        doc.putObject("nested").put("foo", "bar");
+        ArrayNode list = mapper.createArrayNode();
+        list.add("first");
+        list.add(mapper.createObjectNode().put("inner", "second"));
+        doc.set("list", list);
+
+        UpdateOperation<Object, Object> updateOp =
+                UpdateOperation.of(b -> b.index("idx").id("id1").action(a -> 
a.doc(doc)));
+        Operation expectedState = new Operation(updateOp);
+
+        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+        DataOutputStream out = new DataOutputStream(bytes);
+        new OperationSerializer().serialize(expectedState, out);
+
+        ByteArrayInputStream inputBytes = new 
ByteArrayInputStream(bytes.toByteArray());
+        DataInputStream in = new DataInputStream(inputBytes);
+        Operation actualState =
+                new 
OperationSerializer().deserialize(getRequestSize(expectedState), in);
+
+        assertThat(actualState.getBulkOperationVariant())
+                
.usingRecursiveComparison(RecursiveComparisonConfiguration.builder().build())
+                .isEqualTo(expectedState.getBulkOperationVariant());
+    }
 }