diff --git a/CHANGES.txt b/CHANGES.txt index 8611719d56..f288f3206e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0-alpha2 + * Add retries to IR messages (CASSANDRA-18962) * Add metrics and logging to repair retries (CASSANDRA-18952) * Remove deprecated code in Cassandra 1.x and 2.x (CASSANDRA-18959) * ClientRequestSize metrics should not treat CONTAINS restrictions as being equality-based (CASSANDRA-18896) diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java index cf0f2e30a3..c85f0ddeca 100644 --- a/src/java/org/apache/cassandra/net/Verb.java +++ b/src/java/org/apache/cassandra/net/Verb.java @@ -169,12 +169,12 @@ public enum Verb PREPARE_MSG (105, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> PrepareMessage.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), SNAPSHOT_MSG (106, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> SnapshotMessage.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), CLEANUP_MSG (107, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> CleanupMessage.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), - PREPARE_CONSISTENT_RSP (109, P1, repairTimeout, ANTI_ENTROPY, () -> PrepareConsistentResponse.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), - PREPARE_CONSISTENT_REQ (108, P1, repairTimeout, ANTI_ENTROPY, () -> PrepareConsistentRequest.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), - FINALIZE_PROPOSE_MSG (110, P1, repairTimeout, ANTI_ENTROPY, () -> FinalizePropose.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), - FINALIZE_PROMISE_MSG (111, P1, repairTimeout, ANTI_ENTROPY, () -> FinalizePromise.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), - FINALIZE_COMMIT_MSG (112, P1, repairTimeout, ANTI_ENTROPY, () -> FinalizeCommit.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), - FAILED_SESSION_MSG (113, P1, repairTimeout, ANTI_ENTROPY, () -> FailSession.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), + PREPARE_CONSISTENT_RSP (109, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> PrepareConsistentResponse.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), + PREPARE_CONSISTENT_REQ (108, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> PrepareConsistentRequest.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), + FINALIZE_PROPOSE_MSG (110, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> FinalizePropose.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), + FINALIZE_PROMISE_MSG (111, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> FinalizePromise.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), + FINALIZE_COMMIT_MSG (112, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> FinalizeCommit.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), + FAILED_SESSION_MSG (113, P1, repairWithBackoffTimeout, ANTI_ENTROPY, () -> FailSession.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), STATUS_RSP (115, P1, repairTimeout, ANTI_ENTROPY, () -> StatusResponse.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), STATUS_REQ (114, P1, repairTimeout, ANTI_ENTROPY, () -> StatusRequest.serializer, () -> RepairMessageVerbHandler.instance(), REPAIR_RSP ), diff --git a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java index 988159860d..34f20cd708 100644 --- a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java +++ b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java @@ -25,7 +25,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.exceptions.RequestFailureReason; import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; import org.apache.cassandra.repair.messages.*; @@ -286,27 +285,28 @@ public class RepairMessageVerbHandler implements IVerbHandler break; case PREPARE_CONSISTENT_REQ: - ctx.repair().consistent.local.handlePrepareMessage(message.from(), (PrepareConsistentRequest) message.payload); + ctx.repair().consistent.local.handlePrepareMessage(message); break; case PREPARE_CONSISTENT_RSP: - ctx.repair().consistent.coordinated.handlePrepareResponse((PrepareConsistentResponse) message.payload); + ctx.repair().consistent.coordinated.handlePrepareResponse(message); break; case FINALIZE_PROPOSE_MSG: - ctx.repair().consistent.local.handleFinalizeProposeMessage(message.from(), (FinalizePropose) message.payload); + ctx.repair().consistent.local.handleFinalizeProposeMessage(message); break; case FINALIZE_PROMISE_MSG: - ctx.repair().consistent.coordinated.handleFinalizePromiseMessage((FinalizePromise) message.payload); + ctx.repair().consistent.coordinated.handleFinalizePromiseMessage(message); break; case FINALIZE_COMMIT_MSG: - ctx.repair().consistent.local.handleFinalizeCommitMessage(message.from(), (FinalizeCommit) message.payload); + ctx.repair().consistent.local.handleFinalizeCommitMessage(message); break; case FAILED_SESSION_MSG: FailSession failure = (FailSession) message.payload; + sendAck(message); ctx.repair().consistent.coordinated.handleFailSessionMessage(failure); ctx.repair().consistent.local.handleFailSessionMessage(message.from(), failure); break; @@ -410,12 +410,11 @@ public class RepairMessageVerbHandler implements IVerbHandler private void sendFailureResponse(Message respondTo) { - Message reply = respondTo.failureResponse(RequestFailureReason.UNKNOWN); - ctx.messaging().send(reply, respondTo.from()); + RepairMessage.sendFailureResponse(ctx, respondTo); } private void sendAck(Message message) { - ctx.messaging().send(message.emptyResponse(), message.from()); + RepairMessage.sendAck(ctx, message); } } diff --git a/src/java/org/apache/cassandra/repair/SharedContext.java b/src/java/org/apache/cassandra/repair/SharedContext.java index defc59a83c..8ccc88f584 100644 --- a/src/java/org/apache/cassandra/repair/SharedContext.java +++ b/src/java/org/apache/cassandra/repair/SharedContext.java @@ -59,6 +59,16 @@ public interface SharedContext ScheduledExecutorPlus optionalTasks(); MessageDelivery messaging(); + default SharedContext withMessaging(MessageDelivery messaging) + { + return new ForwardingSharedContext(this) { + @Override + public MessageDelivery messaging() + { + return messaging; + } + }; + } IFailureDetector failureDetector(); IEndpointSnitch snitch(); IGossiper gossiper(); @@ -162,4 +172,109 @@ public interface SharedContext return StreamPlan::execute; } } + + class ForwardingSharedContext implements SharedContext + { + private final SharedContext delegate; + + public ForwardingSharedContext(SharedContext delegate) + { + this.delegate = delegate; + } + + protected SharedContext delegate() + { + return delegate; + } + + @Override + public InetAddressAndPort broadcastAddressAndPort() + { + return delegate().broadcastAddressAndPort(); + } + + @Override + public Supplier random() + { + return delegate().random(); + } + + @Override + public Clock clock() + { + return delegate().clock(); + } + + @Override + public ExecutorFactory executorFactory() + { + return delegate().executorFactory(); + } + + @Override + public MBeanWrapper mbean() + { + return delegate().mbean(); + } + + @Override + public ScheduledExecutorPlus optionalTasks() + { + return delegate().optionalTasks(); + } + + @Override + public MessageDelivery messaging() + { + return delegate().messaging(); + } + + @Override + public IFailureDetector failureDetector() + { + return delegate().failureDetector(); + } + + @Override + public IEndpointSnitch snitch() + { + return delegate().snitch(); + } + + @Override + public IGossiper gossiper() + { + return delegate().gossiper(); + } + + @Override + public ICompactionManager compactionManager() + { + return delegate().compactionManager(); + } + + @Override + public ActiveRepairService repair() + { + return delegate().repair(); + } + + @Override + public IValidationManager validationManager() + { + return delegate().validationManager(); + } + + @Override + public TableRepairManager repairManager(ColumnFamilyStore store) + { + return delegate().repairManager(store); + } + + @Override + public StreamExecutor streamExecutor() + { + return delegate().streamExecutor(); + } + } } diff --git a/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java b/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java index 76645f1adc..8101f02912 100644 --- a/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java +++ b/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java @@ -70,7 +70,7 @@ import org.apache.cassandra.utils.TimeUUID; * {@code PREPARED}, and a {@link PrepareConsistentResponse} is sent to the coordinator indicating success or failure. * If the pending anti-compaction fails, the local session state is set to {@code FAILED}. *

- * (see {@link LocalSessions#handlePrepareMessage(InetAddressAndPort, PrepareConsistentRequest)} + * (see {@link LocalSessions#handlePrepareMessage(org.apache.cassandra.net.Message)} *

* Once the coordinator recieves positive {@code PrepareConsistentResponse} messages from all the participants, the * coordinator begins the normal repair process. diff --git a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java index 0ca1576df1..6771dfe07d 100644 --- a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java +++ b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java @@ -31,6 +31,8 @@ import com.google.common.collect.Iterables; import org.apache.cassandra.concurrent.ImmediateExecutor; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.repair.CoordinatedRepairResult; +import org.apache.cassandra.repair.messages.FinalizePromise; +import org.apache.cassandra.repair.messages.PrepareConsistentResponse; import org.apache.cassandra.utils.concurrent.AsyncPromise; import org.apache.cassandra.utils.concurrent.Future; @@ -47,10 +49,14 @@ import org.apache.cassandra.repair.messages.FailSession; import org.apache.cassandra.repair.messages.FinalizeCommit; import org.apache.cassandra.repair.messages.FinalizePropose; import org.apache.cassandra.repair.messages.PrepareConsistentRequest; -import org.apache.cassandra.repair.messages.RepairMessage; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.utils.concurrent.ImmediateFuture; +import static org.apache.cassandra.repair.messages.RepairMessage.notDone; +import static org.apache.cassandra.repair.messages.RepairMessage.sendAck; +import static org.apache.cassandra.repair.messages.RepairMessage.sendFailureResponse; +import static org.apache.cassandra.repair.messages.RepairMessage.sendMessageWithRetries; + /** * Coordinator side logic and state of a consistent repair session. Like {@link ActiveRepairService.ParentRepairSession}, * there is only one {@code CoordinatorSession} per user repair command, regardless of the number of tables and token @@ -154,34 +160,37 @@ public class CoordinatorSession extends ConsistentSession return getState() == State.FAILED || Iterables.any(participantStates.values(), v -> v == State.FAILED); } - protected void sendMessage(InetAddressAndPort destination, Message message) - { - logger.trace("Sending {} to {}", message.payload, destination); - - ctx.messaging().send(message, destination); - } - public Future prepare() { Preconditions.checkArgument(allStates(State.PREPARING)); logger.info("Beginning prepare phase of incremental repair session {}", sessionID); - Message message = - Message.out(Verb.PREPARE_CONSISTENT_REQ, new PrepareConsistentRequest(sessionID, coordinator, participants)); + + PrepareConsistentRequest request = new PrepareConsistentRequest(sessionID, coordinator, participants); for (final InetAddressAndPort participant : participants) { - sendMessage(participant, message); + sendMessageWithRetries(ctx, + notDone(prepareFuture), + request, + Verb.PREPARE_CONSISTENT_REQ, + participant); } return prepareFuture; } - public synchronized void handlePrepareResponse(InetAddressAndPort participant, boolean success) + public synchronized void handlePrepareResponse(Message msg) { + InetAddressAndPort participant = msg.payload.participant; + boolean success = msg.payload.success; if (getState() == State.FAILED) { logger.trace("Incremental repair {} has failed, ignoring prepare response from {}", sessionID, participant); + sendFailureResponse(ctx, msg); return; } + sendAck(ctx, msg); + if (getParticipantState(participant) != State.PREPARING) + return; if (!success) { logger.warn("{} failed the prepare phase for incremental repair session {}", participant, sessionID); @@ -218,19 +227,29 @@ public class CoordinatorSession extends ConsistentSession { Preconditions.checkArgument(allStates(State.REPAIRING)); logger.info("Proposing finalization of repair session {}", sessionID); - Message message = Message.out(Verb.FINALIZE_PROPOSE_MSG, new FinalizePropose(sessionID)); + FinalizePropose request = new FinalizePropose(sessionID); for (final InetAddressAndPort participant : participants) { - sendMessage(participant, message); + sendMessageWithRetries(ctx, notDone(finalizeProposeFuture), request, Verb.FINALIZE_PROPOSE_MSG, participant); } return finalizeProposeFuture; } - public synchronized void handleFinalizePromise(InetAddressAndPort participant, boolean success) + public synchronized void handleFinalizePromise(Message message) { + InetAddressAndPort participant = message.payload.participant; + boolean success = message.payload.promised; if (getState() == State.FAILED) { logger.trace("Incremental repair {} has failed, ignoring finalize promise from {}", sessionID, participant); + sendFailureResponse(ctx, message); + return; + } + sendAck(ctx, message); + if (getParticipantState(participant) != State.REPAIRING) + { + // this message is a retry, or we failed the session; in either case there is nothing more to do than ack + return; } else if (!success) { @@ -253,10 +272,10 @@ public class CoordinatorSession extends ConsistentSession { Preconditions.checkArgument(allStates(State.FINALIZE_PROMISED)); logger.info("Committing finalization of repair session {}", sessionID); - Message message = Message.out(Verb.FINALIZE_COMMIT_MSG, new FinalizeCommit(sessionID)); + FinalizeCommit payload = new FinalizeCommit(sessionID); for (final InetAddressAndPort participant : participants) { - sendMessage(participant, message); + sendMessageWithRetries(ctx, payload, Verb.FINALIZE_COMMIT_MSG, participant); } setAll(State.FINALIZED); logger.info("Incremental repair session {} completed", sessionID); @@ -264,12 +283,12 @@ public class CoordinatorSession extends ConsistentSession private void sendFailureMessageToParticipants() { - Message message = Message.out(Verb.FAILED_SESSION_MSG, new FailSession(sessionID)); + FailSession payload = new FailSession(sessionID); for (final InetAddressAndPort participant : participants) { if (participantStates.get(participant) != State.FAILED) { - sendMessage(participant, message); + sendMessageWithRetries(ctx, payload, Verb.FAILED_SESSION_MSG, participant); } } } diff --git a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java index 1abab8c476..6f02f033d3 100644 --- a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java +++ b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java @@ -24,15 +24,19 @@ import java.util.Set; import com.google.common.base.Preconditions; +import org.apache.cassandra.net.Message; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.repair.messages.FailSession; import org.apache.cassandra.repair.messages.FinalizePromise; import org.apache.cassandra.repair.messages.PrepareConsistentResponse; +import org.apache.cassandra.repair.messages.RepairMessage; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.repair.NoSuchRepairSessionException; import org.apache.cassandra.utils.TimeUUID; +import static org.apache.cassandra.repair.messages.RepairMessage.sendFailureResponse; + /** * Container for all consistent repair sessions a node is coordinating */ @@ -81,21 +85,31 @@ public class CoordinatorSessions return sessions.get(sessionId); } - public void handlePrepareResponse(PrepareConsistentResponse msg) + public void handlePrepareResponse(Message msg) { - CoordinatorSession session = getSession(msg.parentSession); + PrepareConsistentResponse payload = (PrepareConsistentResponse) msg.payload; + CoordinatorSession session = getSession(payload.parentSession); if (session != null) { - session.handlePrepareResponse(msg.participant, msg.success); + session.handlePrepareResponse((Message) msg); + } + else + { + sendFailureResponse(ctx, msg); } } - public void handleFinalizePromiseMessage(FinalizePromise msg) + public void handleFinalizePromiseMessage(Message message) { + FinalizePromise msg = (FinalizePromise) message.payload; CoordinatorSession session = getSession(msg.sessionID); if (session != null) { - session.handleFinalizePromise(msg.participant, msg.promised); + session.handleFinalizePromise((Message) message); + } + else + { + sendFailureResponse(ctx, message); } } diff --git a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java index 85bdf15923..75c50e5dce 100644 --- a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java +++ b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java @@ -36,6 +36,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.BooleanSupplier; import java.util.function.Predicate; +import java.util.function.Supplier; import java.util.stream.Collectors; import javax.annotation.Nullable; @@ -58,6 +59,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.locator.Replica; +import org.apache.cassandra.net.Verb; import org.apache.cassandra.repair.KeyspaceRepairManager; import org.apache.cassandra.repair.consistent.admin.CleanupSummary; import org.apache.cassandra.repair.consistent.admin.PendingStat; @@ -105,6 +107,9 @@ import static org.apache.cassandra.net.Verb.PREPARE_CONSISTENT_RSP; import static org.apache.cassandra.net.Verb.STATUS_REQ; import static org.apache.cassandra.net.Verb.STATUS_RSP; import static org.apache.cassandra.repair.consistent.ConsistentSession.State.*; +import static org.apache.cassandra.repair.messages.RepairMessage.always; +import static org.apache.cassandra.repair.messages.RepairMessage.sendAck; +import static org.apache.cassandra.repair.messages.RepairMessage.sendFailureResponse; /** * Manages all consistent repair sessions a node is participating in. @@ -335,11 +340,11 @@ public class LocalSessions sessionID, session.coordinator); setStateAndSave(session, FAILED); - Message message = Message.out(FAILED_SESSION_MSG, new FailSession(sessionID)); + FailSession payload = new FailSession(sessionID); for (InetAddressAndPort participant : session.participants) { if (!participant.equals(getBroadcastAddressAndPort())) - sendMessage(participant, message); + sendMessageWithRetries(payload, FAILED_SESSION_MSG, participant); } } @@ -637,10 +642,13 @@ public class LocalSessions } @VisibleForTesting - synchronized void putSessionUnsafe(LocalSession session) + synchronized boolean putSessionUnsafe(LocalSession session) { + if (sessions.containsKey(session.sessionID)) + return false; putSession(session); save(session); + return true; } private synchronized void putSession(LocalSession session) @@ -696,12 +704,19 @@ public class LocalSessions @VisibleForTesting void setStateAndSave(LocalSession session, ConsistentSession.State state) + { + maybeSetStateAndSave(session, null, state); + } + + private boolean maybeSetStateAndSave(LocalSession session, @Nullable ConsistentSession.State expected, ConsistentSession.State state) { synchronized (session) { Preconditions.checkArgument(session.getState().canTransitionTo(state), "Invalid state transition %s -> %s", session.getState(), state); + if (expected != null && session.getState() != expected) + return false; logger.trace("Changing LocalSession state from {} -> {} for {}", session.getState(), state, session.sessionID); boolean wasCompleted = session.isCompleted(); session.setState(state); @@ -714,6 +729,7 @@ public class LocalSessions } for (Listener listener : listeners) listener.onIRStateChange(session); + return true; } } @@ -746,7 +762,7 @@ public class LocalSessions } if (sendMessage) { - sendMessage(session.coordinator, Message.out(FAILED_SESSION_MSG, new FailSession(session.sessionID))); + sendMessageWithRetries(new FailSession(session.sessionID), FAILED_SESSION_MSG, session.coordinator); } } } @@ -802,8 +818,10 @@ public class LocalSessions * successfully. If the data preparation fails, a failure message is sent to the coordinator, * cancelling the session. */ - public void handlePrepareMessage(InetAddressAndPort from, PrepareConsistentRequest request) + public void handlePrepareMessage(Message message) { + InetAddressAndPort from = message.from(); + PrepareConsistentRequest request = (PrepareConsistentRequest) message.payload; logger.trace("received {} from {}", request, from); TimeUUID sessionID = request.parentSession; InetAddressAndPort coordinator = request.coordinator; @@ -817,12 +835,15 @@ public class LocalSessions catch (Throwable e) { logger.error("Error retrieving ParentRepairSession for session {}, responding with failure", sessionID); - sendMessage(coordinator, Message.out(PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(sessionID, getBroadcastAddressAndPort(), false))); + sendFailureResponse(ctx, message); + sendMessageWithRetries(always(), new PrepareConsistentResponse(sessionID, getBroadcastAddressAndPort(), false), PREPARE_CONSISTENT_RSP, coordinator); return; } LocalSession session = createSessionUnsafe(sessionID, parentSession, peers); - putSessionUnsafe(session); + sendAck(ctx, message); + if (!putSessionUnsafe(session)) + return; logger.info("Beginning local incremental repair session {}", session); ExecutorService executor = ctx.executorFactory().pooled("Repair-" + sessionID, parentSession.getColumnFamilyStores().size()); @@ -841,10 +862,7 @@ public class LocalSessions logger.info("Prepare phase for incremental repair session {} completed", sessionID); if (!prepareSessionExceptFailed(session)) logger.info("Session {} failed before anticompaction completed", sessionID); - Message message = - Message.out(PREPARE_CONSISTENT_RSP, - new PrepareConsistentResponse(sessionID, getBroadcastAddressAndPort(), session.getState() != FAILED)); - sendMessage(coordinator, message); + sendMessageWithRetries(always(), new PrepareConsistentResponse(sessionID, getBroadcastAddressAndPort(), session.getState() != FAILED), PREPARE_CONSISTENT_RSP, coordinator); } finally { @@ -862,9 +880,7 @@ public class LocalSessions logger.warn("No such repair session: {}", sessionID); else logger.error("Prepare phase for incremental repair session {} failed", sessionID, t); - sendMessage(coordinator, - Message.out(PREPARE_CONSISTENT_RSP, - new PrepareConsistentResponse(sessionID, getBroadcastAddressAndPort(), false))); + sendMessageWithRetries(always(), new PrepareConsistentResponse(sessionID, getBroadcastAddressAndPort(), false), PREPARE_CONSISTENT_RSP, coordinator); failSession(sessionID, false); } finally @@ -875,6 +891,16 @@ public class LocalSessions }); } + private void sendMessageWithRetries(Supplier allowRetry, RepairMessage request, Verb verb, InetAddressAndPort endpoint) + { + RepairMessage.sendMessageWithRetries(ctx, allowRetry, request, verb, endpoint); + } + + private void sendMessageWithRetries(RepairMessage request, Verb verb, InetAddressAndPort endpoint) + { + RepairMessage.sendMessageWithRetries(ctx, request, verb, endpoint); + } + /** * Checks for the session state, and sets it to prepared unless it is on a failed state. * Making the checks inside a synchronized block to prevent the session state from @@ -904,21 +930,26 @@ public class LocalSessions } } - public void handleFinalizeProposeMessage(InetAddressAndPort from, FinalizePropose propose) + public void handleFinalizeProposeMessage(Message message) { + InetAddressAndPort from = message.from(); + FinalizePropose propose = (FinalizePropose) message.payload; logger.trace("received {} from {}", propose, from); TimeUUID sessionID = propose.sessionID; LocalSession session = getSession(sessionID); if (session == null) { logger.info("Received FinalizePropose message for unknown repair session {}, responding with failure", sessionID); - sendMessage(from, Message.out(FAILED_SESSION_MSG, new FailSession(sessionID))); + sendFailureResponse(ctx, message); + sendMessageWithRetries(new FailSession(sessionID), FAILED_SESSION_MSG, from); return; } + sendAck(ctx, message); try { - setStateAndSave(session, FINALIZE_PROMISED); + if (!maybeSetStateAndSave(session, REPAIRING, FINALIZE_PROMISED)) + return; /* Flushing the repairs table here, *before* responding to the coordinator prevents a scenario where we respond @@ -929,7 +960,7 @@ public class LocalSessions */ syncTable(); - sendMessage(from, Message.out(FINALIZE_PROMISE_MSG, new FinalizePromise(sessionID, getBroadcastAddressAndPort(), true))); + RepairMessage.sendMessageWithRetries(ctx, new FinalizePromise(sessionID, getBroadcastAddressAndPort(), true), FINALIZE_PROMISE_MSG, from); logger.info("Received FinalizePropose message for incremental repair session {}, responded with FinalizePromise", sessionID); } catch (IllegalArgumentException e) @@ -959,19 +990,23 @@ public class LocalSessions * as part of the compaction process, and avoids having to worry about in progress compactions interfering with the * promotion. */ - public void handleFinalizeCommitMessage(InetAddressAndPort from, FinalizeCommit commit) + public void handleFinalizeCommitMessage(Message message) { + InetAddressAndPort from = message.from(); + FinalizeCommit commit = (FinalizeCommit) message.payload; logger.trace("received {} from {}", commit, from); TimeUUID sessionID = commit.sessionID; LocalSession session = getSession(sessionID); if (session == null) { logger.warn("Ignoring FinalizeCommit message for unknown repair session {}", sessionID); + sendFailureResponse(ctx, message); return; } + sendAck(ctx, message); - setStateAndSave(session, FINALIZED); - logger.info("Finalized local repair session {}", sessionID); + if (maybeSetStateAndSave(session, FINALIZE_PROMISED, FINALIZED)) + logger.info("Finalized local repair session {}", sessionID); } public void handleFailSessionMessage(InetAddressAndPort from, FailSession msg) diff --git a/src/java/org/apache/cassandra/repair/messages/FailSession.java b/src/java/org/apache/cassandra/repair/messages/FailSession.java index f6826eec6f..19102fb511 100644 --- a/src/java/org/apache/cassandra/repair/messages/FailSession.java +++ b/src/java/org/apache/cassandra/repair/messages/FailSession.java @@ -36,6 +36,12 @@ public class FailSession extends RepairMessage this.sessionID = sessionID; } + @Override + public TimeUUID parentRepairSession() + { + return sessionID; + } + public boolean equals(Object o) { if (this == o) return true; diff --git a/src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java b/src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java index ca4e6d8a88..d1d9efa601 100644 --- a/src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java +++ b/src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java @@ -36,6 +36,12 @@ public class FinalizeCommit extends RepairMessage this.sessionID = sessionID; } + @Override + public TimeUUID parentRepairSession() + { + return sessionID; + } + public boolean equals(Object o) { if (this == o) return true; diff --git a/src/java/org/apache/cassandra/repair/messages/FinalizePromise.java b/src/java/org/apache/cassandra/repair/messages/FinalizePromise.java index c45f46d48e..ed514d859f 100644 --- a/src/java/org/apache/cassandra/repair/messages/FinalizePromise.java +++ b/src/java/org/apache/cassandra/repair/messages/FinalizePromise.java @@ -45,6 +45,12 @@ public class FinalizePromise extends RepairMessage this.promised = promised; } + @Override + public TimeUUID parentRepairSession() + { + return sessionID; + } + public boolean equals(Object o) { if (this == o) return true; diff --git a/src/java/org/apache/cassandra/repair/messages/FinalizePropose.java b/src/java/org/apache/cassandra/repair/messages/FinalizePropose.java index b3c4bfddb9..c1e76e347b 100644 --- a/src/java/org/apache/cassandra/repair/messages/FinalizePropose.java +++ b/src/java/org/apache/cassandra/repair/messages/FinalizePropose.java @@ -36,6 +36,12 @@ public class FinalizePropose extends RepairMessage this.sessionID = sessionID; } + @Override + public TimeUUID parentRepairSession() + { + return sessionID; + } + public boolean equals(Object o) { if (this == o) return true; diff --git a/src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java b/src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java index 9ac1461f65..1a3a6ecd38 100644 --- a/src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java +++ b/src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java @@ -50,6 +50,12 @@ public class PrepareConsistentRequest extends RepairMessage this.participants = ImmutableSet.copyOf(participants); } + @Override + public TimeUUID parentRepairSession() + { + return parentSession; + } + public boolean equals(Object o) { if (this == o) return true; diff --git a/src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java b/src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java index a422d7f9f7..80f62c1c03 100644 --- a/src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java +++ b/src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java @@ -45,6 +45,12 @@ public class PrepareConsistentResponse extends RepairMessage this.success = success; } + @Override + public TimeUUID parentRepairSession() + { + return parentSession; + } + public boolean equals(Object o) { if (this == o) return true; diff --git a/src/java/org/apache/cassandra/repair/messages/RepairMessage.java b/src/java/org/apache/cassandra/repair/messages/RepairMessage.java index 6411d12c78..f0cbf78f38 100644 --- a/src/java/org/apache/cassandra/repair/messages/RepairMessage.java +++ b/src/java/org/apache/cassandra/repair/messages/RepairMessage.java @@ -86,6 +86,13 @@ public abstract class RepairMessage map.put(Verb.SYNC_REQ, timeoutVersion); map.put(Verb.VALIDATION_RSP, SUPPORTS_RETRY); map.put(Verb.SYNC_RSP, SUPPORTS_RETRY); + // IR messages + map.put(Verb.PREPARE_CONSISTENT_REQ, SUPPORTS_RETRY); + map.put(Verb.PREPARE_CONSISTENT_RSP, SUPPORTS_RETRY); + map.put(Verb.FINALIZE_PROPOSE_MSG, SUPPORTS_RETRY); + map.put(Verb.FINALIZE_PROMISE_MSG, SUPPORTS_RETRY); + map.put(Verb.FINALIZE_COMMIT_MSG, SUPPORTS_RETRY); + map.put(Verb.FAILED_SESSION_MSG, SUPPORTS_RETRY); VERB_TIMEOUT_VERSIONS = Collections.unmodifiableMap(map); EnumSet allowsRetry = EnumSet.noneOf(Verb.class); @@ -96,6 +103,13 @@ public abstract class RepairMessage allowsRetry.add(Verb.SYNC_RSP); allowsRetry.add(Verb.SNAPSHOT_MSG); allowsRetry.add(Verb.CLEANUP_MSG); + // IR messages + allowsRetry.add(Verb.PREPARE_CONSISTENT_REQ); + allowsRetry.add(Verb.PREPARE_CONSISTENT_RSP); + allowsRetry.add(Verb.FINALIZE_PROPOSE_MSG); + allowsRetry.add(Verb.FINALIZE_PROMISE_MSG); + allowsRetry.add(Verb.FINALIZE_COMMIT_MSG); + allowsRetry.add(Verb.FAILED_SESSION_MSG); ALLOWS_RETRY = Collections.unmodifiableSet(allowsRetry); } @@ -154,6 +168,11 @@ public abstract class RepairMessage sendMessageWithRetries(ctx, backoff(ctx, verb), always(), request, verb, endpoint, NOOP_CALLBACK, 0); } + public static void sendMessageWithRetries(SharedContext ctx, Supplier allowRetry, RepairMessage request, Verb verb, InetAddressAndPort endpoint) + { + sendMessageWithRetries(ctx, backoff(ctx, verb), allowRetry, request, verb, endpoint, NOOP_CALLBACK, 0); + } + @VisibleForTesting static void sendMessageWithRetries(SharedContext ctx, Backoff backoff, Supplier allowRetry, RepairMessage request, Verb verb, InetAddressAndPort endpoint, RequestCallback finalCallback, int attempt) { @@ -214,7 +233,7 @@ public abstract class RepairMessage } else { - noSpam.warn("{} Failure for repair verb " + verb + "; could not complete within {} attempts", prefix, attempt); + noSpam.warn("{} {} failure for repair verb " + verb + "; could not complete within {} attempts", prefix, reason, attempt); RepairMetrics.retryFailure(verb); } } @@ -279,4 +298,15 @@ public abstract class RepairMessage return ErrorHandling.TIMEOUT; return ErrorHandling.NONE; } + + public static void sendFailureResponse(SharedContext ctx, Message respondTo) + { + Message reply = respondTo.failureResponse(RequestFailureReason.UNKNOWN); + ctx.messaging().send(reply, respondTo.from()); + } + + public static void sendAck(SharedContext ctx, Message message) + { + ctx.messaging().send(message.emptyResponse(), message.from()); + } } diff --git a/test/unit/org/apache/cassandra/net/MatcherResponse.java b/test/unit/org/apache/cassandra/net/MatcherResponse.java index d7b37591df..cad03fcc71 100644 --- a/test/unit/org/apache/cassandra/net/MatcherResponse.java +++ b/test/unit/org/apache/cassandra/net/MatcherResponse.java @@ -187,11 +187,18 @@ public class MatcherResponse Message response = fnResponse.apply(message, to); if (response != null) { - RequestCallbacks.CallbackInfo cb = MessagingService.instance().callbacks.get(message.id(), to); - if (cb != null) - cb.callback.onResponse(response); + if (response.verb().isResponse()) + { + RequestCallbacks.CallbackInfo cb = MessagingService.instance().callbacks.get(message.id(), to); + if (cb != null) + cb.callback.onResponse(response); + else + processResponse(response); + } else + { processResponse(response); + } spy.matchingResponse(response); } diff --git a/test/unit/org/apache/cassandra/repair/ConcurrentIrWithPreviewFuzzTest.java b/test/unit/org/apache/cassandra/repair/ConcurrentIrWithPreviewFuzzTest.java index eef13dc8b6..54ae644e34 100644 --- a/test/unit/org/apache/cassandra/repair/ConcurrentIrWithPreviewFuzzTest.java +++ b/test/unit/org/apache/cassandra/repair/ConcurrentIrWithPreviewFuzzTest.java @@ -28,9 +28,11 @@ import accord.utils.Gen; import accord.utils.Gens; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.RetrySpec; +import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.repair.consistent.LocalSessions; import org.apache.cassandra.repair.state.Completable; import org.apache.cassandra.utils.Closeable; +import org.apache.cassandra.utils.FailingBiConsumer; import org.assertj.core.api.Assertions; import static accord.utils.Property.qt; @@ -62,7 +64,7 @@ public class ConcurrentIrWithPreviewFuzzTest extends FuzzTestBase // cause a delay in validation to have more failing previews closeables.add(cluster.nodes.get(pickParticipant(rs, previewCoordinator, preview)).doValidation(next -> (cfs, validator) -> { if (validator.desc.parentSessionId.equals(preview.state.id)) - cluster.unorderedScheduled.schedule(() -> next.accept(cfs, validator), 1, TimeUnit.HOURS); + delayValidation(cluster, ir, next, cfs, validator); else next.acceptOrFail(cfs, validator); })); // make sure listeners don't leak @@ -88,4 +90,18 @@ public class ConcurrentIrWithPreviewFuzzTest extends FuzzTestBase } }); } + + private void delayValidation(Cluster cluster, RepairCoordinator ir, FailingBiConsumer next, ColumnFamilyStore cfs, Validator validator) + { + cluster.unorderedScheduled.schedule(() -> { + // make sure to wait for IR to complete... + Completable.Result result = ir.state.getResult(); + if (result == null) + { + delayValidation(cluster, ir, next, cfs, validator); + return; + } + next.accept(cfs, validator); + }, 1, TimeUnit.HOURS); + } } diff --git a/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java b/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java index 7c6a4601e1..39576c95f0 100644 --- a/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java +++ b/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java @@ -114,6 +114,7 @@ public class FailingRepairFuzzTest extends FuzzTestBase } cluster.processAll(); + Assertions.assertThat(repair.state.isComplete()).describedAs("Repair job did not complete, and no work is pending...").isTrue(); Assertions.assertThat(repair.state.getResult().kind).describedAs("Unexpected state: %s -> %s; example %d", repair.state, repair.state.getResult(), example).isEqualTo(Completable.Result.Kind.FAILURE); switch (stage) { diff --git a/test/unit/org/apache/cassandra/repair/FuzzTestBase.java b/test/unit/org/apache/cassandra/repair/FuzzTestBase.java index d29f35799e..fb7702cd82 100644 --- a/test/unit/org/apache/cassandra/repair/FuzzTestBase.java +++ b/test/unit/org/apache/cassandra/repair/FuzzTestBase.java @@ -24,6 +24,7 @@ import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.sql.Timestamp; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.Comparator; @@ -73,6 +74,7 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Digest; +import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.compaction.ICompactionManager; import org.apache.cassandra.db.marshal.EmptyType; import org.apache.cassandra.db.repair.CassandraTableRepairManager; @@ -280,6 +282,12 @@ public abstract class FuzzTestBase extends CQLTester.InMemory createSchema(); } + protected void cleanupRepairTables() + { + for (String table : Arrays.asList(SystemKeyspace.REPAIRS)) + execute(String.format("TRUNCATE %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, table)); + } + private void createSchema() { // The main reason to use random here with a fixed seed is just to have a set of tables that are not hard coded. @@ -329,12 +337,8 @@ public abstract class FuzzTestBase extends CQLTester.InMemory switch (message.verb()) { // these messages are not resilent to ephemeral issues - case PREPARE_CONSISTENT_REQ: - case PREPARE_CONSISTENT_RSP: - case FINALIZE_PROPOSE_MSG: - case FINALIZE_PROMISE_MSG: - case FINALIZE_COMMIT_MSG: - case FAILED_SESSION_MSG: + case STATUS_REQ: + case STATUS_RSP: noFaults.add(message.id()); return Faults.NONE; default: @@ -357,7 +361,10 @@ public abstract class FuzzTestBase extends CQLTester.InMemory static void assertSuccess(int example, boolean shouldSync, RepairCoordinator repair) { - Assertions.assertThat(repair.state.getResult()).describedAs("Unexpected state: %s -> %s; example %d", repair.state, repair.state.getResult(), example).isEqualTo(Completable.Result.success(repairSuccessMessage(repair))); + Completable.Result result = repair.state.getResult(); + Assertions.assertThat(result) + .describedAs("Expected repair to have completed with success, but is still running... %s; example %d", repair.state, example).isNotNull() + .describedAs("Unexpected state: %s -> %s; example %d", repair.state, result, example).isEqualTo(Completable.Result.success(repairSuccessMessage(repair))); Assertions.assertThat(repair.state.getStateTimesMillis().keySet()).isEqualTo(EnumSet.allOf(CoordinatorState.State.class)); Assertions.assertThat(repair.state.getSessions()).isNotEmpty(); boolean shouldSnapshot = repair.state.options.getParallelism() != RepairParallelism.PARALLEL @@ -643,6 +650,8 @@ public abstract class FuzzTestBase extends CQLTester.InMemory orderedExecutor = globalExecutor.configureSequential("ignore").build(); unorderedScheduled = globalExecutor.scheduled("ignored"); + + // We run tests in an isolated JVM per class, so not cleaing up is safe... but if that assumption ever changes, will need to cleanup Stage.ANTI_ENTROPY.unsafeSetExecutor(orderedExecutor); Stage.INTERNAL_RESPONSE.unsafeSetExecutor(unorderedScheduled); @@ -747,7 +756,7 @@ public abstract class FuzzTestBase extends CQLTester.InMemory } } - private static class CallbackContext + private class CallbackContext { final RequestCallback callback; @@ -803,6 +812,8 @@ public abstract class FuzzTestBase extends CQLTester.InMemory CallbackContext cb; if (callback != null) { + if (callbacks.containsKey(message.id())) + throw new AssertionError("Message id " + message.id() + " already has a callback"); cb = new CallbackContext(callback); callbacks.put(message.id(), cb); } @@ -854,7 +865,14 @@ public abstract class FuzzTestBase extends CQLTester.InMemory if (ctx != null) { assert ctx == cb; - ctx.onFailure(to, RequestFailureReason.TIMEOUT); + try + { + ctx.onFailure(to, RequestFailureReason.TIMEOUT); + } + catch (Throwable t) + { + failures.add(t); + } } }, message.verb().expiresAfterNanos(), TimeUnit.NANOSECONDS); } diff --git a/test/unit/org/apache/cassandra/repair/SlowMessageFuzzTest.java b/test/unit/org/apache/cassandra/repair/SlowMessageFuzzTest.java index 00fb2817cb..03c151ec68 100644 --- a/test/unit/org/apache/cassandra/repair/SlowMessageFuzzTest.java +++ b/test/unit/org/apache/cassandra/repair/SlowMessageFuzzTest.java @@ -39,6 +39,7 @@ public class SlowMessageFuzzTest extends FuzzTestBase // to avoid unlucky timing issues, retry until success; given enough retries we should eventually become success DatabaseDescriptor.getRepairRetrySpec().maxAttempts = new RetrySpec.MaxAttempt(Integer.MAX_VALUE); qt().withPure(false).withExamples(10).check(rs -> { + cleanupRepairTables(); Cluster cluster = new Cluster(rs); enableMessageFaults(cluster); diff --git a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java index d10aa16cf8..1f91b99795 100644 --- a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java @@ -21,7 +21,6 @@ package org.apache.cassandra.repair.consistent; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -34,6 +33,7 @@ import com.google.common.collect.Sets; import org.junit.Assert; import org.junit.Test; +import org.apache.cassandra.net.Verb; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; @@ -44,8 +44,10 @@ import org.apache.cassandra.repair.CoordinatedRepairResult; import org.apache.cassandra.repair.RepairSessionResult; import org.apache.cassandra.repair.messages.FailSession; import org.apache.cassandra.repair.messages.FinalizeCommit; +import org.apache.cassandra.repair.messages.FinalizePromise; import org.apache.cassandra.repair.messages.FinalizePropose; import org.apache.cassandra.repair.messages.PrepareConsistentRequest; +import org.apache.cassandra.repair.messages.PrepareConsistentResponse; import org.apache.cassandra.repair.messages.RepairMessage; import org.apache.cassandra.utils.concurrent.AsyncPromise; import org.apache.cassandra.utils.concurrent.Future; @@ -82,7 +84,10 @@ public class CoordinatorSessionTest extends AbstractRepairTest static InstrumentedCoordinatorSession createInstrumentedSession() { - return new InstrumentedCoordinatorSession(createBuilder()); + MockMessaging msg = new MockMessaging(); + CoordinatorSession.Builder builder = createBuilder(); + builder.withContext(SharedContext.Global.instance.withMessaging(msg)); + return new InstrumentedCoordinatorSession(msg, builder); } private static RepairSessionResult createResult(CoordinatorSession coordinator) @@ -99,20 +104,11 @@ public class CoordinatorSessionTest extends AbstractRepairTest private static class InstrumentedCoordinatorSession extends CoordinatorSession { - public InstrumentedCoordinatorSession(Builder builder) + private final Map> sentMessages; + public InstrumentedCoordinatorSession(MockMessaging messaging, Builder builder) { super(builder); - } - - Map> sentMessages = new HashMap<>(); - - protected void sendMessage(InetAddressAndPort destination, Message message) - { - if (!sentMessages.containsKey(destination)) - { - sentMessages.put(destination, new ArrayList<>()); - } - sentMessages.get(destination).add(message.payload); + this.sentMessages = messaging.sentMessages; } Runnable onSetRepairing = null; @@ -242,17 +238,17 @@ public class CoordinatorSessionTest extends AbstractRepairTest // participants respond to coordinator, and repair begins once all participants have responded with success Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); - - coordinator.handlePrepareResponse(PARTICIPANT1, true); + + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT1, true))); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); - - coordinator.handlePrepareResponse(PARTICIPANT2, true); + + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT2, true))); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); // set the setRepairing callback to verify the correct state when it's called Assert.assertFalse(coordinator.setRepairingCalled); coordinator.onSetRepairing = () -> Assert.assertEquals(PREPARED, coordinator.getState()); - coordinator.handlePrepareResponse(PARTICIPANT3, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT3, true))); Assert.assertTrue(coordinator.setRepairingCalled); Assert.assertTrue(repairSubmitted.get()); @@ -276,16 +272,16 @@ public class CoordinatorSessionTest extends AbstractRepairTest coordinator.sentMessages.clear(); Assert.assertEquals(ConsistentSession.State.REPAIRING, coordinator.getState()); - coordinator.handleFinalizePromise(PARTICIPANT1, true); + coordinator.handleFinalizePromise(Message.out(Verb.FINALIZE_PROMISE_MSG, new FinalizePromise(coordinator.sessionID, PARTICIPANT1, true))); Assert.assertEquals(ConsistentSession.State.REPAIRING, coordinator.getState()); - coordinator.handleFinalizePromise(PARTICIPANT2, true); + coordinator.handleFinalizePromise(Message.out(Verb.FINALIZE_PROMISE_MSG, new FinalizePromise(coordinator.sessionID, PARTICIPANT2, true))); Assert.assertEquals(ConsistentSession.State.REPAIRING, coordinator.getState()); // set the finalizeCommit callback so we can verify the state when it's called Assert.assertFalse(coordinator.finalizeCommitCalled); coordinator.onFinalizeCommit = () -> Assert.assertEquals(FINALIZE_PROMISED, coordinator.getState()); - coordinator.handleFinalizePromise(PARTICIPANT3, true); + coordinator.handleFinalizePromise(Message.out(Verb.FINALIZE_PROMISE_MSG, new FinalizePromise(coordinator.sessionID, PARTICIPANT3, true))); Assert.assertTrue(coordinator.finalizeCommitCalled); Assert.assertEquals(ConsistentSession.State.FINALIZED, coordinator.getState()); @@ -324,16 +320,16 @@ public class CoordinatorSessionTest extends AbstractRepairTest // participants respond to coordinator, and repair begins once all participants have responded with success Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); - coordinator.handlePrepareResponse(PARTICIPANT1, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT1, true))); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); - coordinator.handlePrepareResponse(PARTICIPANT2, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT2, true))); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); // set the setRepairing callback to verify the correct state when it's called Assert.assertFalse(coordinator.setRepairingCalled); coordinator.onSetRepairing = () -> Assert.assertEquals(PREPARED, coordinator.getState()); - coordinator.handlePrepareResponse(PARTICIPANT3, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT3, true))); Assert.assertTrue(coordinator.setRepairingCalled); Assert.assertTrue(repairSubmitted.get()); @@ -388,14 +384,14 @@ public class CoordinatorSessionTest extends AbstractRepairTest // participants respond to coordinator, and repair begins once all participants have responded Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); - coordinator.handlePrepareResponse(PARTICIPANT1, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT1, true))); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); Assert.assertEquals(PREPARED, coordinator.getParticipantState(PARTICIPANT1)); Assert.assertFalse(sessionResult.isDone()); // participant 2 fails to prepare for consistent repair Assert.assertFalse(coordinator.failCalled); - coordinator.handlePrepareResponse(PARTICIPANT2, false); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT2, false))); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); // we should have sent failure messages to the other participants, but not yet marked them failed internally assertMessageSent(coordinator, PARTICIPANT1, new FailSession(coordinator.sessionID)); @@ -411,7 +407,7 @@ public class CoordinatorSessionTest extends AbstractRepairTest // last outstanding response should cause repair to complete in failed state Assert.assertFalse(coordinator.setRepairingCalled); coordinator.onSetRepairing = Assert::fail; - coordinator.handlePrepareResponse(PARTICIPANT3, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT3, true))); Assert.assertTrue(coordinator.failCalled); Assert.assertFalse(coordinator.setRepairingCalled); Assert.assertFalse(repairSubmitted.get()); @@ -453,16 +449,16 @@ public class CoordinatorSessionTest extends AbstractRepairTest // participants respond to coordinator, and repair begins once all participants have responded with success Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); - coordinator.handlePrepareResponse(PARTICIPANT1, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT1, true))); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); - coordinator.handlePrepareResponse(PARTICIPANT2, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT2, true))); Assert.assertEquals(ConsistentSession.State.PREPARING, coordinator.getState()); // set the setRepairing callback to verify the correct state when it's called Assert.assertFalse(coordinator.setRepairingCalled); coordinator.onSetRepairing = () -> Assert.assertEquals(PREPARED, coordinator.getState()); - coordinator.handlePrepareResponse(PARTICIPANT3, true); + coordinator.handlePrepareResponse(Message.out(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(coordinator.sessionID, PARTICIPANT3, true))); Assert.assertTrue(coordinator.setRepairingCalled); Assert.assertTrue(repairSubmitted.get()); @@ -486,18 +482,18 @@ public class CoordinatorSessionTest extends AbstractRepairTest coordinator.sentMessages.clear(); Assert.assertEquals(ConsistentSession.State.REPAIRING, coordinator.getState()); - coordinator.handleFinalizePromise(PARTICIPANT1, true); + coordinator.handleFinalizePromise(Message.out(Verb.FINALIZE_PROMISE_MSG, new FinalizePromise(coordinator.sessionID, PARTICIPANT1, true))); Assert.assertEquals(ConsistentSession.State.REPAIRING, coordinator.getState()); Assert.assertFalse(coordinator.failCalled); - coordinator.handleFinalizePromise(PARTICIPANT2, false); + coordinator.handleFinalizePromise(Message.out(Verb.FINALIZE_PROMISE_MSG, new FinalizePromise(coordinator.sessionID, PARTICIPANT2, false))); Assert.assertEquals(ConsistentSession.State.FAILED, coordinator.getState()); Assert.assertTrue(coordinator.failCalled); // additional success messages should be ignored Assert.assertFalse(coordinator.finalizeCommitCalled); coordinator.onFinalizeCommit = Assert::fail; - coordinator.handleFinalizePromise(PARTICIPANT3, true); + coordinator.handleFinalizePromise(Message.out(Verb.FINALIZE_PROMISE_MSG, new FinalizePromise(coordinator.sessionID, PARTICIPANT3, true))); Assert.assertFalse(coordinator.finalizeCommitCalled); Assert.assertEquals(ConsistentSession.State.FAILED, coordinator.getState()); diff --git a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java index 9f0ee4635d..cd9dfba4ed 100644 --- a/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java +++ b/test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java @@ -26,6 +26,8 @@ import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.net.Message; +import org.apache.cassandra.net.Verb; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; import org.apache.cassandra.repair.AbstractRepairTest; @@ -60,21 +62,23 @@ public class CoordinatorSessionsTest extends AbstractRepairTest int prepareResponseCalls = 0; InetAddressAndPort preparePeer = null; boolean prepareSuccess = false; - public synchronized void handlePrepareResponse(InetAddressAndPort participant, boolean success) + @Override + public synchronized void handlePrepareResponse(Message msg) { prepareResponseCalls++; - preparePeer = participant; - prepareSuccess = success; + preparePeer = msg.payload.participant; + prepareSuccess = msg.payload.success; } int finalizePromiseCalls = 0; InetAddressAndPort promisePeer = null; boolean promiseSuccess = false; - public synchronized void handleFinalizePromise(InetAddressAndPort participant, boolean success) + @Override + public synchronized void handleFinalizePromise(Message message) { finalizePromiseCalls++; - promisePeer = participant; - promiseSuccess = success; + promisePeer = message.payload.participant; + promiseSuccess = message.payload.promised; } int failCalls = 0; @@ -150,7 +154,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest InstrumentedCoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS, false); Assert.assertEquals(0, session.prepareResponseCalls); - sessions.handlePrepareResponse(new PrepareConsistentResponse(sessionID, PARTICIPANT1, true)); + sessions.handlePrepareResponse(Message.builder(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(sessionID, PARTICIPANT1, true)).build()); Assert.assertEquals(1, session.prepareResponseCalls); Assert.assertEquals(PARTICIPANT1, session.preparePeer); Assert.assertTrue(session.prepareSuccess); @@ -162,7 +166,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest InstrumentedCoordinatorSessions sessions = new InstrumentedCoordinatorSessions(); TimeUUID fakeID = nextTimeUUID(); - sessions.handlePrepareResponse(new PrepareConsistentResponse(fakeID, PARTICIPANT1, true)); + sessions.handlePrepareResponse(Message.builder(Verb.PREPARE_CONSISTENT_RSP, new PrepareConsistentResponse(fakeID, PARTICIPANT1, true)).build()); Assert.assertNull(sessions.getSession(fakeID)); } @@ -175,7 +179,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest InstrumentedCoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS, false); Assert.assertEquals(0, session.finalizePromiseCalls); - sessions.handleFinalizePromiseMessage(new FinalizePromise(sessionID, PARTICIPANT1, true)); + sessions.handleFinalizePromiseMessage(Message.builder(Verb.FINALIZE_PROMISE_MSG, new FinalizePromise(sessionID, PARTICIPANT1, true)).build()); Assert.assertEquals(1, session.finalizePromiseCalls); Assert.assertEquals(PARTICIPANT1, session.promisePeer); Assert.assertTrue(session.promiseSuccess); @@ -187,7 +191,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest InstrumentedCoordinatorSessions sessions = new InstrumentedCoordinatorSessions(); TimeUUID fakeID = nextTimeUUID(); - sessions.handleFinalizePromiseMessage(new FinalizePromise(fakeID, PARTICIPANT1, true)); + sessions.handleFinalizePromiseMessage(Message.builder(Verb.FINALIZE_PROMISE_MSG, new FinalizePromise(fakeID, PARTICIPANT1, true)).build()); Assert.assertNull(sessions.getSession(fakeID)); } diff --git a/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java b/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java index 0f9912437e..e19f8abca9 100644 --- a/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java @@ -39,6 +39,7 @@ import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.net.Verb; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.UntypedResultSet; @@ -130,13 +131,20 @@ public class LocalSessionTest extends AbstractRepairTest static class InstrumentedLocalSessions extends LocalSessions { - Map> sentMessages = new HashMap<>(); + final Map> sentMessages; public InstrumentedLocalSessions() { - super(SharedContext.Global.instance); + this(new MockMessaging()); } + private InstrumentedLocalSessions(MockMessaging messaging) + { + super(SharedContext.Global.instance.withMessaging(messaging)); + sentMessages = messaging.sentMessages; + } + + @Override protected void sendMessage(InetAddressAndPort destination, Message message) { if (!sentMessages.containsKey(destination)) @@ -178,7 +186,7 @@ public class LocalSessionTest extends AbstractRepairTest public LocalSession prepareForTest(TimeUUID sessionID) { prepareSessionFuture = new AsyncPromise<>(); - handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); + handlePrepareMessage(Message.builder(Verb.PREPARE_CONSISTENT_REQ, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)).from(PARTICIPANT1).build()); prepareSessionFuture.trySuccess(null); sentMessages.clear(); return getSession(sessionID); @@ -281,7 +289,7 @@ public class LocalSessionTest extends AbstractRepairTest // replacing future so we can inspect state before and after anti compaction callback sessions.prepareSessionFuture = new AsyncPromise<>(); Assert.assertFalse(sessions.prepareSessionCalled); - sessions.handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); + sessions.handlePrepareMessage(Message.builder(Verb.PREPARE_CONSISTENT_REQ, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)).from(PARTICIPANT1).build()); Assert.assertTrue(sessions.prepareSessionCalled); Assert.assertTrue(sessions.sentMessages.isEmpty()); @@ -316,7 +324,7 @@ public class LocalSessionTest extends AbstractRepairTest // replacing future so we can inspect state before and after anti compaction callback sessions.prepareSessionFuture = new AsyncPromise<>(); Assert.assertFalse(sessions.prepareSessionCalled); - sessions.handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); + sessions.handlePrepareMessage(Message.builder(Verb.PREPARE_CONSISTENT_REQ, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)).from(PARTICIPANT1).build()); Assert.assertTrue(sessions.prepareSessionCalled); Assert.assertTrue(sessions.sentMessages.isEmpty()); @@ -348,7 +356,7 @@ public class LocalSessionTest extends AbstractRepairTest { TimeUUID sessionID = nextTimeUUID(); InstrumentedLocalSessions sessions = new InstrumentedLocalSessions(); - sessions.handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); + sessions.handlePrepareMessage(Message.builder(Verb.PREPARE_CONSISTENT_REQ, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)).from(PARTICIPANT1).build()); Assert.assertNull(sessions.getSession(sessionID)); assertMessagesSent(sessions, COORDINATOR, new PrepareConsistentResponse(sessionID, PARTICIPANT1, false)); } @@ -372,7 +380,7 @@ public class LocalSessionTest extends AbstractRepairTest }; sessions.start(); - sessions.handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); + sessions.handlePrepareMessage(Message.builder(Verb.PREPARE_CONSISTENT_REQ, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)).from(PARTICIPANT1).build()); BooleanSupplier isCancelled = isCancelledRef.get(); Assert.assertNotNull(isCancelled); @@ -470,7 +478,7 @@ public class LocalSessionTest extends AbstractRepairTest // should send a promised message to coordinator and set session state accordingly sessions.sentMessages.clear(); - sessions.handleFinalizeProposeMessage(COORDINATOR, new FinalizePropose(sessionID)); + sessions.handleFinalizeProposeMessage(Message.builder(Verb.FINALIZE_PROPOSE_MSG, new FinalizePropose(sessionID)).from(COORDINATOR).build()); Assert.assertEquals(FINALIZE_PROMISED, session.getState()); Assert.assertEquals(session, sessions.loadUnsafe(sessionID)); assertMessagesSent(sessions, COORDINATOR, new FinalizePromise(sessionID, PARTICIPANT1, true)); @@ -492,7 +500,7 @@ public class LocalSessionTest extends AbstractRepairTest // should fail the session and send a failure message to the coordinator sessions.sentMessages.clear(); - sessions.handleFinalizeProposeMessage(COORDINATOR, new FinalizePropose(sessionID)); + sessions.handleFinalizeProposeMessage(Message.builder(Verb.FINALIZE_PROPOSE_MSG, new FinalizePropose(sessionID)).from(COORDINATOR).build()); Assert.assertEquals(FAILED, session.getState()); Assert.assertEquals(session, sessions.loadUnsafe(sessionID)); assertMessagesSent(sessions, COORDINATOR, new FailSession(sessionID)); @@ -503,7 +511,7 @@ public class LocalSessionTest extends AbstractRepairTest { InstrumentedLocalSessions sessions = new InstrumentedLocalSessions(); TimeUUID fakeID = nextTimeUUID(); - sessions.handleFinalizeProposeMessage(COORDINATOR, new FinalizePropose(fakeID)); + sessions.handleFinalizeProposeMessage(Message.builder(Verb.FINALIZE_PROPOSE_MSG, new FinalizePropose(fakeID)).from(COORDINATOR).build()); Assert.assertNull(sessions.getSession(fakeID)); assertMessagesSent(sessions, COORDINATOR, new FailSession(fakeID)); } @@ -522,12 +530,12 @@ public class LocalSessionTest extends AbstractRepairTest // create session and move to finalized promised sessions.prepareForTest(sessionID); sessions.maybeSetRepairing(sessionID); - sessions.handleFinalizeProposeMessage(COORDINATOR, new FinalizePropose(sessionID)); + sessions.handleFinalizeProposeMessage(Message.builder(Verb.FINALIZE_PROPOSE_MSG, new FinalizePropose(sessionID)).from(COORDINATOR).build()); Assert.assertEquals(0, (int) sessions.completedSessions.getOrDefault(sessionID, 0)); sessions.sentMessages.clear(); LocalSession session = sessions.getSession(sessionID); - sessions.handleFinalizeCommitMessage(PARTICIPANT1, new FinalizeCommit(sessionID)); + sessions.handleFinalizeCommitMessage(Message.builder(Verb.FINALIZE_COMMIT_MSG, new FinalizeCommit(sessionID)).from(PARTICIPANT1).build()); Assert.assertEquals(FINALIZED, session.getState()); Assert.assertEquals(session, sessions.loadUnsafe(sessionID)); @@ -541,7 +549,7 @@ public class LocalSessionTest extends AbstractRepairTest InstrumentedLocalSessions sessions = new InstrumentedLocalSessions(); sessions.start(); TimeUUID fakeID = nextTimeUUID(); - sessions.handleFinalizeCommitMessage(PARTICIPANT1, new FinalizeCommit(fakeID)); + sessions.handleFinalizeCommitMessage(Message.builder(Verb.FINALIZE_COMMIT_MSG, new FinalizeCommit(fakeID)).from(PARTICIPANT1).build()); Assert.assertNull(sessions.getSession(fakeID)); Assert.assertTrue(sessions.sentMessages.isEmpty()); } @@ -717,7 +725,7 @@ public class LocalSessionTest extends AbstractRepairTest InstrumentedLocalSessions sessions = new InstrumentedLocalSessions(); sessions.start(); sessions.prepareSessionFuture = new AsyncPromise<>(); // prevent moving to prepared - sessions.handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); + sessions.handlePrepareMessage(Message.builder(Verb.PREPARE_CONSISTENT_REQ, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)).from(PARTICIPANT1).build()); LocalSession session = sessions.getSession(sessionID); Assert.assertNotNull(session); @@ -744,7 +752,7 @@ public class LocalSessionTest extends AbstractRepairTest InstrumentedLocalSessions sessions = new InstrumentedLocalSessions(); sessions.start(); sessions.prepareSessionFuture = new AsyncPromise<>(); - sessions.handlePrepareMessage(PARTICIPANT1, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)); + sessions.handlePrepareMessage(Message.builder(Verb.PREPARE_CONSISTENT_REQ, new PrepareConsistentRequest(sessionID, COORDINATOR, PARTICIPANTS)).from(PARTICIPANT1).build()); sessions.prepareSessionFuture.trySuccess(null); Assert.assertTrue(sessions.isSessionInProgress(sessionID)); @@ -770,8 +778,8 @@ public class LocalSessionTest extends AbstractRepairTest sessions.prepareForTest(sessionID); sessions.maybeSetRepairing(sessionID); - sessions.handleFinalizeProposeMessage(COORDINATOR, new FinalizePropose(sessionID)); - sessions.handleFinalizeCommitMessage(PARTICIPANT1, new FinalizeCommit(sessionID)); + sessions.handleFinalizeProposeMessage(Message.builder(Verb.FINALIZE_PROPOSE_MSG, new FinalizePropose(sessionID)).from(COORDINATOR).build()); + sessions.handleFinalizeCommitMessage(Message.builder(Verb.FINALIZE_COMMIT_MSG, new FinalizeCommit(sessionID)).from(PARTICIPANT1).build()); LocalSession session = sessions.getSession(sessionID); Assert.assertTrue(session.repairedAt != ActiveRepairService.UNREPAIRED_SSTABLE); diff --git a/test/unit/org/apache/cassandra/repair/consistent/MockMessaging.java b/test/unit/org/apache/cassandra/repair/consistent/MockMessaging.java new file mode 100644 index 0000000000..5a59ef2639 --- /dev/null +++ b/test/unit/org/apache/cassandra/repair/consistent/MockMessaging.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.repair.consistent; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.ConnectionType; +import org.apache.cassandra.net.Message; +import org.apache.cassandra.net.MessageDelivery; +import org.apache.cassandra.net.NoPayload; +import org.apache.cassandra.net.RequestCallback; +import org.apache.cassandra.net.Verb; +import org.apache.cassandra.repair.messages.RepairMessage; +import org.apache.cassandra.utils.concurrent.Future; + +class MockMessaging implements MessageDelivery +{ + Map> sentMessages = new HashMap<>(); + Map acks = new HashMap<>(); + Map failures = new HashMap<>(); + + @Override + public void send(Message message, InetAddressAndPort destination) + { + if (message.verb() == Verb.REPAIR_RSP && message.payload instanceof NoPayload) + { + acks.compute(destination, (ignore, accum) -> accum == null ? 1 : accum + 1); + return; + } + if (message.verb() == Verb.FAILURE_RSP) + { + failures.compute(destination, (ignore, accum) -> accum == null ? 1 : accum + 1); + return; + } + if (!(message.payload instanceof RepairMessage)) + throw new AssertionError("Unexpected message: " + message); + + if (!sentMessages.containsKey(destination)) + { + sentMessages.put(destination, new ArrayList<>()); + } + sentMessages.get(destination).add((RepairMessage) message.payload); + } + + @Override + public void sendWithCallback(Message message, InetAddressAndPort to, RequestCallback cb) + { + send(message, to); + } + + @Override + public void sendWithCallback(Message message, InetAddressAndPort to, RequestCallback cb, ConnectionType specifyConnection) + { + send(message, to); + } + + @Override + public Future> sendWithResult(Message message, InetAddressAndPort to) + { + throw new UnsupportedOperationException(); + } + + @Override + public void respond(V response, Message message) + { + throw new UnsupportedOperationException(); + } +}