maedhroz commented on code in PR #2660:
URL: https://github.com/apache/cassandra/pull/2660#discussion_r1319393686
##########
src/java/org/apache/cassandra/repair/messages/RepairMessage.java:
##########
@@ -51,48 +65,165 @@ protected RepairMessage(RepairJobDesc desc)
this.desc = desc;
}
+ public TimeUUID parentRepairSession()
+ {
+ return desc.parentSessionId;
+ }
+
public interface RepairFailureCallback
{
void onFailure(Exception e);
}
- public static void sendMessageWithFailureCB(RepairMessage request, Verb
verb, InetAddressAndPort endpoint, RepairFailureCallback failureCallback)
+ private static Backoff backoff(SharedContext.MessagingContext ctx, Verb
verb)
+ {
+ RepairRetrySpec.Verb configVerb = toConfigVerb(verb);
+ if (configVerb == null)
+ return Backoff.None.INSTANCE;
+ RepairRetrySpec retrySpec = DatabaseDescriptor.getRepairRetrys();
+ RetrySpec spec = retrySpec.get(configVerb);
+ if (!spec.isEnabled())
+ return Backoff.None.INSTANCE;
+ switch (spec.type)
+ {
+ case Expoential:
+ return new Backoff.ExpoentialBackoff(spec.maxAttempts.value,
spec.baseSleepTime.toMilliseconds(), spec.maxSleepTime.toMilliseconds(),
ctx.random().get()::nextDouble);
+ default:
+ throw new IllegalArgumentException("Unknown type: " +
spec.type);
+ }
+ }
+
+ @Nullable
+ private static RepairRetrySpec.Verb toConfigVerb(Verb verb)
{
- RequestCallback<?> callback = new RequestCallback<Object>()
+ switch (verb)
+ {
+ case PREPARE_MSG: return RepairRetrySpec.Verb.PREPARE;
+ case VALIDATION_REQ: return RepairRetrySpec.Verb.VALIDATION_REQ;
+ case VALIDATION_RSP: return RepairRetrySpec.Verb.VALIDATION_RSP;
+ case SYNC_REQ: return RepairRetrySpec.Verb.SYNC_REQ;
+ case SYNC_RSP: return RepairRetrySpec.Verb.SYNC_RSP;
+ case SNAPSHOT_MSG: return RepairRetrySpec.Verb.SNAPSHOT;
+ case CLEANUP_MSG: return RepairRetrySpec.Verb.CLEANUP;
+ default: return null;
+ }
+ }
+
+ public static <T> void
sendMessageWithRetries(SharedContext.MessagingContext ctx, RepairMessage
request, Verb verb, InetAddressAndPort endpoint, RequestCallback<T>
finalCallback)
+ {
+ // 0 means no retries
+ sendMessageWithRetries(ctx, backoff(ctx, verb), request, verb,
endpoint, finalCallback, 0);
+ }
+
+ public static <T> void
sendMessageWithRetries(SharedContext.MessagingContext ctx, RepairMessage
request, Verb verb, InetAddressAndPort endpoint)
Review Comment:
So the places we call this method are the places we used to use
`MessagingService.instance().send(...)` and just fire-and-forget? If those were
correct before, do we have to do anything w/ this callback?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
