[Improve][Zeta] Remove serialize(deserialize) cost when use shuffle action

seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/checkpoint/CheckpointType.java

@@ -31,6 +31,15 @@ public enum CheckpointType {
     private final boolean auto;
     private final String name;
 
+    public static CheckpointType fromName(String name) {
+        for (CheckpointType type : CheckpointType.values()) {
+            if (type.name.equals(name)) {
+                return type;
+            }
+        }
+        throw new IllegalArgumentException("Unknown checkpoint type: " + name);
+    }
+
     CheckpointType(boolean auto, String name) {
         this.auto = auto;
         this.name = name;

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/RecordSerializer.java

@@ -67,7 +67,7 @@ public Record read(ObjectDataInput in) throws IOException {
         if (dataType == RecordDataType.CHECKPOINT_BARRIER.ordinal()) {
             data =
                     new CheckpointBarrier(
-                            in.readLong(), in.readLong(), CheckpointType.valueOf(in.readString()));
+                            in.readLong(), in.readLong(), CheckpointType.fromName(in.readString()));
         } else if (dataType == RecordDataType.SEATUNNEL_ROW.ordinal()) {
             String tableId = in.readString();
             byte rowKind = in.readByte();

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/RecordSerializerHook.java

@@ -19,11 +19,9 @@
 
 import org.apache.seatunnel.api.table.type.Record;
 
-import com.google.auto.service.AutoService;
 import com.hazelcast.nio.serialization.Serializer;
 import com.hazelcast.nio.serialization.SerializerHook;
 
-@AutoService(SerializerHook.class)
 public class RecordSerializerHook implements SerializerHook<Record> {
 
     @Override

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/ShuffleSinkFlowLifeCycle.java

@@ -168,7 +168,8 @@ private synchronized void shuffleFlush() {
         for (Map.Entry<String, Queue<Record<?>>> shuffleBatch : shuffleBuffer.entrySet()) {
             IQueue<Record<?>> shuffleQueue = shuffles.get(shuffleBatch.getKey());
             Queue<Record<?>> shuffleQueueBatch = shuffleBatch.getValue();
-            if (!shuffleQueue.addAll(shuffleBatch.getValue())) {
+            if (shuffleQueue.remainingCapacity() <= 0
+                    || !shuffleQueue.addAll(shuffleBatch.getValue())) {
                 for (; ; ) {
                     Record<?> shuffleItem = shuffleQueueBatch.poll();
                     if (shuffleItem == null) {

seatunnel-engine/seatunnel-engine-server/src/main/resources/META-INF/services/com.hazelcast.SerializerHook

@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.seatunnel.engine.server.serializable.RecordSerializerHook