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

ZooKeeperCommandExecutor介绍

暂无

代码示例

代码示例来源：origin: line/centraldogma

private <T> T blockingExecute(Command<T> command) throws Exception {
  createParentNodes();
  try (SafeLock ignored = safeLock(command.executionPath())) {
    // NB: We are sure no other replicas will append the conflicting logs (the commands with the
    //     same execution path) while we hold the lock for the command's execution path.
    //
    //     Other replicas may still append the logs with different execution paths, because, by design,
    //     two commands never conflict with each other if they have different execution paths.
    final List<String> recentRevisions = curator.getChildren().forPath(absolutePath(LOG_PATH));
    if (!recentRevisions.isEmpty()) {
      final long lastRevision = recentRevisions.stream().mapToLong(Long::parseLong).max().getAsLong();
      replayLogs(lastRevision);
    }
    final T result = delegate.execute(command).get();
    final ReplicationLog<T> log = new ReplicationLog<>(replicaId(), command, result);
    // Store the command execution log to ZooKeeper.
    final long revision = storeLog(log);
    logger.debug("logging OK. revision = {}, log = {}", revision, log);
    return result;
  }
}

代码示例来源：origin: com.linecorp.centraldogma/centraldogma-server-shaded

@Override
protected <T> CompletableFuture<T> doExecute(int replicaId, Command<T> command) throws Exception {
  final CompletableFuture<T> future = new CompletableFuture<>();
  executor.execute(() -> {
    try {
      future.complete(blockingExecute(replicaId, command));
    } catch (Throwable t) {
      future.completeExceptionally(t);
    }
  });
  return future;
}

代码示例来源：origin: com.linecorp.centraldogma/centraldogma-server

private void createParentNodes() throws Exception {
  if (createdParentNodes) {
    return;
  }
  // Create the zkPath if it does not exist.
  createZkPathIfMissing(absolutePath());
  createZkPathIfMissing(absolutePath(LOG_PATH));
  createZkPathIfMissing(absolutePath(LOG_BLOCK_PATH));
  createZkPathIfMissing(absolutePath(LOCK_PATH));
  createdParentNodes = true;
}

代码示例来源：origin: line/centraldogma

@VisibleForTesting
Optional<ReplicationLog<?>> loadLog(long revision, boolean skipIfSameReplica) {
  try {
    createParentNodes();
    final String logPath = absolutePath(LOG_PATH) + '/' + pathFromRevision(revision);
    final LogMeta logMeta = Jackson.readValue(curator.getData().forPath(logPath), LogMeta.class);
    if (skipIfSameReplica && replicaId() == logMeta.replicaId()) {
      return Optional.empty();
    }
    final byte[] bytes = new byte[logMeta.size()];
    int offset = 0;
    for (long blockId : logMeta.blocks()) {
      final String blockPath = absolutePath(LOG_BLOCK_PATH) + '/' + pathFromRevision(blockId);
      final byte[] b = curator.getData().forPath(blockPath);
      System.arraycopy(b, 0, bytes, offset, b.length);
      offset += b.length;
    }
    assert logMeta.size() == offset;
    final ReplicationLog<?> log = Jackson.readValue(bytes, ReplicationLog.class);
    return Optional.of(log);
  } catch (Exception e) {
    logger.error("Failed to load a log at revision {}; entering read-only mode", revision, e);
    stopLater();
    throw new ReplicationException("failed to load a log at revision " + revision, e);
  }
}

代码示例来源：origin: line/centraldogma

for (;;) {
    final long nextRevision = info.lastReplayedRevision + 1;
    final Optional<ReplicationLog<?>> log = loadLog(nextRevision, true);
    if (log.isPresent()) {
      final ReplicationLog<?> l = log.get();
  logger.error("Failed to replay a log at revision {}; entering read-only mode",
         info.lastReplayedRevision, e);
  stopLater();
  updateLastReplayedRevision(targetRevision);
} catch (Exception e) {
  logger.error("Failed to update {} to {}; entering read-only mode",
         revisionFile, targetRevision, e);
  stopLater();
  throw new ReplicationException("failed to update " + revisionFile + " to " + targetRevision, e);

代码示例来源：origin: com.linecorp.centraldogma/centraldogma-server

private long storeLog(ReplicationLog<?> log) {
  try {
    final byte[] bytes = Jackson.writeValueAsBytes(log);
    assert bytes.length > 0;
    final LogMeta logMeta = new LogMeta(log.replicaId(), System.currentTimeMillis(), bytes.length);
    final int count = (bytes.length + MAX_BYTES - 1) / MAX_BYTES;
    for (int i = 0; i < count; ++i) {
      final int start = i * MAX_BYTES;
      final int end = Math.min((i + 1) * MAX_BYTES, bytes.length);
      final byte[] b = Arrays.copyOfRange(bytes, start, end);
      final String blockPath = curator.create()
                      .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                      .forPath(absolutePath(LOG_BLOCK_PATH) + '/', b);
      final long blockId = revisionFromPath(blockPath);
      logMeta.appendBlock(blockId);
    }
    final String logPath =
        curator.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL)
            .forPath(absolutePath(LOG_PATH) + '/', Jackson.writeValueAsBytes(logMeta));
    return revisionFromPath(logPath);
  } catch (Exception e) {
    logger.error("Failed to store a log; entering read-only mode: {}", log, e);
    stopLater();
    throw new ReplicationException("failed to store a log: " + log, e);
  }
}

代码示例来源：origin: line/centraldogma

lastReplayedRevision = getLastReplayedRevision();
  listenerInfo = new ListenerInfo(lastReplayedRevision, onTakeLeadership, onReleaseLeadership);
} catch (Exception e) {
quorumPeer = startZooKeeper();
retryPolicy = RETRY_POLICY_ALWAYS;
logWatcher = new PathChildrenCache(curator, absolutePath(LOG_PATH),
                  true, false, logWatcherExecutor);
logWatcher.getListenable().addListener(this, MoreExecutors.directExecutor());
leaderSelectorExecutor = Executors.newSingleThreadExecutor(
    new DefaultThreadFactory("zookeeper-leader-selector", true));
leaderSelector = new LeaderSelector(curator, absolutePath(LEADER_PATH),
                  leaderSelectorExecutor, oldLogRemover);
leaderSelector.start();

代码示例来源：origin: line/centraldogma

private SafeLock safeLock(String executionPath) {
  final InterProcessMutex mtx = mutexMap.computeIfAbsent(
      executionPath, k -> new InterProcessMutex(curator, absolutePath(LOCK_PATH, executionPath)));
  try {
    mtx.acquire();
  } catch (Exception e) {
    logger.error("Failed to acquire a lock for {}; entering read-only mode", executionPath, e);
    stopLater();
    throw new ReplicationException("failed to acquire a lock for " + executionPath, e);
  }
  return () -> {
    try {
      mtx.release();
    } catch (Exception ignored) {
      // Ignore.
    }
  };
}

代码示例来源：origin: line/centraldogma

copyZkProperty(zkProps, "initLimit", "5");
copyZkProperty(zkProps, "syncLimit", "10");
copyZkProperty(zkProps, "tickTime", "3000");
copyZkProperty(zkProps, "syncEnabled", "true");
copyZkProperty(zkProps, "autopurge.snapRetainCount", "3");
copyZkProperty(zkProps, "autopurge.purgeInterval", "1");
  if (isStopping()) {
    throw new InterruptedException("Stop requested before joining the cluster");

代码示例来源：origin: line/centraldogma

final Replica replica3 = cluster.get(2);
final Replica replica4 = cluster.get(3);
replica4.rm.stop().join();
  replica1.rm.execute(command1).join();
  final Optional<ReplicationLog<?>> commandResult2 = replica1.rm.loadLog(0, false);
  assertThat(commandResult2.get().command()).isEqualTo(command1);
  assertThat(commandResult2.get().result()).isNull();
  replica3.rm.stop().join();
  replica1.rm.execute(command2).join();
  await().untilAsserted(() -> verify(replica1.delegate).apply(eq(command2)));
  await().untilAsserted(() -> verify(replica2.delegate).apply(eq(command2)));
  replica3.rm.start().join();
  verifyTwoIndependentCommands(replica3, command1, command2);
  replica4.rm.start().join();
  verifyTwoIndependentCommands(replica4, command1, command2);
} finally {
  for (Replica r : cluster) {
    r.rm.stop();

代码示例来源：origin: line/centraldogma

for (int j = 0; j < COMMANDS_PER_REPLICA; j++) {
      try {
        r.rm.execute(command).join();
      } catch (Exception e) {
        throw new Error(e);
    @SuppressWarnings("unchecked")
    final ReplicationLog<Revision> log =
        (ReplicationLog<Revision>) r.rm.loadLog(i, false).get();
replicas.forEach(r -> r.rm.stop());

代码示例来源：origin: line/centraldogma

Replica(InstanceSpec spec, Map<Integer, ZooKeeperAddress> servers,
    Function<Command<?>, CompletableFuture<?>> delegate, boolean start) throws Exception {
  this.delegate = delegate;
  dataDir = spec.getDataDirectory();
  final int id = spec.getServerId();
  final ZooKeeperReplicationConfig zkCfg = new ZooKeeperReplicationConfig(id, servers);
  rm = new ZooKeeperCommandExecutor(zkCfg, dataDir, new AbstractCommandExecutor(null, null) {
    @Override
    public int replicaId() {
      return id;
    }
    @Override
    protected void doStart(@Nullable Runnable onTakeLeadership,
                @Nullable Runnable onReleaseLeadership) {}
    @Override
    protected void doStop(@Nullable Runnable onReleaseLeadership) {}
    @Override
    @SuppressWarnings("unchecked")
    protected <T> CompletableFuture<T> doExecute(Command<T> command) {
      return (CompletableFuture<T>) delegate.apply(command);
    }
  }, null, null);
  startFuture = start ? rm.start() : null;
}

代码示例来源：origin: line/centraldogma

private CommandExecutor newZooKeeperCommandExecutor(ProjectManager pm, Executor repositoryWorker,
                          @Nullable SessionManager sessionManager,
                          @Nullable Consumer<CommandExecutor> onTakeLeadership,
                          @Nullable Runnable onReleaseLeadership) {
  final ZooKeeperReplicationConfig zkCfg = (ZooKeeperReplicationConfig) cfg.replicationConfig();
  // Delete the old UUID replica ID which is not used anymore.
  new File(cfg.dataDir(), "replica_id").delete();
  // TODO(trustin): Provide a way to restart/reload the replicator
  //                so that we can recover from ZooKeeper maintenance automatically.
  return new ZooKeeperCommandExecutor(zkCfg, cfg.dataDir(),
                    new StandaloneCommandExecutor(pm,
                                   repositoryWorker,
                                   sessionManager,
                                   null, null),
                    onTakeLeadership, onReleaseLeadership);
}

代码示例来源：origin: line/centraldogma

private static String absolutePath(String... pathElements) {
  if (pathElements.length == 0) {
    return PATH_PREFIX;
  }
  return path(PATH_PREFIX, path(pathElements));
}

代码示例来源：origin: com.linecorp.centraldogma/centraldogma-server

@VisibleForTesting
Optional<ReplicationLog<?>> loadLog(long revision, boolean skipIfSameReplica) {
  try {
    createParentNodes();
    final String logPath = absolutePath(LOG_PATH) + '/' + pathFromRevision(revision);
    final LogMeta logMeta = Jackson.readValue(curator.getData().forPath(logPath), LogMeta.class);
    if (skipIfSameReplica && replicaId() == logMeta.replicaId()) {
      return Optional.empty();
    }
    final byte[] bytes = new byte[logMeta.size()];
    int offset = 0;
    for (long blockId : logMeta.blocks()) {
      final String blockPath = absolutePath(LOG_BLOCK_PATH) + '/' + pathFromRevision(blockId);
      final byte[] b = curator.getData().forPath(blockPath);
      System.arraycopy(b, 0, bytes, offset, b.length);
      offset += b.length;
    }
    assert logMeta.size() == offset;
    final ReplicationLog<?> log = Jackson.readValue(bytes, ReplicationLog.class);
    return Optional.of(log);
  } catch (Exception e) {
    logger.error("Failed to load a log at revision {}; entering read-only mode", revision, e);
    stopLater();
    throw new ReplicationException("failed to load a log at revision " + revision, e);
  }
}

代码示例来源：origin: com.linecorp.centraldogma/centraldogma-server

for (;;) {
    final long nextRevision = info.lastReplayedRevision + 1;
    final Optional<ReplicationLog<?>> log = loadLog(nextRevision, true);
    if (log.isPresent()) {
      final ReplicationLog<?> l = log.get();
  logger.error("Failed to replay a log at revision {}; entering read-only mode",
         info.lastReplayedRevision, e);
  stopLater();
  updateLastReplayedRevision(targetRevision);
} catch (Exception e) {
  logger.error("Failed to update {} to {}; entering read-only mode",
         revisionFile, targetRevision, e);
  stopLater();
  throw new ReplicationException("failed to update " + revisionFile + " to " + targetRevision, e);

代码示例来源：origin: line/centraldogma

private long storeLog(ReplicationLog<?> log) {
  try {
    final byte[] bytes = Jackson.writeValueAsBytes(log);
    assert bytes.length > 0;
    final LogMeta logMeta = new LogMeta(log.replicaId(), System.currentTimeMillis(), bytes.length);
    final int count = (bytes.length + MAX_BYTES - 1) / MAX_BYTES;
    for (int i = 0; i < count; ++i) {
      final int start = i * MAX_BYTES;
      final int end = Math.min((i + 1) * MAX_BYTES, bytes.length);
      final byte[] b = Arrays.copyOfRange(bytes, start, end);
      final String blockPath = curator.create()
                      .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                      .forPath(absolutePath(LOG_BLOCK_PATH) + '/', b);
      final long blockId = revisionFromPath(blockPath);
      logMeta.appendBlock(blockId);
    }
    final String logPath =
        curator.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL)
            .forPath(absolutePath(LOG_PATH) + '/', Jackson.writeValueAsBytes(logMeta));
    return revisionFromPath(logPath);
  } catch (Exception e) {
    logger.error("Failed to store a log; entering read-only mode: {}", log, e);
    stopLater();
    throw new ReplicationException("failed to store a log: " + log, e);
  }
}

代码示例来源：origin: com.linecorp.centraldogma/centraldogma-server-shaded

lastReplayedRevision = getLastReplayedRevision();
  listenerInfo = new ListenerInfo(lastReplayedRevision, onTakeLeadership, onReleaseLeadership);
} catch (Exception e) {
quorumPeer = startZooKeeper();
retryPolicy = RETRY_POLICY_ALWAYS;
logWatcher = new PathChildrenCache(curator, absolutePath(LOG_PATH),
                  true, false, logWatcherExecutor);
logWatcher.getListenable().addListener(this, MoreExecutors.directExecutor());
leaderSelectorExecutor = Executors.newSingleThreadExecutor(
    new DefaultThreadFactory("zookeeper-leader-selector", true));
leaderSelector = new LeaderSelector(curator, absolutePath(LEADER_PATH),
                  leaderSelectorExecutor, oldLogRemover);
leaderSelector.start();

代码示例来源：origin: com.linecorp.centraldogma/centraldogma-server

private SafeLock safeLock(String executionPath) {
  final InterProcessMutex mtx = mutexMap.computeIfAbsent(
      executionPath, k -> new InterProcessMutex(curator, absolutePath(LOCK_PATH, executionPath)));
  try {
    mtx.acquire();
  } catch (Exception e) {
    logger.error("Failed to acquire a lock for {}; entering read-only mode", executionPath, e);
    stopLater();
    throw new ReplicationException("failed to acquire a lock for " + executionPath, e);
  }
  return () -> {
    try {
      mtx.release();
    } catch (Exception ignored) {
      // Ignore.
    }
  };
}

代码示例来源：origin: com.linecorp.centraldogma/centraldogma-server-shaded

copyZkProperty(zkProps, "initLimit", "5");
copyZkProperty(zkProps, "syncLimit", "10");
copyZkProperty(zkProps, "tickTime", "3000");
copyZkProperty(zkProps, "syncEnabled", "true");
copyZkProperty(zkProps, "autopurge.snapRetainCount", "7");
copyZkProperty(zkProps, "autopurge.purgeInterval", "24");
  if (isStopping()) {
    throw new InterruptedException("Stop requested before joining the cluster");