@Test public void testInstallSnapshotCatchUp() throws InterruptedException, ExecutionException { WaitableNewNodeCatchUpTaskContext taskContext = new WaitableNewNodeCatchUpTaskContext(); NewNodeCatchUpTask task = new NewNodeCatchUpTask( taskContext, new NodeEndpoint("D", "localhost", 2336), new NodeConfig() ); Future<NewNodeCatchUpTaskResult> future = taskExecutor.submit(task); taskContext.awaitReplicateLog(); InstallSnapshotRpc rpc = new InstallSnapshotRpc(); rpc.setLastIndex(2); rpc.setData(new byte[0]); rpc.setDone(true); task.onReceiveInstallSnapshotResult(new InstallSnapshotResultMessage( new InstallSnapshotResult(1), NodeId.of("D"), rpc ), 3); NewNodeCatchUpTaskResult result = future.get(); Assert.assertEquals(NewNodeCatchUpTaskResult.State.OK, result.getState()); Assert.assertEquals(2, result.getMatchIndex()); Assert.assertEquals(3, result.getNextIndex()); }
synchronized void onReceiveInstallSnapshotResult(InstallSnapshotResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } InstallSnapshotRpc rpc = resultMessage.getRpc(); if (rpc.isDone()) { matchIndex = rpc.getLastIndex(); nextIndex = rpc.getLastIndex() + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } round++; context.doReplicateLog(endpoint, nextIndex); } else { context.sendInstallSnapshot(endpoint, rpc.getOffset() + rpc.getDataLength()); } lastReplicateAt = System.currentTimeMillis(); notify(); }
NewNodeCatchUpTask implements Callable<NewNodeCatchUpTaskResult> { synchronized void onReceiveInstallSnapshotResult(InstallSnapshotResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } InstallSnapshotRpc rpc = resultMessage.getRpc(); if (rpc.isDone()) { matchIndex = rpc.getLastIndex(); nextIndex = rpc.getLastIndex() + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } round++; context.doReplicateLog(endpoint, nextIndex); } else { context.sendInstallSnapshot(endpoint, rpc.getOffset() + rpc.getDataLength()); } lastReplicateAt = System.currentTimeMillis(); notify(); } }
NewNodeCatchUpTask implements Callable<NewNodeCatchUpTaskResult> { synchronized void onReceiveInstallSnapshotResult(InstallSnapshotResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } InstallSnapshotRpc rpc = resultMessage.getRpc(); if (rpc.isDone()) { matchIndex = rpc.getLastIndex(); nextIndex = rpc.getLastIndex() + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } round++; context.doReplicateLog(endpoint, nextIndex); } else { context.sendInstallSnapshot(endpoint, rpc.getOffset() + rpc.getDataLength()); } lastReplicateAt = System.currentTimeMillis(); notify(); } NewNodeCatchUpTask(NewNodeCatchUpTaskContext context, NodeEndpoint endpoint, NodeConfig config); }
NewNodeCatchUpTask implements Callable<NewNodeCatchUpTaskResult> { synchronized void onReceiveInstallSnapshotResult(InstallSnapshotResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } InstallSnapshotRpc rpc = resultMessage.getRpc(); if (rpc.isDone()) { matchIndex = rpc.getLastIndex(); nextIndex = rpc.getLastIndex() + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } round++; context.doReplicateLog(endpoint, nextIndex); } else { context.sendInstallSnapshot(endpoint, rpc.getOffset() + rpc.getDataLength()); } lastReplicateAt = System.currentTimeMillis(); notify(); } NewNodeCatchUpTask(NewNodeCatchUpTaskContext context, NodeEndpoint endpoint, NodeConfig config); NodeId getNodeId(); @Override synchronized NewNodeCatchUpTaskResult call(); @Override String toString(); }
NewNodeCatchUpTask implements Callable<NewNodeCatchUpTaskResult> { synchronized void onReceiveInstallSnapshotResult(InstallSnapshotResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } InstallSnapshotRpc rpc = resultMessage.getRpc(); if (rpc.isDone()) { matchIndex = rpc.getLastIndex(); nextIndex = rpc.getLastIndex() + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } round++; context.doReplicateLog(endpoint, nextIndex); } else { context.sendInstallSnapshot(endpoint, rpc.getOffset() + rpc.getDataLength()); } lastReplicateAt = System.currentTimeMillis(); notify(); } NewNodeCatchUpTask(NewNodeCatchUpTaskContext context, NodeEndpoint endpoint, NodeConfig config); NodeId getNodeId(); @Override synchronized NewNodeCatchUpTaskResult call(); @Override String toString();  }