@Test void testLogWatch() throws Exception { final List<Replica> cluster = buildCluster(5, true , ZooKeeperCommandExecutorTest::newMockDelegate); final Replica replica1 = cluster.get(0); final Replica replica2 = cluster.get(1); final Replica replica3 = cluster.get(2); final Replica replica4 = cluster.get(3); replica4.rm.stop().join(); try { final Command<Void> command1 = Command.createRepository(Author.SYSTEM, "project", "repo1"); replica1.rm.execute(command1).join(); final Optional<ReplicationLog<?>> commandResult2 = replica1.rm.loadLog(0, false); assertThat(commandResult2).isPresent(); assertThat(commandResult2.get().command()).isEqualTo(command1); assertThat(commandResult2.get().result()).isNull(); await().untilAsserted(() -> verify(replica1.delegate).apply(eq(command1))); await().untilAsserted(() -> verify(replica2.delegate).apply(eq(command1))); await().untilAsserted(() -> verify(replica3.delegate).apply(eq(command1))); await().until(replica1::existsLocalRevision); await().until(replica2::existsLocalRevision); await().until(replica3::existsLocalRevision); assertThat(replica1.localRevision()).isEqualTo(0L); assertThat(replica2.localRevision()).isEqualTo(0L); assertThat(replica3.localRevision()).isEqualTo(0L); replica3.rm.stop().join(); final Command<?> command2 = Command.createProject(Author.SYSTEM, "foo"); replica1.rm.execute(command2).join(); await().untilAsserted(() -> verify(replica1.delegate).apply(eq(command2))); await().untilAsserted(() -> verify(replica2.delegate).apply(eq(command2))); await().untilAsserted(() -> verify(replica3.delegate, times(0)).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(); } } }
@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);  }