@Override
    public void rollback(ClientIdentity clientIdentity) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
        if (!hasSubscribe) {
            return;
        }

        synchronized (canalInstance) {
            // 清除batch信息
            canalInstance.getMetaManager().clearAllBatchs(clientIdentity);
            // rollback eventStore中的状态信息
            canalInstance.getEventStore().rollback();
            logger.info("rollback successfully, clientId:{}", new Object[] { clientIdentity.getClientId() });
        }
    }
@ Override public void rollback ( ClientIdentity clientIdentity ) throws CanalServerException { checkStart ( clientIdentity . getDestination ( ) ) ; CanalInstance canalInstance = canalInstances . get ( clientIdentity . getDestination ( ) ) ; boolean hasSubscribe = canalInstance . getMetaManager ( ) . hasSubscribe ( clientIdentity ) ; if ( ! hasSubscribe ) { return ; } synchronized ( canalInstance ) { canalInstance . getMetaManager ( ) . clearAllBatchs ( clientIdentity ) ; canalInstance . getEventStore ( ) . rollback ( ) ; logger . info ( "rollback successfully, clientId:{}" , new Object [ ] { clientIdentity . getClientId ( ) } ) ; } }
回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿