@Override
    public void flatMap(
            byte[] bytes,
            Collector<Row> collector) throws Exception {
        InLongMsg inLongMsg = InLongMsg.parseFrom(bytes);

        for (String attr : inLongMsg.getAttrs()) {
            Iterator<byte[]> iterator = inLongMsg.getIterator(attr);
            if (iterator == null) {
                continue;
            }

            InLongMsgHead head;
            try {
                head = parseHead(attr);
            } catch (Exception e) {
                failureHandler.onParsingHeadFailure(attr, e);
                continue;
            }

            while (iterator.hasNext()) {

                byte[] bodyBytes = iterator.next();
                if (bodyBytes == null || bodyBytes.length == 0) {
                    continue;
                }

                List<InLongMsgBody> bodyList;
                try {
                    bodyList = parseBodyList(bodyBytes);
                } catch (Exception e) {
                    failureHandler.onParsingBodyFailure(head, bodyBytes, e);
                    continue;
                }

                for (InLongMsgBody body : bodyList) {
                    List<Row> rows;
                    try {
                        rows = convertRows(head, body);
                    } catch (Exception e) {
                        failureHandler.onConvertingRowFailure(head, body, e);
                        continue;
                    }

                    if (rows != null) {
                        for (Row row : rows) {
                            collector.collect(row);
                        }
                    }
                }
            }
        }
    }
Converts the InLongMsg record into row list.
inlong-sort/sort-formats/format-row/format-inlongmsg-base/src/main/java/org/apache/inlong/sort/formats/inlongmsg/row/AbstractInLongMsgFormatDeserializer.java