mansikhara commented on code in PR #346:
URL: https://github.com/apache/cassandra-sidecar/pull/346#discussion_r3277482944
##########
server/src/main/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandler.java:
##########
@@ -87,36 +91,39 @@ protected void handleInternal(RoutingContext context,
{
RoutingContextUtils
.getAsFuture(context, SC_RESTORE_JOB)
- .compose(job -> {
- if (job.status.isFinal())
+ .compose(existingJob -> {
+ if (existingJob.status.isFinal())
{
// skip the update, since the job is in the final state already
- logger.debug("The job has completed already. job={}", job);
+ logger.debug("The job has completed already. job={}",
existingJob);
return
Future.failedFuture(wrapHttpException(HttpResponseStatus.CONFLICT,
- "Job is already
in final state: " + job.status));
+ "Job is already
in final state: " + existingJob.status));
}
return executorPools.service()
- .executeBlocking(() ->
restoreJobDatabaseAccessor.update(requestPayload, job.jobId));
+ .executeBlocking(() ->
restoreJobDatabaseAccessor.update(requestPayload, existingJob));
})
- .onSuccess(job -> {
+ .onSuccess(updatedJob -> {
logger.info("Successfully updated restore job. job={}, request={},
remoteAddress={}, instance={}",
- job, requestPayload, remoteAddress, host);
- if (job.status == RestoreJobStatus.SUCCEEDED)
+ updatedJob, requestPayload, remoteAddress, host);
+ if (updatedJob.status == RestoreJobStatus.SUCCEEDED)
{
metrics.successfulJobs.metric.update(1);
- long startMillis = UUIDs.unixTimestamp(job.jobId);
+ long startMillis = UUIDs.unixTimestamp(updatedJob.jobId);
long durationMillis = System.currentTimeMillis() - startMillis;
// toNanos does not overflow. Nanos in `long` can at most
represent 106,751 days.
metrics.jobCompletionTime.metric.update(durationMillis,
TimeUnit.MILLISECONDS);
}
- if (job.secrets != null)
+ if (updatedJob.secrets != null)
{
metrics.tokenRefreshed.metric.update(1);
}
context.response().setStatusCode(HttpResponseStatus.OK.code()).end();
+ // Fire-and-forget on a worker thread — notifying the restore
system should not
+ // block the event loop or delay the HTTP response.
+ executorPools.service().runBlocking(() ->
notifyPhaseSignalMaybe(updatedJob));
Review Comment:
done!
##########
server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java:
##########
@@ -628,6 +628,22 @@ Map<Integer, Map<UUID, RestoreJobStatus>> jobsByDay()
}
}
+ /**
+ * Immediately processes a restore job without waiting for the next
discovery loop iteration.
+ * Called by UpdateRestoreJobHandler after a phase signal (STAGE_READY) is
written to DB.
+ * This is safe to call concurrently with the discovery loop — the DB
write is the durable
+ * source of truth, and duplicate processing is deduplicated by existing
idempotency checks.
+ *
+ * @param restoreJob the restore job to process immediately
+ */
+ public void processJobNow(RestoreJob restoreJob)
+ {
+ initLocalDatacenterMaybe();
+ RestoreJobManagerGroup restoreJobManagers =
restoreJobManagerGroupSingleton.get();
+ restoreJobManagers.updateRestoreJob(restoreJob);
+ processSidecarManagedJobMaybe(restoreJob);
Review Comment:
done!
--
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]
