@Override
    public Map<String, String> tableOptions() {
        Map<String, String> map = super.tableOptions();
        map.put(TubeMQConstant.CONNECTOR, TubeMQConstant.TUBEMQ);
        map.putAll(format.generateOptions(false));
        map.put(TubeMQConstant.TOPIC, topic);
        map.put(TubeMQConstant.MASTER_RPC, masterRpc);
        map.put(TubeMQConstant.CONSUME_GROUP, consumeGroup);
        map.put(TubeMQConstant.SESSION_KEY, sessionKey);

        if (null != streamId && !streamId.isEmpty()) {
            map.put(TubeMQConstant.STREAMID, StringUtils.concatCsv(streamId.toArray(new String[0]),
                    ',', null, null));
        }

        return map;
    }
The tubemq consumers use this streamId set to filter records reading from server.
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/TubeMQExtractNode.java