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

ZooKeeperStateHandleStore介绍

[英]Class which stores state via the provided RetrievableStateStorageHelper and writes the returned state handle to ZooKeeper. The ZooKeeper node can be locked by creating an ephemeral child and only allowing the deletion of the ZooKeeper node if it does not have any children. That way we protect concurrent accesses from different ZooKeeperStateHandleStore instances.

Added state is persisted via RetrievableStateHandle, which in turn are written to ZooKeeper. This level of indirection is necessary to keep the amount of data in ZooKeeper small. ZooKeeper is build for data in the KB range whereas state can grow to multiple MBs.

State modifications require some care, because it is possible that certain failures bring the state handle backend and ZooKeeper out of sync.

ZooKeeper holds the ground truth about state handles, i.e. the following holds:

State handle in ZooKeeper => State handle exists

But not:

State handle exists => State handle in ZooKeeper

There can be lingering state handles when failures happen during operation. They need to be cleaned up manually (see FLINK-2513 about a possible way to overcome this).
[中]类，该类通过提供的RetrievableStateStorageHelper存储状态，并将返回的状态句柄写入ZooKeeper。ZooKeeper节点可以通过创建短暂的子节点来锁定，并且只有在没有子节点时才允许删除ZooKeeper节点。这样我们就可以保护来自不同ZookePerstateHandleStore实例的并发访问。
添加的状态通过RetrievableStateHandle持久化，然后写入ZooKeeper。这种级别的间接寻址对于保持ZooKeeper中的数据量较小是必要的。ZooKeeper是为KB范围内的数据构建的，而state可以增长到多MB。
状态修改需要谨慎，因为某些故障可能会导致状态句柄后端和ZooKeeper不同步。
ZooKeeper持有关于州手柄的基本事实，即以下观点：

State handle in ZooKeeper => State handle exists

但不是：

State handle exists => State handle in ZooKeeper

当操作过程中发生故障时，可能会有延迟状态句柄。它们需要手动清理（请参阅FLINK-2513了解克服此问题的可能方法）。

代码示例

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

@Override
public void putWorker(MesosWorkerStore.Worker worker) throws Exception {
  checkNotNull(worker, "worker");
  String path = getPathForWorker(worker.taskID());
  synchronized (startStopLock) {
    verifyIsRunning();
    int currentVersion = workersInZooKeeper.exists(path);
    if (currentVersion == -1) {
      workersInZooKeeper.addAndLock(path, worker);
      LOG.debug("Added {} in ZooKeeper.", worker);
    } else {
      workersInZooKeeper.replace(path, currentVersion, worker);
      LOG.debug("Updated {} in ZooKeeper.", worker);
    }
  }
}

代码示例来源：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: org.apache.flink/flink-runtime_2.11

/**
 * Releases the lock for the given state node and tries to remove the state node if it is no longer locked.
 * It returns the {@link RetrievableStateHandle} stored under the given state node if any.
 *
 * @param pathInZooKeeper Path of state handle to remove
 * @return True if the state handle could be released
 * @throws Exception If the ZooKeeper operation or discarding the state handle fails
 */
@Nullable
public boolean releaseAndTryRemove(String pathInZooKeeper) throws Exception {
  checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
  final String path = normalizePath(pathInZooKeeper);
  RetrievableStateHandle<T> stateHandle = null;
  try {
    stateHandle = get(path, false);
  } catch (Exception e) {
    LOG.warn("Could not retrieve the state handle from node {}.", path, e);
  }
  release(pathInZooKeeper);
  try {
    client.delete().forPath(path);
  } catch (KeeperException.NotEmptyException ignored) {
    LOG.debug("Could not delete znode {} because it is still locked.", path);
    return false;
  }
  if (stateHandle != null) {
    stateHandle.discardState();
  }
  return true;
}

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

checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
final String path = normalizePath(pathInZooKeeper);
    client.create().withMode(CreateMode.EPHEMERAL).forPath(getLockPath(path));
  } catch (KeeperException.NodeExistsException ignored) {
  if (!success && lock) {
    release(path);

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

/**
 * Releases the lock from the node under the given ZooKeeper path. If no lock exists, then nothing happens.
 *
 * @param pathInZooKeeper Path describing the ZooKeeper node
 * @throws Exception if the delete operation of the lock node fails
 */
public void release(String pathInZooKeeper) throws Exception {
  final String path = normalizePath(pathInZooKeeper);
  try {
    client.delete().forPath(getLockPath(path));
  } catch (KeeperException.NoNodeException ignored) {
    // we have never locked this node
  } catch (Exception e) {
    throw new Exception("Could not release the lock: " + getLockPath(pathInZooKeeper) + '.', e);
  }
}

代码示例来源：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 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: com.alibaba.blink/flink-runtime

/**
 * Releases all lock nodes of this ZooKeeperStateHandleStore.
 *
 * @throws Exception if the delete operation of a lock file fails
 */
public void releaseAll() throws Exception {
  Collection<String> children = getAllPaths();
  Exception exception = null;
  for (String child: children) {
    try {
      release(child);
    } catch (Exception e) {
      exception = ExceptionUtils.firstOrSuppressed(e, exception);
    }
  }
  if (exception != null) {
    throw new Exception("Could not properly release all state nodes.", exception);
  }
}

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

checkNotNull(state, "State");
final String path = normalizePath(pathInZooKeeper);
RetrievableStateHandle<T> oldStateHandle = get(path, false);

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

jobGraphRetrievableStateHandle = jobGraphsInZooKeeper.getAndLock(path);
  } catch (KeeperException.NoNodeException ignored) {
    success = true;
} finally {
  if (!success) {
    jobGraphsInZooKeeper.release(path);

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

/**
 * Creates a {@link ZooKeeperStateHandleStore} instance with the provided arguments.
 *
 * @param zkStateHandleStorePath specifying the path in ZooKeeper to store the state handles to
 * @param stateStorageHelper storing the actual state data
 * @param <T> Type of the state to be stored
 * @return a ZooKeeperStateHandleStore instance
 * @throws Exception if ZooKeeper could not create the provided state handle store path in
 *     ZooKeeper
 */
public <T extends Serializable> ZooKeeperStateHandleStore<T> createZooKeeperStateHandleStore(
    String zkStateHandleStorePath,
    RetrievableStateStorageHelper<T> stateStorageHelper) throws Exception {
  facade.newNamespaceAwareEnsurePath(zkStateHandleStorePath).ensure(facade.getZookeeperClient());
  CuratorFramework stateHandleStoreFacade = facade.usingNamespace(
    ZooKeeperUtils.generateZookeeperPath(
      facade.getNamespace(),
      zkStateHandleStorePath));
  return new ZooKeeperStateHandleStore<>(stateHandleStoreFacade, stateStorageHelper);
}

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

final RetrievableStateHandle<T> stateHandle = getAndLock(path);
  stateHandles.add(new Tuple2<>(stateHandle, path));
} catch (KeeperException.NoNodeException ignored) {
    "corrupted data. Releasing and trying to remove this node.", path, ioException);
  releaseAndTryRemove(path);

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

/**
 * Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones.
 *
 * @param checkpoint Completed checkpoint to add.
 */
@Override
public void addCheckpoint(final CompletedCheckpoint checkpoint) throws Exception {
  checkNotNull(checkpoint, "Checkpoint");
  final String path = checkpointIdToPath(checkpoint.getCheckpointID());
  // Now add the new one. If it fails, we don't want to loose existing data.
  checkpointsInZooKeeper.addAndLock(path, checkpoint);
  completedCheckpoints.addLast(checkpoint);
  // Everything worked, let's remove a previous checkpoint if necessary.
  while (completedCheckpoints.size() > maxNumberOfCheckpointsToRetain) {
    try {
      removeSubsumed(completedCheckpoints.removeFirst());
    } catch (Exception e) {
      LOG.warn("Failed to subsume the old checkpoint", e);
    }
  }
  LOG.debug("Added {} to {}.", checkpoint, path);
}

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

/**
 * Gets the {@link RetrievableStateHandle} stored in the given ZooKeeper node and locks it. A
 * locked node cannot be removed by another {@link ZooKeeperStateHandleStore} instance as long
 * as this instance remains connected to ZooKeeper.
 *
 * @param pathInZooKeeper Path to the ZooKeeper node which contains the state handle
 * @return The retrieved state handle from the specified ZooKeeper node
 * @throws IOException Thrown if the method failed to deserialize the stored state handle
 * @throws Exception Thrown if a ZooKeeper operation failed
 */
public RetrievableStateHandle<T> getAndLock(String pathInZooKeeper) throws Exception {
  return get(pathInZooKeeper, true);
}

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

/**
 * Returns the version of the node if it exists or <code>-1</code> if it doesn't.
 *
 * @param pathInZooKeeper Path in ZooKeeper to check
 * @return Version of the ZNode if the path exists, <code>-1</code> otherwise.
 * @throws Exception If the ZooKeeper operation fails
 */
public int exists(String pathInZooKeeper) throws Exception {
  checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
  final String path = normalizePath(pathInZooKeeper);
  Stat stat = client.checkExists().forPath(path);
  if (stat != null) {
    return stat.getVersion();
  }
  return -1;
}

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

@Override
public Collection<JobID> getJobIds() throws Exception {
  Collection<String> paths;
  LOG.debug("Retrieving all stored job ids from ZooKeeper under {}.", zooKeeperFullBasePath);
  try {
    paths = jobGraphsInZooKeeper.getAllPaths();
  } catch (Exception e) {
    throw new Exception("Failed to retrieve entry paths from ZooKeeperStateHandleStore.", e);
  }
  List<JobID> jobIds = new ArrayList<>(paths.size());
  for (String path : paths) {
    try {
      jobIds.add(jobIdfromPath(path));
    } catch (Exception exception) {
      LOG.warn("Could not parse job id from {}. This indicates a malformed path.", path, exception);
    }
  }
  return jobIds;
}

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

final RetrievableStateHandle<T> stateHandle = getAndLock(path);
  stateHandles.add(new Tuple2<>(stateHandle, path));
} catch (KeeperException.NoNodeException ignored) {

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

/**
 * Releases the lock for the given state node and tries to remove the state node if it is no longer locked.
 * It returns the {@link RetrievableStateHandle} stored under the given state node if any.
 *
 * @param pathInZooKeeper Path of state handle to remove
 * @return True if the state handle could be released
 * @throws Exception If the ZooKeeper operation or discarding the state handle fails
 */
@Nullable
public boolean releaseAndTryRemove(String pathInZooKeeper) throws Exception {
  checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
  final String path = normalizePath(pathInZooKeeper);
  RetrievableStateHandle<T> stateHandle = null;
  try {
    stateHandle = get(path, false);
  } catch (Exception e) {
    LOG.warn("Could not retrieve the state handle from node {}.", path, e);
  }
  release(pathInZooKeeper);
  try {
    client.delete().forPath(path);
  } catch (KeeperException.NotEmptyException ignored) {
    LOG.debug("Could not delete znode {} because it is still locked.", path);
    return false;
  }
  if (stateHandle != null) {
    stateHandle.discardState();
  }
  return true;
}

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

checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
final String path = normalizePath(pathInZooKeeper);
    client.create().withMode(CreateMode.EPHEMERAL).forPath(getLockPath(path));
  } catch (KeeperException.NodeExistsException ignored) {
  if (!success && lock) {
    release(path);

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

/**
 * Releases the lock from the node under the given ZooKeeper path. If no lock exists, then nothing happens.
 *
 * @param pathInZooKeeper Path describing the ZooKeeper node
 * @throws Exception if the delete operation of the lock node fails
 */
public void release(String pathInZooKeeper) throws Exception {
  final String path = normalizePath(pathInZooKeeper);
  try {
    client.delete().forPath(getLockPath(path));
  } catch (KeeperException.NoNodeException ignored) {
    // we have never locked this node
  } catch (Exception e) {
    throw new Exception("Could not release the lock: " + getLockPath(pathInZooKeeper) + '.', e);
  }
}