docs: @Override
      public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException {
          checkStart(clientIdentity.getDestination());
          CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
  
          // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
          boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
          if (!hasSubscribe) {
              return;
          }
          synchronized (canalInstance) {
              // 清除batch信息
              PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity,
                  batchId);
              if (positionRanges == null) { // 说明是重复的ack/rollback
                  throw new CanalServerException(String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
                      clientIdentity.getClientId(),
                      batchId));
              }
  
              // lastRollbackPostions.put(clientIdentity,
              // positionRanges.getEnd());// 记录一下最后rollback的位置
              // TODO 后续rollback到指定的batchId位置
              canalInstance.getEventStore().rollback();// rollback
                                                       // eventStore中的状态信息
              logger.info("rollback successfully, clientId:{} batchId:{} position:{}",
                  clientIdentity.getClientId(),
                  batchId,
                  positionRanges);
          }
      }