code: @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() });
          }
      }
docs: 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿