[Sort]Upgrade sort format to support InLongMsg body containing multip…

…le pieces of data (#9520) (#9522)

inlong-common/src/main/java/org/apache/inlong/common/msg/AttributeConstants.java

@@ -35,11 +35,6 @@ public interface AttributeConstants {
      */
     String STREAM_ID = "streamId";
 
-    /**
-     * iname is like a streamId but used in file protocol(m=xxx)
-     */
-    String INAME = "iname";
-
     /**
      * data time
      */

inlong-common/src/main/java/org/apache/inlong/common/msg/InLongMsg.java

@@ -98,10 +98,14 @@ public void write(byte[] array, int position, int len)
     private boolean compress;
     private boolean isNumGroupId = false;
     private boolean ischeck = true;
-
+    private boolean isSupportLF = false;
     private final Version version;
     private long timeoffset = 0;
 
+    public int getVersion() {
+        return version.intValue();
+    }
+
     public void setTimeoffset(long offset) {
         this.timeoffset = offset;
     }
@@ -306,6 +310,10 @@ private boolean getBinNumFlag(ByteBuffer data) {
         return (data.getShort(BIN_MSG_EXTFIELD_OFFSET) & 0x4) == 0;
     }
 
+    private boolean getBinisSupportLF(ByteBuffer data) {
+        return (data.getShort(BIN_MSG_EXTFIELD_OFFSET) & 0x20) == 0x20;
+    }
+
     private boolean checkBinData(ByteBuffer data) {
         int totalLen = data.getInt(BIN_MSG_TOTALLEN_OFFSET);
         int bodyLen = data.getInt(BIN_MSG_BODYLEN_OFFSET);
@@ -644,6 +652,7 @@ private InLongMsg(ByteBuffer buffer, Version magic) throws IOException {
             this.createtime = getBinCreatetime(parsedBinInput);
             this.msgcnt = getBinMsgCnt(parsedBinInput);
             this.isNumGroupId = getBinNumFlag(parsedBinInput);
+            this.isSupportLF = getBinisSupportLF(parsedBinInput);
         }
     }
 
@@ -800,7 +809,7 @@ private void parseBinMsg() throws IOException {
         }
 
         boolean hasOtherAttr = ((extField & 0x1) == 0x1);
-        commonAttrMap.put(AttributeConstants.MESSAGE_COUNT, "1");
+        commonAttrMap.put(AttributeConstants.MESSAGE_COUNT, String.valueOf(this.msgcnt));
         // with private attributes,
         // need to splice private attributes + public attributes
         if (!hasOtherAttr) {
@@ -1117,4 +1126,8 @@ public boolean isNumGroupId() {
         checkMode(false);
         return isNumGroupId;
     }
+
+    public boolean isSupportLF() {
+        return isSupportLF;
+    }
 }

inlong-common/src/main/java/org/apache/inlong/common/msg/InLongMsgAttrBuilder.java

@@ -158,7 +158,7 @@ public String buildAttr() throws Exception {
                 throw new Exception("t is null");
             }
 
-            attrBuffer.append("&iname=").append(id);
+            attrBuffer.append("&streamId=").append(id);
 
             Date d = transData(this.tt, t);
             String tstr = null;
@@ -251,7 +251,7 @@ public MsgAttrProtocolM100 setPartitionUnit(PartitionUnit p) {
         public String buildAttr() throws Exception {
             // #lizard forgives
             if (id != null) {
-                attrBuffer.append("&iname=").append(id);
+                attrBuffer.append("&streamId=").append(id);
             } else if (idp >= 0) {
                 attrBuffer.append("&idp=").append(idp);
             }

inlong-sort/sort-formats/format-base/src/main/java/org/apache/inlong/sort/formats/base/DefaultDeserializationSchema.java

@@ -0,0 +1,104 @@
+/*
+ * 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.
+ */
+
+package org.apache.inlong.sort.formats.base;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.util.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * The base implementation of {@link DeserializationSchema}.
+ */
+public abstract class DefaultDeserializationSchema<T> implements DeserializationSchema<T> {
+
+    private static final long serialVersionUID = 1L;
+
+    private static final Logger LOG = LoggerFactory.getLogger(DefaultDeserializationSchema.class);
+
+    /**
+     * If true, the deserialization error will be ignored.
+     */
+    protected final boolean ignoreErrors;
+
+    /**
+     * If true, a parsing error is occurred.
+     */
+    private boolean errorOccurred = false;
+
+    /**
+     * The metric counting the number of ignored errors.
+     */
+    private transient Counter numIgnoredErrors;
+
+    public DefaultDeserializationSchema(boolean ignoreErrors) {
+        this.ignoreErrors = ignoreErrors;
+    }
+
+    /**
+     * Deserialize the data and handle the failure.
+     *
+     * <p>Note: Returns null if the message could not be properly deserialized.
+     */
+    @Override
+    public T deserialize(byte[] bytes) throws IOException {
+        try {
+            T result = deserializeInternal(bytes);
+            // reset error state after deserialize success
+            errorOccurred = false;
+            return result;
+        } catch (Exception e) {
+            if (ignoreErrors) {
+                errorOccurred = true;
+                getNumIgnoredErrors().inc();
+                if (bytes == null) {
+                    LOG.warn("Could not properly deserialize the data null.");
+                } else {
+                    LOG.warn("Could not properly deserialize the data {}.",
+                            StringUtils.byteToHexString(bytes), e);
+                }
+                return null;
+            }
+            throw new IOException("Failed to deserialize data " +
+                    StringUtils.byteToHexString(bytes), e);
+        }
+    }
+
+    protected abstract T deserializeInternal(byte[] bytes) throws IOException;
+
+    public boolean skipCurrentRecord(T t) {
+        return ignoreErrors && errorOccurred;
+    }
+
+    public long numSkippedRecords() {
+        return getNumIgnoredErrors().getCount();
+    }
+
+    // This method is used to ensure initialization of the 'numIgnoredErrors' field. The field is non-serializable and
+    // marked as transient
+    private Counter getNumIgnoredErrors() {
+        if (numIgnoredErrors == null) {
+            numIgnoredErrors = new SimpleCounter();
+        }
+        return numIgnoredErrors;
+    }
+}

inlong-sort/sort-formats/format-base/src/main/java/org/apache/inlong/sort/formats/base/DefaultSerializationSchema.java

@@ -0,0 +1,97 @@
+/*
+ * 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.
+ */
+
+package org.apache.inlong.sort.formats.base;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * The base implementation of {@link SerializationSchema}.
+ */
+public abstract class DefaultSerializationSchema<T> implements SerializationSchema<T> {
+
+    private static final long serialVersionUID = 1L;
+
+    private static final Logger LOG = LoggerFactory.getLogger(DefaultSerializationSchema.class);
+
+    /**
+     * If true, the serialization error will be ignored.
+     */
+    private final boolean ignoreErrors;
+
+    /**
+     * If true, a parsing error is occurred.
+     */
+    private boolean errorOccurred = false;
+
+    /**
+     * The metric counting the number of ignored errors.
+     */
+    private transient Counter numIgnoredErrors;
+
+    public DefaultSerializationSchema(boolean ignoreErrors) {
+        this.ignoreErrors = ignoreErrors;
+    }
+
+    /**
+     * Serialize the data and handle the failure.
+     *
+     * <p>Note: Returns null if the message could not be properly serialized.
+     */
+    @Override
+    public byte[] serialize(T data) {
+        try {
+            byte[] result = serializeInternal(data);
+            // reset error state after serialize success
+            errorOccurred = false;
+            return result;
+        } catch (Exception e) {
+            if (ignoreErrors) {
+                errorOccurred = true;
+                getNumIgnoredErrors().inc();
+                LOG.warn("Could not properly serialize the data {}", data, e);
+                return null;
+            }
+            throw new RuntimeException("Failed to serialize data " + data, e);
+        }
+    }
+
+    protected abstract byte[] serializeInternal(T data) throws IOException;
+
+    public boolean skipCurrentRecord(T t) {
+        return ignoreErrors && errorOccurred;
+    }
+
+    public long numSkippedRecords() {
+        return getNumIgnoredErrors().getCount();
+    }
+
+    // This method is used to ensure initialization of the numIgnoredErrors field. The field is non-serializable and
+    // marked as transient
+    private Counter getNumIgnoredErrors() {
+        if (numIgnoredErrors == null) {
+            numIgnoredErrors = new SimpleCounter();
+        }
+        return numIgnoredErrors;
+    }
+}

inlong-sort/sort-formats/format-base/src/main/java/org/apache/inlong/sort/formats/base/DefaultTableFormatDeserializer.java

@@ -30,7 +30,7 @@
 /**
  * The default implementation of {@link TableFormatDeserializer}.
  */
-public class DefaultTableFormatDeserializer implements TableFormatDeserializer {
+public class DefaultTableFormatDeserializer extends TableFormatDeserializer {
 
     private static final long serialVersionUID = 1L;
 

inlong-sort/sort-formats/format-base/src/main/java/org/apache/inlong/sort/formats/base/DefaultTableFormatSerializer.java

@@ -30,25 +30,23 @@
 /**
  * The default implementation of {@link TableFormatSerializer}.
  */
-public class DefaultTableFormatSerializer implements TableFormatSerializer {
+public class DefaultTableFormatSerializer extends TableFormatSerializer {
 
     private static final long serialVersionUID = 1L;
 
     private static final Logger LOG = LoggerFactory.getLogger(DefaultTableFormatSerializer.class);
-
     /**
      * The delegated serialization schema for rows.
      */
     private final SerializationSchema<Row> serializationSchema;
 
     /**
-     * True if ignore errors in the serialization.
+     * True if ignore errors in the deserialization.
      */
     private final boolean ignoreErrors;
 
     public DefaultTableFormatSerializer(
-            SerializationSchema<Row> serializationSchema,
-            boolean ignoreErrors) {
+            SerializationSchema<Row> serializationSchema, boolean ignoreErrors) {
         this.serializationSchema = serializationSchema;
         this.ignoreErrors = ignoreErrors;
     }
@@ -75,7 +73,11 @@ public void flatMap(
             }
         }
 
-        collector.collect(bytes);
+        if (bytes != null) {
+            collector.collect(bytes);
+        } else {
+            LOG.warn("Data is Empty.");
+        }
     }
 
     @Override
@@ -89,20 +91,17 @@ public boolean equals(Object o) {
         }
 
         DefaultTableFormatSerializer that = (DefaultTableFormatSerializer) o;
-        return ignoreErrors == that.ignoreErrors
-                && Objects.equals(serializationSchema, that.serializationSchema);
+        return Objects.equals(serializationSchema, that.serializationSchema);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(serializationSchema, ignoreErrors);
+        return Objects.hash(serializationSchema);
     }
 
     @Override
     public String toString() {
-        return "DefaultTableFormatSerializer{"
-                + "serializationSchema=" + serializationSchema
-                + ", ignoreErrors=" + ignoreErrors
-                + '}';
+        return "DefaultTableFormatSerializer{" +
+                "serializationSchema=" + serializationSchema + '}';
     }
 }