@Override
    public void fail(final Object msgId) {
        if (msgId instanceof Long) {
            final long id = (Long) msgId;
            final MetaMessageWrapper wrapper = this.id2wrapperMap.remove(id);
            if (wrapper == null) {
                log.warn(String.format("don't know how to reject(%s: %s)", msgId.getClass().getName(), msgId));
                return;
            }
            wrapper.success = false;
            wrapper.latch.countDown();
        }
        else {
            log.warn(String.format("don't know how to reject(%s: %s)", msgId.getClass().getName(), msgId));
        }

    }
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