@Override
    public void run() {
        Message msg;
        while (true) {
            try {
                msg = bar.takeOrWait(queueIndex);
            } catch (InterruptedException e) {
                logger.info("interrupted queue index {}", queueIndex);
                break;
            }
            int msgId = msg.getId();
            recheck:
            while (true) {
                try {
                    bar.ensureIdIsMax(msgId);
                } catch (InterruptedException e) {
                    logger.info("ignore interrupted queue index {}", queueIndex);
                    continue;
                }
                Message next;
                while ((next = bar.poll(queueIndex)) != null) {
                    if (next.getId() < msgId) {
                        handler.pass(next);
                    } else {
                        handler.pass(msg);
                        msg = next;
                        msgId = msg.getId();
                        continue recheck;
                    }
                }
                handler.pass(msg);
                break;
            }
        }
    }