docs: @Override
      public void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException {
          checkStart(clientIdentity.getDestination());
          checkSubscribe(clientIdentity);
  
          CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
          PositionRange<LogPosition> positionRanges = null;
          positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity, batchId); // 更新位置
          if (positionRanges == null) { // 说明是重复的ack/rollback
              throw new CanalServerException(String.format("ack error , clientId:%s batchId:%d is not exist , please check",
                  clientIdentity.getClientId(),
                  batchId));
          }
  
          // 更新cursor最好严格判断下位置是否有跳跃更新
          // Position position = lastRollbackPostions.get(clientIdentity);
          // if (position != null) {
          // // Position position =
          // canalInstance.getMetaManager().getCursor(clientIdentity);
          // LogPosition minPosition =
          // CanalEventUtils.min(positionRanges.getStart(), (LogPosition)
          // position);
          // if (minPosition == position) {// ack的position要晚于该最后ack的位置，可能有丢数据
          // throw new CanalServerException(
          // String.format(
          // "ack error , clientId:%s batchId:%d %s is jump ack , last ack:%s",
          // clientIdentity.getClientId(), batchId, positionRanges,
          // position));
          // }
          // }
  
          // 更新cursor
          if (positionRanges.getAck() != null) {
              canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck());
              if (logger.isInfoEnabled()) {
                  logger.info("ack successfully, clientId:{} batchId:{} position:{}",
                      clientIdentity.getClientId(),
                      batchId,
                      positionRanges);
              }
          }
  
          // 可定时清理数据
          canalInstance.getEventStore().ack(positionRanges.getEnd());
  
      }