public static Message buildMessage(String topicName, String filterItem,
            byte[] bodyData, long serialId) {
        // build message to be sent
        Message message = new Message(topicName, bodyData);
        long currTimeMillis = System.currentTimeMillis();
        // added a serial number and data generation time to each message
        message.setAttrKeyVal("serialId", String.valueOf(serialId));
        message.setAttrKeyVal("dataTime", String.valueOf(currTimeMillis));
        // add filter attribute information, time require yyyyMMddHHmm format
        message.putSystemHeader(filterItem,
                DateTimeConvertUtils.ms2yyyyMMddHHmm(currTimeMillis));
        return message;
    }
parse topic Parameter with format topic_1[,topic_2[:filterCond_2.1[;filterCond_2.2]]]
 topicParam->set(filterCond) map
@param topicParam - composite string
@return - map of topic->set(filterCond)
inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/utils/MixedUtils.java