natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add retries to IR messages 

patch by David Capwell; reviewed by Blake Eggleston for CASSANDRA-18962
This commit is contained in:
David Capwell 2023-10-31 13:56:29 -07:00
parent 7463a7a977
commit dca76145c2
24 changed files with 524 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

12
src/java/org/apache/cassandra/net/Verb.java
View File

@ -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 ),

17
src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java
View File

@ -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<RepairMessage>
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<RepairMessage>
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<RepairMessage> message)
{
ctx.messaging().send(message.emptyResponse(), message.from());
RepairMessage.sendAck(ctx, message);
}
}

115
src/java/org/apache/cassandra/repair/SharedContext.java
View File

@ -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> 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();
}
}
}

2
src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
View File

@ -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}.
* <p/>
* (see {@link LocalSessions#handlePrepareMessage(InetAddressAndPort, PrepareConsistentRequest)}
* (see {@link LocalSessions#handlePrepareMessage(org.apache.cassandra.net.Message)}
* <p/>
* Once the coordinator recieves positive {@code PrepareConsistentResponse} messages from all the participants, the
* coordinator begins the normal repair process.

57
src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
View File

@ -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<RepairMessage> message)
{
logger.trace("Sending {} to {}", message.payload, destination);
ctx.messaging().send(message, destination);
}
public Future<Void> prepare()
{
Preconditions.checkArgument(allStates(State.PREPARING));
logger.info("Beginning prepare phase of incremental repair session {}", sessionID);
Message<RepairMessage> 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<PrepareConsistentResponse> 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<RepairMessage> 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<FinalizePromise> 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<RepairMessage> 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<RepairMessage> 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);
}
}
}

24
src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java
View File

@ -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<? extends RepairMessage> 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<PrepareConsistentResponse>) msg);
}
else
{
sendFailureResponse(ctx, msg);
}
}
public void handleFinalizePromiseMessage(FinalizePromise msg)
public void handleFinalizePromiseMessage(Message<? extends RepairMessage> message)
{
FinalizePromise msg = (FinalizePromise) message.payload;
CoordinatorSession session = getSession(msg.sessionID);
if (session != null)
{
session.handleFinalizePromise(msg.participant, msg.promised);
session.handleFinalizePromise((Message<FinalizePromise>) message);
}
else
{
sendFailureResponse(ctx, message);
}
}

77
src/java/org/apache/cassandra/repair/consistent/LocalSessions.java
View File

@ -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<FailSession> 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<? extends RepairMessage> 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<PrepareConsistentResponse> 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<Boolean> 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<? extends RepairMessage> 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<? extends RepairMessage> 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)

6
src/java/org/apache/cassandra/repair/messages/FailSession.java
View File

@ -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;

6
src/java/org/apache/cassandra/repair/messages/FinalizeCommit.java
View File

@ -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;

6
src/java/org/apache/cassandra/repair/messages/FinalizePromise.java
View File

@ -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;

6
src/java/org/apache/cassandra/repair/messages/FinalizePropose.java
View File

@ -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;

6
src/java/org/apache/cassandra/repair/messages/PrepareConsistentRequest.java
View File

@ -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;

6
src/java/org/apache/cassandra/repair/messages/PrepareConsistentResponse.java
View File

@ -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;

32
src/java/org/apache/cassandra/repair/messages/RepairMessage.java
View File

@ -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<Verb> 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<Boolean> allowRetry, RepairMessage request, Verb verb, InetAddressAndPort endpoint)
{
sendMessageWithRetries(ctx, backoff(ctx, verb), allowRetry, request, verb, endpoint, NOOP_CALLBACK, 0);
}
@VisibleForTesting
static <T> void sendMessageWithRetries(SharedContext ctx, Backoff backoff, Supplier<Boolean> allowRetry, RepairMessage request, Verb verb, InetAddressAndPort endpoint, RequestCallback<T> 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<? extends RepairMessage> message)
{
ctx.messaging().send(message.emptyResponse(), message.from());
}
}

13
test/unit/org/apache/cassandra/net/MatcherResponse.java
View File

@ -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);
}

18
test/unit/org/apache/cassandra/repair/ConcurrentIrWithPreviewFuzzTest.java
View File

@ -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<ColumnFamilyStore, Validator> 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);
}
}

1
test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java
View File

@ -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)
{

36
test/unit/org/apache/cassandra/repair/FuzzTestBase.java
View File

@ -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);
}

1
test/unit/org/apache/cassandra/repair/SlowMessageFuzzTest.java
View File

@ -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);

64
test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionTest.java
View File

@ -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<InetAddressAndPort, List<RepairMessage>> sentMessages;
public InstrumentedCoordinatorSession(MockMessaging messaging, Builder builder)
{
super(builder);
}
Map<InetAddressAndPort, List<RepairMessage>> sentMessages = new HashMap<>();
protected void sendMessage(InetAddressAndPort destination, Message<RepairMessage> 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());

24
test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java
View File

@ -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<PrepareConsistentResponse> 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<FinalizePromise> 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));
}

42
test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java
View File

@ -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<InetAddressAndPort, List<RepairMessage>> sentMessages = new HashMap<>();
final Map<InetAddressAndPort, List<RepairMessage>> 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<? extends RepairMessage> 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);

88
test/unit/org/apache/cassandra/repair/consistent/MockMessaging.java Normal file
View File

@ -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<InetAddressAndPort, List<RepairMessage>> sentMessages = new HashMap<>();
Map<InetAddressAndPort, Integer> acks = new HashMap<>();
Map<InetAddressAndPort, Integer> failures = new HashMap<>();
@Override
public <REQ> void send(Message<REQ> 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 <REQ, RSP> void sendWithCallback(Message<REQ> message, InetAddressAndPort to, RequestCallback<RSP> cb)
{
send(message, to);
}
@Override
public <REQ, RSP> void sendWithCallback(Message<REQ> message, InetAddressAndPort to, RequestCallback<RSP> cb, ConnectionType specifyConnection)
{
send(message, to);
}
@Override
public <REQ, RSP> Future<Message<RSP>> sendWithResult(Message<REQ> message, InetAddressAndPort to)
{
throw new UnsupportedOperationException();
}
@Override
public <V> void respond(V response, Message<?> message)
{
throw new UnsupportedOperationException();
}
}