@Test void testRace() throws Exception { final List<Replica> replicas = buildCluster(NUM_REPLICAS, true , () -> { final AtomicInteger counter = new AtomicInteger(); return command -> completedFuture(new Revision(counter.incrementAndGet())); }); try { final Command<Revision> command = Command.push(Author.SYSTEM, "foo", "bar", Revision.HEAD, "", "", Markup.PLAINTEXT); final int COMMANDS_PER_REPLICA = 7; final List<CompletableFuture<Void>> futures = new ArrayList<>(); for (final Replica r : replicas) { futures.add(CompletableFuture.runAsync(() -> { for (int j = 0; j < COMMANDS_PER_REPLICA; j++) { try { r.rm.execute(command).join(); } catch (Exception e) { throw new Error(e); } } })); } for (CompletableFuture<Void> f : futures) { f.get(); } for (Replica r : replicas) { for (int i = 0; i < COMMANDS_PER_REPLICA * replicas.size(); i++) { @SuppressWarnings("unchecked") final ReplicationLog<Revision> log = (ReplicationLog<Revision>) r.rm.loadLog(i, false).get(); assertThat(log.result().major()).isEqualTo(i + 1); } } } finally { replicas.forEach(r -> r.rm.stop()); } }
@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); } }
ZooKeeperCommandExecutor extends AbstractCommandExecutor implements PathChildrenCacheListener { @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); } } }
ZooKeeperCommandExecutor extends AbstractCommandExecutor implements PathChildrenCacheListener { @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); } } ZooKeeperCommandExecutor(ZooKeeperReplicationConfig cfg,
                                    File dataDir, CommandExecutor delegate,
                                    MeterRegistry meterRegistry,
                                    @Nullable Consumer<CommandExecutor> onTakeLeadership,
                                    @Nullable Consumer<CommandExecutor> onReleaseLeadership); }
ZooKeeperCommandExecutor extends AbstractCommandExecutor implements PathChildrenCacheListener { @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); } } ZooKeeperCommandExecutor(ZooKeeperReplicationConfig cfg,
                                    File dataDir, CommandExecutor delegate,
                                    MeterRegistry meterRegistry,
                                    @Nullable Consumer<CommandExecutor> onTakeLeadership,
                                    @Nullable Consumer<CommandExecutor> onReleaseLeadership); @Override int replicaId(); @Override void childEvent(CuratorFramework unused, PathChildrenCacheEvent event); }
ZooKeeperCommandExecutor extends AbstractCommandExecutor implements PathChildrenCacheListener { @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); } } ZooKeeperCommandExecutor(ZooKeeperReplicationConfig cfg,
                                    File dataDir, CommandExecutor delegate,
                                    MeterRegistry meterRegistry,
                                    @Nullable Consumer<CommandExecutor> onTakeLeadership,
                                    @Nullable Consumer<CommandExecutor> onReleaseLeadership); @Override int replicaId(); @Override void childEvent(CuratorFramework unused, PathChildrenCacheEvent event);  }