@Test public void testExceedMaxRound() throws ExecutionException, InterruptedException { WaitableNewNodeCatchUpTaskContext taskContext = new WaitableNewNodeCatchUpTaskContext(); NodeConfig config = new NodeConfig(); config.setNewNodeMaxRound(2); NewNodeCatchUpTask task = new NewNodeCatchUpTask( taskContext, new NodeEndpoint("D", "localhost", 2336), config ); Future<NewNodeCatchUpTaskResult> future = taskExecutor.submit(task); taskContext.awaitReplicateLog(); task.onReceiveAppendEntriesResult( new AppendEntriesResultMessage( new AppendEntriesResult("", 1, true), NodeId.of("D"), createAppendEntriesRpc(1) ), 10 ); task.onReceiveAppendEntriesResult( new AppendEntriesResultMessage( new AppendEntriesResult("", 1, true), NodeId.of("D"), createAppendEntriesRpc(2) ), 10 ); NewNodeCatchUpTaskResult result = future.get(); Assert.assertEquals(NewNodeCatchUpTaskResult.State.TIMEOUT, result.getState()); }
synchronized void onReceiveAppendEntriesResult(AppendEntriesResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } if (nextIndex == 0) { nextIndex = nextLogIndex; } logger.debug("replication state of new node {}, next index {}, match index {}", nodeId, nextIndex, matchIndex); if (resultMessage.get().isSuccess()) { int lastEntryIndex = resultMessage.getRpc().getLastEntryIndex(); assert lastEntryIndex >= 0; matchIndex = lastEntryIndex; nextIndex = lastEntryIndex + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } if ((++round) > config.getNewNodeMaxRound()) { logger.info("node {} cannot catch up within max round", nodeId); setStateAndNotify(State.TIMEOUT); return; } } else { if (nextIndex <= 1) { logger.warn("node {} cannot back off next index more, stop replication", nodeId); setStateAndNotify(State.REPLICATION_FAILED); return; } nextIndex--; if (System.currentTimeMillis() - lastAdvanceAt >= config.getNewNodeAdvanceTimeout()) { logger.debug("node {} cannot make progress within timeout", nodeId); setStateAndNotify(State.TIMEOUT); return; } } context.doReplicateLog(endpoint, nextIndex); lastReplicateAt = System.currentTimeMillis(); notify(); }
NewNodeCatchUpTask implements Callable<NewNodeCatchUpTaskResult> { synchronized void onReceiveAppendEntriesResult(AppendEntriesResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } if (nextIndex == 0) { nextIndex = nextLogIndex; } logger.debug("replication state of new node {}, next index {}, match index {}", nodeId, nextIndex, matchIndex); if (resultMessage.get().isSuccess()) { int lastEntryIndex = resultMessage.getRpc().getLastEntryIndex(); assert lastEntryIndex >= 0; matchIndex = lastEntryIndex; nextIndex = lastEntryIndex + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } if ((++round) > config.getNewNodeMaxRound()) { logger.info("node {} cannot catch up within max round", nodeId); setStateAndNotify(State.TIMEOUT); return; } } else { if (nextIndex <= 1) { logger.warn("node {} cannot back off next index more, stop replication", nodeId); setStateAndNotify(State.REPLICATION_FAILED); return; } nextIndex--; if (System.currentTimeMillis() - lastAdvanceAt >= config.getNewNodeAdvanceTimeout()) { logger.debug("node {} cannot make progress within timeout", nodeId); setStateAndNotify(State.TIMEOUT); return; } } context.doReplicateLog(endpoint, nextIndex); lastReplicateAt = System.currentTimeMillis(); notify(); } }
NewNodeCatchUpTask implements Callable<NewNodeCatchUpTaskResult> { synchronized void onReceiveAppendEntriesResult(AppendEntriesResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } if (nextIndex == 0) { nextIndex = nextLogIndex; } logger.debug("replication state of new node {}, next index {}, match index {}", nodeId, nextIndex, matchIndex); if (resultMessage.get().isSuccess()) { int lastEntryIndex = resultMessage.getRpc().getLastEntryIndex(); assert lastEntryIndex >= 0; matchIndex = lastEntryIndex; nextIndex = lastEntryIndex + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } if ((++round) > config.getNewNodeMaxRound()) { logger.info("node {} cannot catch up within max round", nodeId); setStateAndNotify(State.TIMEOUT); return; } } else { if (nextIndex <= 1) { logger.warn("node {} cannot back off next index more, stop replication", nodeId); setStateAndNotify(State.REPLICATION_FAILED); return; } nextIndex--; if (System.currentTimeMillis() - lastAdvanceAt >= config.getNewNodeAdvanceTimeout()) { logger.debug("node {} cannot make progress within timeout", nodeId); setStateAndNotify(State.TIMEOUT); return; } } context.doReplicateLog(endpoint, nextIndex); lastReplicateAt = System.currentTimeMillis(); notify(); } NewNodeCatchUpTask(NewNodeCatchUpTaskContext context, NodeEndpoint endpoint, NodeConfig config); }
NewNodeCatchUpTask implements Callable<NewNodeCatchUpTaskResult> { synchronized void onReceiveAppendEntriesResult(AppendEntriesResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } if (nextIndex == 0) { nextIndex = nextLogIndex; } logger.debug("replication state of new node {}, next index {}, match index {}", nodeId, nextIndex, matchIndex); if (resultMessage.get().isSuccess()) { int lastEntryIndex = resultMessage.getRpc().getLastEntryIndex(); assert lastEntryIndex >= 0; matchIndex = lastEntryIndex; nextIndex = lastEntryIndex + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } if ((++round) > config.getNewNodeMaxRound()) { logger.info("node {} cannot catch up within max round", nodeId); setStateAndNotify(State.TIMEOUT); return; } } else { if (nextIndex <= 1) { logger.warn("node {} cannot back off next index more, stop replication", nodeId); setStateAndNotify(State.REPLICATION_FAILED); return; } nextIndex--; if (System.currentTimeMillis() - lastAdvanceAt >= config.getNewNodeAdvanceTimeout()) { logger.debug("node {} cannot make progress within timeout", nodeId); setStateAndNotify(State.TIMEOUT); return; } } context.doReplicateLog(endpoint, nextIndex); 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 onReceiveAppendEntriesResult(AppendEntriesResultMessage resultMessage, int nextLogIndex) { assert nodeId.equals(resultMessage.getSourceNodeId()); if (state != State.REPLICATING) { throw new IllegalStateException("receive append entries result when state is not replicating"); } if (nextIndex == 0) { nextIndex = nextLogIndex; } logger.debug("replication state of new node {}, next index {}, match index {}", nodeId, nextIndex, matchIndex); if (resultMessage.get().isSuccess()) { int lastEntryIndex = resultMessage.getRpc().getLastEntryIndex(); assert lastEntryIndex >= 0; matchIndex = lastEntryIndex; nextIndex = lastEntryIndex + 1; lastAdvanceAt = System.currentTimeMillis(); if (nextIndex >= nextLogIndex) { setStateAndNotify(State.REPLICATION_CATCH_UP); return; } if ((++round) > config.getNewNodeMaxRound()) { logger.info("node {} cannot catch up within max round", nodeId); setStateAndNotify(State.TIMEOUT); return; } } else { if (nextIndex <= 1) { logger.warn("node {} cannot back off next index more, stop replication", nodeId); setStateAndNotify(State.REPLICATION_FAILED); return; } nextIndex--; if (System.currentTimeMillis() - lastAdvanceAt >= config.getNewNodeAdvanceTimeout()) { logger.debug("node {} cannot make progress within timeout", nodeId); setStateAndNotify(State.TIMEOUT); return; } } context.doReplicateLog(endpoint, nextIndex); lastReplicateAt = System.currentTimeMillis(); notify(); } NewNodeCatchUpTask(NewNodeCatchUpTaskContext context, NodeEndpoint endpoint, NodeConfig config); NodeId getNodeId(); @Override synchronized NewNodeCatchUpTaskResult call(); @Override String toString();  }