public void putSystemHeader(final String msgType, final String msgTime) {
        this.msgType = null;
        this.msgTime = null;
        this.sysAttributes = null;
        StringBuilder strBuff = new StringBuilder(512);
        // valid parameters
        if (TStringUtils.isNotBlank(msgType)) {
            this.msgType = msgType.trim();
            this.sysAttributes = strBuff.append(TokenConstants.TOKEN_MSG_TYPE)
                    .append(TokenConstants.EQ).append(this.msgType).toString();
            strBuff.delete(0, strBuff.length());
        }
        if (TStringUtils.isNotBlank(msgTime)) {
            String tmpMsgTime = msgTime.trim();
            if (tmpMsgTime.length() != DateTimeConvertUtils.LENGTH_YYYYMMDDHHMM) {
                throw new IllegalArgumentException(strBuff
                        .append("Illegal parameter: msgTime's value must '")
                        .append(DateTimeConvertUtils.PAT_YYYYMMDDHHMM)
                        .append("' format and length must equal ")
                        .append(DateTimeConvertUtils.LENGTH_YYYYMMDDHHMM).toString());
            }
            try {
                DateTimeConvertUtils.yyyyMMddHHmm2ms(tmpMsgTime);
            } catch (Throwable ex) {
                throw new IllegalArgumentException(strBuff
                        .append("Illegal parameter: msgTime's value parse failure: ")
                        .append(ex.getMessage()).toString());
            }
            this.msgTime = tmpMsgTime;
            if (TStringUtils.isNotEmpty(this.sysAttributes)) {
                strBuff.append(sysAttributes).append(TokenConstants.ARRAY_SEP);
            }
            this.sysAttributes = strBuff.append(TokenConstants.TOKEN_MSG_TIME)
                    .append(TokenConstants.EQ).append(this.msgTime).toString();
            strBuff.delete(0, strBuff.length());
        }
        // rebuild attributes
        if (TStringUtils.isNotEmpty(this.sysAttributes)) {
            strBuff.append(this.sysAttributes);
        }
        if (TStringUtils.isNotBlank(this.attribute)) {
            String[] strAttrs = this.attribute.split(TokenConstants.ARRAY_SEP);
            for (String strAttrItem : strAttrs) {
                if (strAttrItem != null && !(strAttrItem.contains(TokenConstants.TOKEN_MSG_TYPE)
                        || strAttrItem.contains(TokenConstants.TOKEN_MSG_TIME))) {
                    if (strBuff.length() == 0) {
                        strBuff.append(strAttrItem);
                    } else {
                        strBuff.append(TokenConstants.ARRAY_SEP).append(strAttrItem);
                    }
                }
            }
        }
        this.attribute = strBuff.toString();
    }
Set the filtering items that need to be filtered and the message forwarding time

@param msgType   the ID of the stream that needs to be filtered,
                if not required, set null
@param msgTime   the time when the message was sent, the format is yyyyMMddHHmm,
                 if not required, set null
inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/Message.java