public static Message buildMessage(String topicName, Event event) {
        Map<String, String> headers = event.getHeaders();
        Message message = new Message(topicName, event.getBody());
        String pkgVersion = headers.get(ConfigConstants.MSG_ENCODE_VER);
        long dataTimeL = Long.parseLong(headers.get(ConfigConstants.PKG_TIME_KEY));
        message.putSystemHeader(headers.get(Constants.INLONG_STREAM_ID),
                DateTimeUtils.ms2yyyyMMddHHmm(dataTimeL));
        Map<String, String> extraAttrMap = MessageUtils.getXfsAttrs(headers, pkgVersion);
        for (Map.Entry<String, String> entry : extraAttrMap.entrySet()) {
            message.setAttrKeyVal(entry.getKey(), entry.getValue());
        }
        return message;
    }
Build TubeMQ's message

@param topicName      the topic name of message
@param event          the DataProxy event
@return   the message object
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/TubeUtils.java