本文整理了Java中org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore.releaseAndTryRemove()方法的一些代码示例，展示了ZooKeeperStateHandleStore.releaseAndTryRemove()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ZooKeeperStateHandleStore.releaseAndTryRemove()方法的具体详情如下：
包路径：org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore
类名称：ZooKeeperStateHandleStore
方法名：releaseAndTryRemove

ZooKeeperStateHandleStore.releaseAndTryRemove介绍

[英]Releases the lock for the given state node and tries to remove the state node if it is no longer locked. The deletion of the state node is executed asynchronously.

Important: This also discards the stored state handle after the given action has been executed.
[中]释放给定状态节点的锁，并在状态节点不再锁定时尝试移除该节点。状态节点的删除是异步执行的。
重要提示：这也会在执行给定操作后丢弃存储的状态句柄。

代码示例

代码示例来源：origin: apache/flink

@Override
public boolean removeWorker(Protos.TaskID taskID) throws Exception {
  checkNotNull(taskID, "taskID");
  String path = getPathForWorker(taskID);
  synchronized (startStopLock) {
    verifyIsRunning();
    if (workersInZooKeeper.exists(path) == -1) {
      LOG.debug("No such worker {} in ZooKeeper.", taskID);
      return false;
    }
    workersInZooKeeper.releaseAndTryRemove(path);
    LOG.debug("Removed worker {} from ZooKeeper.", taskID);
    return true;
  }
}

代码示例来源：origin: com.alibaba.blink/flink-runtime

/**
 * Releases the lock for the given state node and tries to remove the state node if it is no longer locked.
 * The deletion of the state node is executed asynchronously.
 *
 * <p><strong>Important</strong>: This also discards the stored state handle after the given action
 * has been executed.
 *
 * @param pathInZooKeeper Path of state handle to remove (expected to start with a '/')
 * @throws Exception If the ZooKeeper operation fails
 */
public void releaseAndTryRemove(String pathInZooKeeper) throws Exception {
  releaseAndTryRemove(pathInZooKeeper, null);
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

/**
 * Releases the lock for the given state node and tries to remove the state node if it is no longer locked.
 * The deletion of the state node is executed asynchronously.
 *
 * <p><strong>Important</strong>: This also discards the stored state handle after the given action
 * has been executed.
 *
 * @param pathInZooKeeper Path of state handle to remove (expected to start with a '/')
 * @throws Exception If the ZooKeeper operation fails
 */
public void releaseAndTryRemove(String pathInZooKeeper) throws Exception {
  releaseAndTryRemove(pathInZooKeeper, null);
}

代码示例来源：origin: org.apache.flink/flink-runtime

/**
 * Tries to remove the checkpoint identified by the given checkpoint id.
 *
 * @param checkpointId identifying the checkpoint to remove
 * @return true if the checkpoint could be removed
 */
private boolean tryRemove(long checkpointId) throws Exception {
  return checkpointsInZooKeeper.releaseAndTryRemove(checkpointIdToPath(checkpointId));
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.11

/**
 * Tries to remove the checkpoint identified by the given checkpoint id.
 *
 * @param checkpointId identifying the checkpoint to remove
 * @return true if the checkpoint could be removed
 */
private boolean tryRemove(long checkpointId) throws Exception {
  return checkpointsInZooKeeper.releaseAndTryRemove(checkpointIdToPath(checkpointId));
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

@Override
public void removeJobGraph(JobID jobId) throws Exception {
  checkNotNull(jobId, "Job ID");
  String path = getPathForJob(jobId);
  LOG.debug("Removing job graph {} from {}{}.", jobId, zooKeeperFullBasePath, path);
  synchronized (cacheLock) {
    if (addedJobGraphs.contains(jobId)) {
      jobGraphsInZooKeeper.releaseAndTryRemove(path);
      addedJobGraphs.remove(jobId);
    }
  }
  LOG.info("Removed job graph {} from ZooKeeper.", jobId);
}

代码示例来源：origin: com.alibaba.blink/flink-runtime

@Override
public void removeJobGraph(JobID jobId) throws Exception {
  checkNotNull(jobId, "Job ID");
  String path = getPathForJob(jobId);
  LOG.debug("Removing job graph {} from {}{}.", jobId, zooKeeperFullBasePath, path);
  synchronized (cacheLock) {
    if (addedJobGraphs.contains(jobId)) {
      jobGraphsInZooKeeper.releaseAndTryRemove(path);
      addedJobGraphs.remove(jobId);
    }
  }
  LOG.info("Removed job graph {} from ZooKeeper.", jobId);
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.11

@Override
public void removeJobGraph(JobID jobId) throws Exception {
  checkNotNull(jobId, "Job ID");
  String path = getPathForJob(jobId);
  LOG.debug("Removing job graph {} from {}{}.", jobId, zooKeeperFullBasePath, path);
  synchronized (cacheLock) {
    if (addedJobGraphs.contains(jobId)) {
      if (jobGraphsInZooKeeper.releaseAndTryRemove(path)) {
        addedJobGraphs.remove(jobId);
      } else {
        throw new FlinkException(String.format("Could not remove job graph with job id %s from ZooKeeper.", jobId));
      }
    }
  }
  LOG.info("Removed job graph {} from ZooKeeper.", jobId);
}

代码示例来源：origin: org.apache.flink/flink-runtime

@Override
public void removeJobGraph(JobID jobId) throws Exception {
  checkNotNull(jobId, "Job ID");
  String path = getPathForJob(jobId);
  LOG.debug("Removing job graph {} from {}{}.", jobId, zooKeeperFullBasePath, path);
  synchronized (cacheLock) {
    if (addedJobGraphs.contains(jobId)) {
      if (jobGraphsInZooKeeper.releaseAndTryRemove(path)) {
        addedJobGraphs.remove(jobId);
      } else {
        throw new FlinkException(String.format("Could not remove job graph with job id %s from ZooKeeper.", jobId));
      }
    }
  }
  LOG.info("Removed job graph {} from ZooKeeper.", jobId);
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

/**
 * Releases all lock nodes of this ZooKeeperStateHandleStores and tries to remove all state nodes which
 * are not locked anymore.
 *
 * <p>The delete operation is executed asynchronously
 *
 * @throws Exception if the delete operation fails
 */
public void releaseAndTryRemoveAll() throws Exception {
  Collection<String> children = getAllPaths();
  Exception exception = null;
  for (String child : children) {
    try {
      releaseAndTryRemove('/' + child);
    } catch (Exception e) {
      exception = ExceptionUtils.firstOrSuppressed(e, exception);
    }
  }
  if (exception != null) {
    throw new Exception("Could not properly release and try removing all state nodes.", exception);
  }
}

代码示例来源：origin: com.alibaba.blink/flink-runtime

/**
 * Releases all lock nodes of this ZooKeeperStateHandleStores and tries to remove all state nodes which
 * are not locked anymore.
 *
 * <p>The delete operation is executed asynchronously
 *
 * @throws Exception if the delete operation fails
 */
public void releaseAndTryRemoveAll() throws Exception {
  Collection<String> children = getAllPaths();
  Exception exception = null;
  for (String child : children) {
    try {
      releaseAndTryRemove('/' + child);
    } catch (Exception e) {
      exception = ExceptionUtils.firstOrSuppressed(e, exception);
    }
  }
  if (exception != null) {
    throw new Exception("Could not properly release and try removing all state nodes.", exception);
  }
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.11

/**
 * Releases all lock nodes of this ZooKeeperStateHandleStores and tries to remove all state nodes which
 * are not locked anymore.
 *
 * <p>The delete operation is executed asynchronously
 *
 * @throws Exception if the delete operation fails
 */
public void releaseAndTryRemoveAll() throws Exception {
  Collection<String> children = getAllPaths();
  Exception exception = null;
  for (String child : children) {
    try {
      releaseAndTryRemove('/' + child);
    } catch (Exception e) {
      exception = ExceptionUtils.firstOrSuppressed(e, exception);
    }
  }
  if (exception != null) {
    throw new Exception("Could not properly release and try removing all state nodes.", exception);
  }
}

代码示例来源：origin: org.apache.flink/flink-runtime

/**
 * Releases all lock nodes of this ZooKeeperStateHandleStores and tries to remove all state nodes which
 * are not locked anymore.
 *
 * <p>The delete operation is executed asynchronously
 *
 * @throws Exception if the delete operation fails
 */
public void releaseAndTryRemoveAll() throws Exception {
  Collection<String> children = getAllPaths();
  Exception exception = null;
  for (String child : children) {
    try {
      releaseAndTryRemove('/' + child);
    } catch (Exception e) {
      exception = ExceptionUtils.firstOrSuppressed(e, exception);
    }
  }
  if (exception != null) {
    throw new Exception("Could not properly release and try removing all state nodes.", exception);
  }
}

代码示例来源：origin: com.alibaba.blink/flink-runtime

/**
 * Removes a checkpoint from ZooKeeper because of Job shutdown and drops the state.
 */
private void removeShutdown(
    final CompletedCheckpoint completedCheckpoint,
    final JobStatus jobStatus) throws Exception {
  if (completedCheckpoint == null) {
    return;
  }
  ZooKeeperStateHandleStore.RemoveCallback<CompletedCheckpoint> removeAction = new ZooKeeperStateHandleStore.RemoveCallback<CompletedCheckpoint>() {
    @Override
    public void apply(@Nullable RetrievableStateHandle<CompletedCheckpoint> value) throws FlinkException {
      try {
        completedCheckpoint.discardOnShutdown(jobStatus);
      } catch (Exception e) {
        throw new FlinkException("Could not discard the completed checkpoint on subsume.", e);
      }
    }
  };
  checkpointsInZooKeeper.releaseAndTryRemove(
    checkpointIdToPath(completedCheckpoint.getCheckpointID()),
    removeAction);
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

/**
 * Removes a subsumed checkpoint from ZooKeeper and drops the state.
 */
private void removeSubsumed(
  final CompletedCheckpoint completedCheckpoint) throws Exception {
  if (completedCheckpoint == null) {
    return;
  }
  ZooKeeperStateHandleStore.RemoveCallback<CompletedCheckpoint> action =
    new ZooKeeperStateHandleStore.RemoveCallback<CompletedCheckpoint>() {
      @Override
      public void apply(@Nullable RetrievableStateHandle<CompletedCheckpoint> value) throws FlinkException {
        if (value != null) {
          try {
            completedCheckpoint.discardOnSubsume();
          } catch (Exception e) {
            throw new FlinkException("Could not discard the completed checkpoint on subsume.", e);
          }
        }
      }
    };
  checkpointsInZooKeeper.releaseAndTryRemove(
    checkpointIdToPath(completedCheckpoint.getCheckpointID()),
    action);
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

/**
 * Removes a checkpoint from ZooKeeper because of Job shutdown and drops the state.
 */
private void removeShutdown(
    final CompletedCheckpoint completedCheckpoint,
    final JobStatus jobStatus) throws Exception {
  if (completedCheckpoint == null) {
    return;
  }
  ZooKeeperStateHandleStore.RemoveCallback<CompletedCheckpoint> removeAction = new ZooKeeperStateHandleStore.RemoveCallback<CompletedCheckpoint>() {
    @Override
    public void apply(@Nullable RetrievableStateHandle<CompletedCheckpoint> value) throws FlinkException {
      try {
        completedCheckpoint.discardOnShutdown(jobStatus);
      } catch (Exception e) {
        throw new FlinkException("Could not discard the completed checkpoint on subsume.", e);
      }
    }
  };
  checkpointsInZooKeeper.releaseAndTryRemove(
    checkpointIdToPath(completedCheckpoint.getCheckpointID()),
    removeAction);
}

代码示例来源：origin: com.alibaba.blink/flink-runtime

/**
 * Removes a subsumed checkpoint from ZooKeeper and drops the state.
 */
private void removeSubsumed(
  final CompletedCheckpoint completedCheckpoint) throws Exception {
  if (completedCheckpoint == null) {
    return;
  }
  ZooKeeperStateHandleStore.RemoveCallback<CompletedCheckpoint> action =
    new ZooKeeperStateHandleStore.RemoveCallback<CompletedCheckpoint>() {
      @Override
      public void apply(@Nullable RetrievableStateHandle<CompletedCheckpoint> value) throws FlinkException {
        if (value != null) {
          try {
            completedCheckpoint.discardOnSubsume();
          } catch (Exception e) {
            throw new FlinkException("Could not discard the completed checkpoint on subsume.", e);
          }
        }
      }
    };
  checkpointsInZooKeeper.releaseAndTryRemove(
    checkpointIdToPath(completedCheckpoint.getCheckpointID()),
    action);
}

代码示例来源：origin: org.apache.flink/flink-runtime_2.10

"corrupted data. Releasing and trying to remove this node.", path, ioException);
releaseAndTryRemove(path);

代码示例来源：origin: com.alibaba.blink/flink-runtime

"corrupted data. Releasing and trying to remove this node.", path, ioException);
releaseAndTryRemove(path);