### Question:
Replicator implements ThreadId.OnError { @SuppressWarnings("unused") static void onTimeoutNowReturned(final ThreadId id, final Status status, final TimeoutNowRequest request, final TimeoutNowResponse response, final boolean stopAfterFinish) { final Replicator r = (Replicator) id.lock(); if (r == null) { return; } final boolean isLogDebugEnabled = LOG.isDebugEnabled(); StringBuilder sb = null; if (isLogDebugEnabled) { sb = new StringBuilder("Node "). append(r.options.getGroupId()).append(":").append(r.options.getServerId()). append(" received TimeoutNowResponse from "). append(r.options.getPeerId()); } if (!status.isOk()) { if (isLogDebugEnabled) { sb.append(" fail:").append(status); LOG.debug(sb.toString()); } notifyReplicatorStatusListener(r, ReplicatorEvent.ERROR, status); if (stopAfterFinish) { r.notifyOnCaughtUp(RaftError.ESTOP.getNumber(), true); r.destroy(); } else { id.unlock(); } return; } if (isLogDebugEnabled) { sb.append(response.getSuccess() ? " success" : " fail"); LOG.debug(sb.toString()); } if (response.getTerm() > r.options.getTerm()) { final NodeImpl node = r.options.getNode(); r.notifyOnCaughtUp(RaftError.EPERM.getNumber(), true); r.destroy(); node.increaseTermTo(response.getTerm(), new Status(RaftError.EHIGHERTERMRESPONSE, "Leader receives higher term timeout_now_response from peer:%s", r.options.getPeerId())); return; } if (stopAfterFinish) { r.notifyOnCaughtUp(RaftError.ESTOP.getNumber(), true); r.destroy(); } else { id.unlock(); } } Replicator(final ReplicatorOptions replicatorOptions, final RaftOptions raftOptions); @OnlyForTest PriorityQueue<RpcResponse> getPendingResponses(); static ThreadId start(final ReplicatorOptions opts, final RaftOptions raftOptions); static void waitForCaughtUp(final ThreadId id, final long maxMargin, final long dueTime,
                                       final CatchUpClosure done); @Override String toString(); @Override void onError(final ThreadId id, final Object data, final int errorCode); static void sendHeartbeat(final ThreadId id, final RpcResponseClosure<AppendEntriesResponse> closure); static boolean stop(final ThreadId id); static boolean join(final ThreadId id); static long getLastRpcSendTimestamp(final ThreadId id); static boolean transferLeadership(final ThreadId id, final long logIndex); static boolean stopTransferLeadership(final ThreadId id); static boolean sendTimeoutNowAndStop(final ThreadId id, final int timeoutMs); static long getNextIndex(final ThreadId id);  }### Answer:
@Test public void testOnTimeoutNowReturnedRpcErrorAndStop() { final Replicator r = getReplicator(); final RpcRequests.TimeoutNowRequest request = createTimeoutnowRequest(); this.id.unlock(); Replicator.onTimeoutNowReturned(this.id, new Status(-1, "test"), request, null, true); assertNull(r.id); }
@Test public void testOnTimeoutNowReturnedTermMismatch() { final Replicator r = getReplicator(); this.id.unlock(); final RpcRequests.TimeoutNowRequest request = createTimeoutnowRequest(); final RpcRequests.TimeoutNowResponse response = RpcRequests.TimeoutNowResponse.newBuilder() .setSuccess(false) .setTerm(12) .build(); this.id.unlock(); Replicator.onTimeoutNowReturned(this.id, Status.OK(), request, response, false); Mockito.verify(this.node).increaseTermTo( 12, new Status(RaftError.EHIGHERTERMRESPONSE, "Leader receives higher term timeout_now_response from peer:%s", this.peerId)); assertNull(r.id); }