This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12176-10ed8d433f07b01591d1c8cc699a8d7b7fe0e2a9 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 438f9ccfbef643b8c39ccb3f4a6261345c478ab7 Author: zhiweiniu <[email protected]> AuthorDate: Tue Sep 8 12:26:14 2026 +0000 [Improve][Zeta] Reduce lock contention in ProtoStuff schema lookup (#12176) --- .../serializer/protobuf/ProtoStuffSerializer.java | 13 ++++- .../protobuf/ProtoStuffSerializerTest.java | 56 ++++++++++++++++++++++ 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/seatunnel-engine/seatunnel-engine-serializer/serializer-protobuf/src/main/java/org/apache/seatunnel/engine/serializer/protobuf/ProtoStuffSerializer.java b/seatunnel-engine/seatunnel-engine-serializer/serializer-protobuf/src/main/java/org/apache/seatunnel/engine/serializer/protobuf/ProtoStuffSerializer.java index 7a408766a4..e387158161 100644 --- a/seatunnel-engine/seatunnel-engine-serializer/serializer-protobuf/src/main/java/org/apache/seatunnel/engine/serializer/protobuf/ProtoStuffSerializer.java +++ b/seatunnel-engine/seatunnel-engine-serializer/serializer-protobuf/src/main/java/org/apache/seatunnel/engine/serializer/protobuf/ProtoStuffSerializer.java @@ -46,10 +46,19 @@ public class ProtoStuffSerializer implements Serializer { /** At the moment it looks like we only have one Schema. */ private static final Map<Class<?>, Schema<?>> SCHEMA_CACHE = new ConcurrentHashMap<>(); + static { + // Configure null preservation once, before creating the first runtime schema. + System.setProperty("protostuff.runtime.preserve_null_elements", "true"); + } + @SuppressWarnings("unchecked") private static <T> Schema<T> getSchema(Class<T> clazz) { - System.setProperty("protostuff.runtime.preserve_null_elements", "true"); - return (Schema<T>) SCHEMA_CACHE.computeIfAbsent(clazz, RuntimeSchema::createFrom); + // Return cached schemas directly, using computeIfAbsent only on cache misses. + Schema<?> schema = SCHEMA_CACHE.get(clazz); + if (schema == null) { + return (Schema<T>) SCHEMA_CACHE.computeIfAbsent(clazz, RuntimeSchema::createFrom); + } + return (Schema<T>) schema; } private static final Set<Class<?>> WRAPPERS = new HashSet<>(); diff --git a/seatunnel-engine/seatunnel-engine-serializer/serializer-protobuf/src/test/java/org/apache/seatunnel/engine/serializer/protobuf/ProtoStuffSerializerTest.java b/seatunnel-engine/seatunnel-engine-serializer/serializer-protobuf/src/test/java/org/apache/seatunnel/engine/serializer/protobuf/ProtoStuffSerializerTest.java index 71cc62ae45..8200f500be 100644 --- a/seatunnel-engine/seatunnel-engine-serializer/serializer-protobuf/src/test/java/org/apache/seatunnel/engine/serializer/protobuf/ProtoStuffSerializerTest.java +++ b/seatunnel-engine/seatunnel-engine-serializer/serializer-protobuf/src/test/java/org/apache/seatunnel/engine/serializer/protobuf/ProtoStuffSerializerTest.java @@ -23,8 +23,64 @@ package org.apache.seatunnel.engine.serializer.protobuf; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + public class ProtoStuffSerializerTest { + @Test + public void testConcurrentSchemaInitializationAndReuse() throws Exception { + int threads = 8; + ExecutorService executor = Executors.newFixedThreadPool(threads); + CountDownLatch ready = new CountDownLatch(threads); + CountDownLatch start = new CountDownLatch(1); + ProtoStuffSerializer serializer = new ProtoStuffSerializer(); + List<Future<?>> results = new ArrayList<>(); + try { + for (int thread = 0; thread < threads; thread++) { + final int id = thread; + results.add( + executor.submit( + () -> { + ready.countDown(); + Assertions.assertTrue(start.await(30, TimeUnit.SECONDS)); + for (int iteration = 0; iteration < 100; iteration++) { + ConcurrentPayload input = new ConcurrentPayload(); + input.id = id; + input.values = Arrays.asList("before", null, "after"); + ConcurrentPayload output = + serializer.deserialize( + serializer.serialize(input), + ConcurrentPayload.class); + Assertions.assertEquals(input.id, output.id); + Assertions.assertEquals(input.values, output.values); + } + return null; + })); + } + Assertions.assertTrue(ready.await(30, TimeUnit.SECONDS)); + start.countDown(); + for (Future<?> result : results) { + result.get(30, TimeUnit.SECONDS); + } + } finally { + start.countDown(); + executor.shutdownNow(); + Assertions.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + + public static class ConcurrentPayload { + public int id; + public List<String> values; + } + @Test public void testProtoStuffSerializerForArrayType() { Long[] longs = new Long[10];
