public static void addConsumeRecord(Map<String, TrafficInfo> trafficInfos) {
        if (!auditConfig.isAuditEnable() || trafficInfos == null || trafficInfos.isEmpty()) {
            return;
        }
        for (Map.Entry<String, TrafficInfo> entry : trafficInfos.entrySet()) {
            if (entry == null || entry.getKey() == null || entry.getValue() == null) {
                continue;
            }
            String statKey = entry.getKey();
            String[] statKeyItems = statKey.split(TokenConstants.SEGMENT_SEP, -1);
            if (statKeyItems.length < 8) {
                continue;
            }
            if (TStringUtils.isEmpty(statKeyItems[0])) {
                continue;
            }
            // like: test_1#127.0.0.1#test_consume_127.0.0.1-32677-1656672066382-1-Pull-3.9.2
            // #127.0.0.1#32677#test_consume#2#202207041219
            // topicName, brokerIP, clientId,
            // clientIP, client processId, consume group, partitionId, msgTime
            AuditOperator.getInstance().add(auditConfig.getAuditIdConsume(),
                    statKeyItems[0], statKeyItems[5], DateTimeConvertUtils.yyyyMMddHHmm2ms(statKeyItems[7]),
                    entry.getValue().getMsgCount(), entry.getValue().getMsgSize());
        }
    }
add consume record

@param trafficInfos the consumed traffic information
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/audit/AuditUtils.java