private List<InLongMessage> decodeInlongMsg(
            ClientContext context,
            InLongTopic inLongTopic,
            byte[] msgBytes,
            Map<String, String> headers) {
        List<InLongMessage> messageList = new ArrayList<>();

        InLongMsg inLongMsg = InLongMsg.parseFrom(msgBytes);
        for (String attr : inLongMsg.getAttrs()) {
            Map<String, String> attributes = StringUtil.splitKv(attr, INLONGMSG_ATTR_ENTRY_DELIMITER,
                    INLONGMSG_ATTR_KV_DELIMITER, null, null);

            String groupId = Optional.ofNullable(attributes.get(INLONGMSG_ATTR_GROUP_ID))
                    .orElseThrow(() -> new IllegalArgumentException(String.format(PARSE_ATTR_ERROR_STRING,
                            INLONGMSG_ATTR_GROUP_ID)));

            String streamId = Optional.ofNullable(attributes.get(INLONGMSG_ATTR_STREAM_ID))
                    .orElseThrow(() -> new IllegalArgumentException(String.format(PARSE_ATTR_ERROR_STRING,
                            INLONGMSG_ATTR_STREAM_ID)));

            // Extracts time from the attributes
            long msgTime;
            if (attributes.containsKey(INLONGMSG_ATTR_TIME_T)) {
                String date = attributes.get(INLONGMSG_ATTR_TIME_T).trim();
                msgTime = StringUtil.parseDateTime(date);
            } else if (attributes.containsKey(INLONGMSG_ATTR_TIME_DT)) {
                String epoch = attributes.get(INLONGMSG_ATTR_TIME_DT).trim();
                msgTime = Long.parseLong(epoch);
            } else {
                throw new IllegalArgumentException(String.format(PARSE_ATTR_ERROR_STRING,
                        INLONGMSG_ATTR_TIME_T + " or " + INLONGMSG_ATTR_TIME_DT));
            }

            String srcIp = Optional.ofNullable(attributes.get(INLONGMSG_ATTR_CLIENT_IP))
                    .orElse(DEFAULT_IP);

            Iterator<byte[]> iterator = inLongMsg.getIterator(attr);
            while (iterator.hasNext()) {
                byte[] bodyBytes = iterator.next();
                if (Objects.isNull(bodyBytes)) {
                    continue;
                }
                InLongMessage inLongMessage = new InLongMessage(groupId, streamId, msgTime,
                        srcIp, bodyBytes, attributes);
                messageList.add(inLongMessage);
            }
        }
        return messageList;
    }
transform MessageObjs to SortSdkMessage

@param messageObjs {@link MessageObjs}
@return {@link List}
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java