router 消费的offset 无法commit about dbus HOT 1 OPEN

从代码看，这里应该把

                
if (preValue != null) this.callBack.ack(preValue);

移到后面

    public void flush() {
        if (System.currentTimeMillis() - baseTime >= time) {
            for (String key : ackBooks.keySet()) {
                Map map = ackBooks.get(key);
                Ack preValue = null;
                boolean isFail = false;
                List okOffsetList = new ArrayList<>();
                for (Map.Entry entry : map.entrySet()) {
                    Ack value = entry.getValue();
                    if (Ack.INIT == value.getStatus()) {
                        logger.info("topic:{}, partaiton:{}, offset:{}, status: init",
                                value.getTopic(), value.getPartition(), value.getOffset());
                        break;
                    } else if (Ack.OK == value.getStatus()) {
                        logger.info("topic:{}, partaiton:{}, offset:{}, status: ok",
                                value.getTopic(), value.getPartition(), value.getOffset());
                        okOffsetList.add(value.getOffset());
                        preValue = value;
                        continue;
                    } else if (Ack.FAIL == value.getStatus()) {
                        logger.info("topic:{}, partaiton:{}, offset:{}, status: fail",
                                value.getTopic(), value.getPartition(), value.getOffset());
                        this.callBack.fail(value);
                        isFail = true;
                        break;
                    }
                }
                if (isFail) {
                    ackBooks.get(key).clear();
                } else {
                    if (preValue != null)
                        this.callBack.ack(preValue);
                    for (Long offset : okOffsetList) {
                        ackBooks.get(key).remove(offset);
                    }
                }
            }
            baseTime = System.currentTimeMillis();
        }
    }

from dbus.