in src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java [97:377]
public void doVerb(final Message<RepairMessage> message)
{
if (DatabaseDescriptor.getAccordTransactionsEnabled()
&& ctx.cms().maybeFetchLogFromPeerOrCMSAsync(ctx.messaging(), message, () -> doVerb(message)))
return;
// TODO add cancel/interrupt message
RepairJobDesc desc = message.payload.desc;
try
{
switch (message.verb())
{
case PREPARE_MSG:
{
PrepareMessage prepareMessage = (PrepareMessage) message.payload;
logger.debug("Preparing, {}", prepareMessage);
ParticipateState state = new ParticipateState(ctx.clock(), message.from(), prepareMessage);
if (!ctx.repair().register(state))
{
replyDedup(ctx.repair().participate(state.id), message);
return;
}
if (!ctx.repair().verifyCompactionsPendingThreshold(prepareMessage.parentRepairSession, prepareMessage.previewKind))
{
// error is logged in verifyCompactionsPendingThreshold
state.phase.fail("Too many pending compactions");
sendFailureResponse(message);
return;
}
if (!ActiveRepairService.verifyDiskHeadroomThreshold(prepareMessage.parentRepairSession, prepareMessage.previewKind, prepareMessage.isIncremental))
{
// error is logged in verifyDiskHeadroomThreshold
state.phase.fail("Not enough disk headroom to perform incremental repair");
sendFailureResponse(message);
return;
}
List<ColumnFamilyStore> columnFamilyStores = new ArrayList<>(prepareMessage.tableIds.size());
for (TableId tableId : prepareMessage.tableIds)
{
ColumnFamilyStore columnFamilyStore = ColumnFamilyStore.getIfExists(tableId);
if (columnFamilyStore == null)
{
String reason = String.format("Table with id %s was dropped during prepare phase of repair",
tableId);
state.phase.fail(reason);
logErrorAndSendFailureResponse(reason, message);
return;
}
columnFamilyStores.add(columnFamilyStore);
}
state.phase.accept();
ctx.repair().registerParentRepairSession(prepareMessage.parentRepairSession,
message.from(),
columnFamilyStores,
prepareMessage.ranges,
prepareMessage.isIncremental,
prepareMessage.repairedAt,
prepareMessage.isGlobal,
prepareMessage.previewKind);
sendAck(message);
}
break;
case SNAPSHOT_MSG:
{
logger.debug("Snapshotting {}", desc);
ParticipateState state = ctx.repair().participate(desc.parentSessionId);
if (state == null)
{
logErrorAndSendFailureResponse("Unknown repair " + desc.parentSessionId, message);
return;
}
final ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(desc.keyspace, desc.columnFamily);
if (cfs == null)
{
String reason = String.format("Table %s.%s was dropped during snapshot phase of repair %s",
desc.keyspace, desc.columnFamily, desc.parentSessionId);
state.phase.fail(reason);
logErrorAndSendFailureResponse(reason, message);
return;
}
ActiveRepairService.ParentRepairSession prs = ctx.repair().getParentRepairSession(desc.parentSessionId);
if (prs.setHasSnapshots())
{
state.getOrCreateJob(desc).snapshot();
TableRepairManager repairManager = cfs.getRepairManager();
if (prs.isGlobal)
{
repairManager.snapshot(desc.parentSessionId.toString(), prs.getRanges(), false);
}
else
{
repairManager.snapshot(desc.parentSessionId.toString(), desc.ranges, true);
}
logger.debug("Enqueuing response to snapshot request {} to {}", desc.sessionId, message.from());
}
sendAck(message);
}
break;
case VALIDATION_REQ:
{
ValidationRequest validationRequest = (ValidationRequest) message.payload;
logger.debug("Validating {}", validationRequest);
ParticipateState participate = ctx.repair().participate(desc.parentSessionId);
if (participate == null)
{
logErrorAndSendFailureResponse("Unknown repair " + desc.parentSessionId, message);
return;
}
ValidationState vState = new ValidationState(ctx.clock(), desc, message.from());
if (!register(message, participate, vState,
participate::register,
(d, i) -> participate.validation(d)))
return;
try
{
// trigger read-only compaction
ColumnFamilyStore store = ColumnFamilyStore.getIfExists(desc.keyspace, desc.columnFamily);
if (store == null)
{
String msg = String.format("Table %s.%s was dropped during validation phase of repair %s", desc.keyspace, desc.columnFamily, desc.parentSessionId);
vState.phase.fail(msg);
logErrorAndSendFailureResponse(msg, message);
return;
}
try
{
ctx.repair().consistent.local.maybeSetRepairing(desc.parentSessionId);
}
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(t);
vState.phase.fail(t.toString());
logErrorAndSendFailureResponse(t.toString(), message);
return;
}
PreviewKind previewKind;
try
{
previewKind = previewKind(desc.parentSessionId);
}
catch (NoSuchRepairSessionException e)
{
logger.warn("Parent repair session {} has been removed, failing repair", desc.parentSessionId);
vState.phase.fail(e);
sendFailureResponse(message);
return;
}
if (!acceptMessage(validationRequest, ctx.broadcastAddressAndPort(), message.from()))
{
RepairOutOfTokenRangeException e = new RepairOutOfTokenRangeException(validationRequest.desc.ranges);
logger.error("Got out-of-range repair request from " + message.from() + ": " + validationRequest.desc.ranges, e);
vState.phase.fail(e);
sendFailureResponse(message);
return;
}
vState.phase.accept();
sendAck(message);
Validator validator = new Validator(ctx, vState, validationRequest.nowInSec,
isIncremental(desc.parentSessionId),
previewKind,
validationRequest.dontPurgeTombstones);
ctx.validationManager().submitValidation(store, validator);
}
catch (Throwable t)
{
vState.phase.fail(t);
throw t;
}
}
break;
case SYNC_REQ:
{
// forwarded sync request
SyncRequest request = (SyncRequest) message.payload;
logger.debug("Syncing {}", request);
ParticipateState participate = ctx.repair().participate(desc.parentSessionId);
if (participate == null)
{
logErrorAndSendFailureResponse("Unknown repair " + desc.parentSessionId, message);
return;
}
SyncState state = new SyncState(ctx.clock(), desc, request.initiator, request.src, request.dst);
if (!register(message, participate, state,
participate::register,
participate::sync))
return;
state.phase.accept();
StreamingRepairTask task = new StreamingRepairTask(ctx, state, desc,
request.initiator,
request.src,
request.dst,
request.ranges,
isIncremental(desc.parentSessionId) ? desc.parentSessionId : null,
request.previewKind,
request.asymmetric);
task.run();
sendAck(message);
}
break;
case CLEANUP_MSG:
{
logger.debug("cleaning up repair");
CleanupMessage cleanup = (CleanupMessage) message.payload;
ParticipateState state = ctx.repair().participate(cleanup.parentRepairSession);
if (state != null)
state.phase.success("Cleanup message recieved");
ctx.repair().removeParentRepairSession(cleanup.parentRepairSession);
sendAck(message);
}
break;
case PREPARE_CONSISTENT_REQ:
ctx.repair().consistent.local.handlePrepareMessage(message);
break;
case PREPARE_CONSISTENT_RSP:
ctx.repair().consistent.coordinated.handlePrepareResponse(message);
break;
case FINALIZE_PROPOSE_MSG:
ctx.repair().consistent.local.handleFinalizeProposeMessage(message);
break;
case FINALIZE_PROMISE_MSG:
ctx.repair().consistent.coordinated.handleFinalizePromiseMessage(message);
break;
case FINALIZE_COMMIT_MSG:
ctx.repair().consistent.local.handleFinalizeCommitMessage(message);
break;
case FAILED_SESSION_MSG:
FailSession failure = (FailSession) message.payload;
sendAck(message);
ParticipateState p = ctx.repair().participate(failure.sessionID);
if (p != null)
p.phase.fail("Failure message from " + message.from());
ctx.repair().consistent.coordinated.handleFailSessionMessage(failure);
ctx.repair().consistent.local.handleFailSessionMessage(message.from(), failure);
break;
case STATUS_REQ:
ctx.repair().consistent.local.handleStatusRequest(message.from(), (StatusRequest) message.payload);
break;
case STATUS_RSP:
ctx.repair().consistent.local.handleStatusResponse(message.from(), (StatusResponse) message.payload);
break;
default:
ctx.repair().handleMessage(message);
break;
}
}
catch (Exception e)
{
logger.error("Got error, removing parent repair session");
if (desc != null && desc.parentSessionId != null)
{
ParticipateState parcipate = ctx.repair().participate(desc.parentSessionId);
if (parcipate != null)
parcipate.phase.fail(e);
ctx.repair().removeParentRepairSession(desc.parentSessionId);
}
throw new RuntimeException(e);
}
}