@Override
    public void open(final Map conf, final TopologyContext context, final SpoutOutputCollector collector) {
        final String topic = (String) conf.get(TOPIC);
        if (topic == null) {
            throw new IllegalArgumentException(TOPIC + " is null");
        }
        Integer maxSize = (Integer) conf.get(FETCH_MAX_SIZE);
        if (maxSize == null) {
            log.warn("Using default FETCH_MAX_SIZE");
            maxSize = DEFAULT_MAX_SIZE;
        }
        this.id2wrapperMap = new ConcurrentHashMap<Long, MetaMessageWrapper>();
        this.messageQueue = new LinkedTransferQueue<MetaMessageWrapper>();
        try {
            this.collector = collector;
            this.setUpMeta(topic, maxSize);
        }
        catch (final MetaClientException e) {
            log.error("Setup meta consumer failed", e);
        }
    }
Time in milliseconds to wait for a message from the queue if there is no
message ready when the topology requests a tuple (via
{@link #nextTuple()}).
metamorphosis-storm-spout/src/main/java/com/taobao/metamorphosis/storm/spout/MetaSpout.java