You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sa...@apache.org on 2026/06/24 04:41:09 UTC
(cassandra-sidecar) branch trunk updated: CASSSIDECAR-462: Split RestoreJobDiscoverer into a fast status-check loop and a slow slice-discovery loop (#350)
This is an automated email from the ASF dual-hosted git repository.
sarankk pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2cfea840 CASSSIDECAR-462: Split RestoreJobDiscoverer into a fast status-check loop and a slow slice-discovery loop (#350)
2cfea840 is described below
commit 2cfea840910c53ca6055a0a86c954d3de40bb5b8
Author: Mansi Khara <kh...@gmail.com>
AuthorDate: Tue Jun 23 21:41:02 2026 -0700
CASSSIDECAR-462: Split RestoreJobDiscoverer into a fast status-check loop and a slow slice-discovery loop (#350)
Patch by Mansi Khara; reviewed by Yifan Cai, Shailaja Koppu, Saranya Krishnakumar for CASSSIDECAR-462
---
CHANGES.txt | 1 +
conf/sidecar.yaml | 1 +
.../RestoreJobDiscovererStatusCheckIntTest.java | 133 ++++++++++
.../sidecar/config/RestoreJobConfiguration.java | 7 +
.../config/yaml/RestoreJobConfigurationImpl.java | 37 +++
.../sidecar/restore/RestoreJobDiscoverer.java | 264 ++++++++++++++++---
.../sidecar/restore/RestoreJobDiscovererTest.java | 293 +++++++++++++++++++++
7 files changed, 697 insertions(+), 39 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 707df09c..4d0d10e6 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -6,6 +6,7 @@
0.4.0
-----
+ * Split RestoreJobDiscoverer into a fast status-check loop and a slow slice-discovery loop (CASSSIDECAR-462)
* Ability to load Cassandra connection secrets from filesystem (CASSSIDECAR-407)
* Sidecar should wake up immediately on the instance receiving a phase signal instead of waiting for the discovery loop (CASSSIDECAR-454)
* Fix schema initialization race, CI test reporting, and build stability (CASSSIDECAR-468)
diff --git a/conf/sidecar.yaml b/conf/sidecar.yaml
index 5c564f86..37cbf739 100644
--- a/conf/sidecar.yaml
+++ b/conf/sidecar.yaml
@@ -451,6 +451,7 @@ cassandra_input_validation:
blob_restore:
job_discovery_active_loop_delay: 5m
job_discovery_idle_loop_delay: 10m
+ job_discovery_status_check_interval: 1s
job_discovery_recency_days: 5
slice_process_max_concurrency: 20
restore_job_tables_ttl: 90d
diff --git a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererStatusCheckIntTest.java b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererStatusCheckIntTest.java
new file mode 100644
index 00000000..4013c994
--- /dev/null
+++ b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererStatusCheckIntTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.sidecar.restore;
+
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+
+import com.datastax.driver.core.utils.UUIDs;
+import org.apache.cassandra.sidecar.common.data.RestoreJobSecrets;
+import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
+import org.apache.cassandra.sidecar.common.data.StorageCredentials;
+import org.apache.cassandra.sidecar.common.request.data.CreateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
+import org.apache.cassandra.sidecar.common.server.data.QualifiedTableName;
+import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.config.yaml.RestoreJobConfigurationImpl;
+import org.apache.cassandra.sidecar.config.yaml.SidecarConfigurationImpl;
+import org.apache.cassandra.sidecar.db.RestoreJobDatabaseAccessor;
+import org.apache.cassandra.sidecar.testing.SharedClusterIntegrationTestBase;
+
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Verifies that the fast status-check loop owned by {@link RestoreJobDiscoverer} (its non-static inner
+ * {@code StatusCheckTask}) reacts to a restore-job status transition within the configured fast-loop
+ * interval, without waiting for the slow discovery loop.
+ *
+ * <p>The test seeds an in-flight job, runs one discovery pass to populate the in-flight set,
+ * then flips the job's status directly in the {@code sidecar_internal} keyspace and asserts
+ * that the discoverer sees the terminal status (via {@link RestoreJobDiscoverer#inflightJobIds()})
+ * well within the slow-loop interval. The slow loop is configured to a much larger interval so
+ * a positive observation can only be the work of the fast status-check loop.
+ */
+class RestoreJobDiscovererStatusCheckIntTest extends SharedClusterIntegrationTestBase
+{
+ private static final QualifiedTableName JOB_TABLE = new QualifiedTableName("ks", "tbl");
+
+ @Override
+ protected Function<SidecarConfigurationImpl.Builder, SidecarConfigurationImpl.Builder> configurationOverrides()
+ {
+ return builder -> builder.restoreJobConfiguration(
+ RestoreJobConfigurationImpl.builder()
+ .jobDiscoveryStatusCheckInterval(MillisecondBoundConfiguration.parse("200ms"))
+ // Make slow loops effectively never fire during the test so a
+ // positive observation can only be the fast loop reacting.
+ .jobDiscoveryActiveLoopDelay(MillisecondBoundConfiguration.parse("1h"))
+ .jobDiscoveryIdleLoopDelay(MillisecondBoundConfiguration.parse("1h"))
+ .build());
+ }
+
+ @Override
+ protected void initializeSchemaForTest()
+ {
+ // Restore job rows live in sidecar_internal; no user schema setup required.
+ }
+
+ @Override
+ protected void beforeTestStart()
+ {
+ waitForSchemaReady(30, TimeUnit.SECONDS);
+ }
+
+ @Test
+ void testFastLoopDetectsTerminalTransition()
+ {
+ RestoreJobDatabaseAccessor jobAccessor = serverWrapper.injector.getInstance(RestoreJobDatabaseAccessor.class);
+ RestoreJobDiscoverer discoverer = serverWrapper.injector.getInstance(RestoreJobDiscoverer.class);
+
+ UUID jobId = UUIDs.timeBased();
+ long expireAt = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(5);
+ CreateRestoreJobRequestPayload createPayload = CreateRestoreJobRequestPayload
+ .builder(genRestoreJobSecrets(), expireAt)
+ .jobId(jobId)
+ .jobAgent("int-test")
+ .build();
+ jobAccessor.create(createPayload, JOB_TABLE);
+
+ // Run one discovery pass synchronously so the job lands in the discoverer's in-flight set.
+ // We do not wait for the configured slow loop to fire (it's pinned to 1h above).
+ discoverer.tryExecuteDiscovery();
+ assertThat(discoverer.inflightJobIds())
+ .describedAs("Discovery should track the newly created CREATED job as in-flight")
+ .contains(jobId);
+
+ // Simulate another sidecar (or the storage system) marking the job SUCCEEDED in the shared DB.
+ jobAccessor.update(UpdateRestoreJobRequestPayload.builder()
+ .withStatus(RestoreJobStatus.SUCCEEDED)
+ .build(),
+ jobId);
+
+ // The slow loop is effectively disabled, so this can only succeed via the 200ms fast loop.
+ loopAssert(10, 200, () -> assertThat(discoverer.inflightJobIds())
+ .describedAs("Fast status-check loop should detect the terminal transition")
+ .doesNotContain(jobId));
+ }
+
+ private static RestoreJobSecrets genRestoreJobSecrets()
+ {
+ return new RestoreJobSecrets(genStorageCredentials("read"), genStorageCredentials("write"));
+ }
+
+ private static StorageCredentials genStorageCredentials(String permission)
+ {
+ long nonce = ThreadLocalRandom.current().nextLong();
+ return StorageCredentials.builder()
+ .accessKeyId(permission + "-accessKeyId-" + nonce)
+ .secretAccessKey(permission + "-secretAccessKey-" + nonce)
+ .sessionToken(permission + "-sessionToken-" + nonce)
+ .region(permission + "-region-" + nonce)
+ .build();
+ }
+}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/RestoreJobConfiguration.java b/server/src/main/java/org/apache/cassandra/sidecar/config/RestoreJobConfiguration.java
index dbbb9f03..e80a45e7 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/config/RestoreJobConfiguration.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/config/RestoreJobConfiguration.java
@@ -37,6 +37,13 @@ public interface RestoreJobConfiguration
*/
MillisecondBoundConfiguration jobDiscoveryIdleLoopDelay();
+ /**
+ * @return the interval between cheap status-only checks of known in-flight restore jobs.
+ * The check reacts on phase transitions (STAGE_READY, IMPORT_READY, terminal) without
+ * waiting for the next full {@link #jobDiscoveryActiveLoopDelay()} / {@link #jobDiscoveryIdleLoopDelay()} cycle.
+ */
+ MillisecondBoundConfiguration jobDiscoveryStatusCheckInterval();
+
/**
* @return the minimum number of days in the past to look up the restore jobs
*/
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/RestoreJobConfigurationImpl.java b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/RestoreJobConfigurationImpl.java
index 8ef95bf1..d4e97f8a 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/RestoreJobConfigurationImpl.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/RestoreJobConfigurationImpl.java
@@ -42,6 +42,11 @@ public class RestoreJobConfigurationImpl implements RestoreJobConfiguration
MillisecondBoundConfiguration.parse("5m");
private static final MillisecondBoundConfiguration DEFAULT_JOB_DISCOVERY_IDLE_LOOP_DELAY =
MillisecondBoundConfiguration.parse("10m");
+ // Caps the wake-up latency on peer Sidecars for a phase transition: the Sidecar that
+ // receives the HTTP phase signal reacts immediately, peers learn via DB point-read on
+ // this interval (vs. the 5m/10m slow loop). Costs N reads/sec where N is in-flight jobs.
+ private static final MillisecondBoundConfiguration DEFAULT_JOB_DISCOVERY_STATUS_CHECK_INTERVAL =
+ MillisecondBoundConfiguration.parse("1s");
private static final int DEFAULT_JOB_DISCOVERY_MINIMUM_RECENCY_DAYS = 5;
private static final int DEFAULT_PROCESS_MAX_CONCURRENCY = 20; // process at most 20 slices concurrently
private static final SecondBoundConfiguration DEFAULT_RESTORE_JOB_TABLES_TTL =
@@ -56,6 +61,8 @@ public class RestoreJobConfigurationImpl implements RestoreJobConfiguration
protected MillisecondBoundConfiguration jobDiscoveryIdleLoopDelay;
+ protected MillisecondBoundConfiguration jobDiscoveryStatusCheckInterval;
+
@JsonProperty(value = "job_discovery_minimum_recency_days")
protected final int jobDiscoveryMinimumRecencyDays;
@@ -79,6 +86,7 @@ public class RestoreJobConfigurationImpl implements RestoreJobConfiguration
{
this.jobDiscoveryActiveLoopDelay = builder.jobDiscoveryActiveLoopDelay;
this.jobDiscoveryIdleLoopDelay = builder.jobDiscoveryIdleLoopDelay;
+ this.jobDiscoveryStatusCheckInterval = builder.jobDiscoveryStatusCheckInterval;
this.jobDiscoveryMinimumRecencyDays = builder.jobDiscoveryMinimumRecencyDays;
this.processMaxConcurrency = builder.processMaxConcurrency;
this.restoreJobTablesTtl = builder.restoreJobTablesTtl;
@@ -163,6 +171,22 @@ public class RestoreJobConfigurationImpl implements RestoreJobConfiguration
setJobDiscoveryIdleLoopDelay(new MillisecondBoundConfiguration(jobDiscoveryIdleLoopDelayMillis, TimeUnit.MILLISECONDS));
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @JsonProperty(value = "job_discovery_status_check_interval")
+ public MillisecondBoundConfiguration jobDiscoveryStatusCheckInterval()
+ {
+ return jobDiscoveryStatusCheckInterval;
+ }
+
+ @JsonProperty(value = "job_discovery_status_check_interval")
+ public void setJobDiscoveryStatusCheckInterval(MillisecondBoundConfiguration jobDiscoveryStatusCheckInterval)
+ {
+ this.jobDiscoveryStatusCheckInterval = jobDiscoveryStatusCheckInterval;
+ }
+
/**
* {@inheritDoc}
*/
@@ -314,6 +338,7 @@ public class RestoreJobConfigurationImpl implements RestoreJobConfiguration
private SecondBoundConfiguration slowTaskReportDelay = DEFAULT_RESTORE_JOB_SLOW_TASK_REPORT_DELAY;
private MillisecondBoundConfiguration jobDiscoveryActiveLoopDelay = DEFAULT_JOB_DISCOVERY_ACTIVE_LOOP_DELAY;
private MillisecondBoundConfiguration jobDiscoveryIdleLoopDelay = DEFAULT_JOB_DISCOVERY_IDLE_LOOP_DELAY;
+ private MillisecondBoundConfiguration jobDiscoveryStatusCheckInterval = DEFAULT_JOB_DISCOVERY_STATUS_CHECK_INTERVAL;
private int jobDiscoveryMinimumRecencyDays = DEFAULT_JOB_DISCOVERY_MINIMUM_RECENCY_DAYS;
private int processMaxConcurrency = DEFAULT_PROCESS_MAX_CONCURRENCY;
private SecondBoundConfiguration restoreJobTablesTtl = DEFAULT_RESTORE_JOB_TABLES_TTL;
@@ -353,6 +378,18 @@ public class RestoreJobConfigurationImpl implements RestoreJobConfiguration
return update(b -> b.jobDiscoveryIdleLoopDelay = jobDiscoveryIdleLoopDelay);
}
+ /**
+ * Sets the {@code jobDiscoveryStatusCheckInterval} and returns a reference to this Builder enabling
+ * method chaining.
+ *
+ * @param jobDiscoveryStatusCheckInterval the {@code jobDiscoveryStatusCheckInterval} to set
+ * @return a reference to this Builder
+ */
+ public Builder jobDiscoveryStatusCheckInterval(MillisecondBoundConfiguration jobDiscoveryStatusCheckInterval)
+ {
+ return update(b -> b.jobDiscoveryStatusCheckInterval = jobDiscoveryStatusCheckInterval);
+ }
+
/**
* Sets the {@code jobDiscoveryMinimumRecencyDays} and returns a reference to this Builder enabling
* method chaining.
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
index 73c6cabe..7bfd73e5 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
@@ -27,7 +27,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
@@ -86,9 +86,26 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
private final RestoreMetrics metrics;
private final JobIdsByDay jobIdsByDay;
private final RingTopologyRefresher ringTopologyRefresher;
- private final AtomicBoolean isExecuting = new AtomicBoolean(false);
- private String localDatacenter = null;
- private int inflightJobsCount = 0;
+ /**
+ * Single mutual-exclusion gate for restore-job discovery. The slow loop ({@link #tryExecuteDiscovery})
+ * acquires it via {@link ReentrantLock#lock()} — it blocks briefly if a fast-loop tick or
+ * {@link #processJobNow} is in progress, but it never skips. The fast {@link StatusCheckTask}
+ * and {@link #processJobNow} acquire it via {@link ReentrantLock#tryLock()} — they skip if the
+ * slow loop is in. This encodes the priority asymmetry (slow = correctness, fast = best-effort)
+ * directly at each call site and keeps {@link JobIdsByDay} accessed by exactly one thread at a
+ * time without any internal locking.
+ */
+ private final ReentrantLock executionLock = new ReentrantLock();
+ private final StatusCheckTask statusCheckTask = new StatusCheckTask();
+ // Volatile because it is read by the periodic-task scheduler thread (via scheduleDecision /
+ // hasInflightJobs / delay) without holding executionLock. Written under the lock at the end of
+ // every slow- and fast-loop pass to the freshly recounted in-flight job count. Reads outside
+ // the lock can be slightly stale (one pass behind) but never torn or drifted.
+ private volatile int inflightJobsCount = 0;
+ // Volatile because initLocalDatacenterMaybe writes it without holding the lock (the JMX/CQL
+ // call may block, so we keep it off the critical section). Idempotent — concurrent writers
+ // would assign the same value.
+ private volatile String localDatacenter = null;
private int jobDiscoveryRecencyDays;
private PeriodicTaskExecutor periodicTaskExecutor;
@@ -149,6 +166,9 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
{
this.periodicTaskExecutor = executor;
PeriodicTask.super.deploy(vertx, executor);
+ // Co-deploy the fast status-check loop. It reads only job.status for already-known
+ // in-flight jobs and reacts to phase transitions without waiting for the slow loop.
+ executor.schedule(statusCheckTask);
}
@Override
@@ -176,43 +196,31 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
}
/**
- * Try to execute the job discovery task in the blocking way. It returns immediately, if another thread is executing already.
- * The method can be invoked by external call-sites.
+ * Run the slow discovery pass synchronously, blocking until {@link #executionLock} is free.
+ * Unlike a fast-loop tick or a {@link #processJobNow} wake-up, the slow pass never skips —
+ * it is the correctness guarantee for restore-job discovery.
*/
public void tryExecuteDiscovery()
{
- if (!isExecuting.compareAndSet(false, true))
- {
- LOGGER.debug("Another thread is executing the restore job discovery already. Skipping...");
- return;
- }
-
+ executionLock.lock();
try
{
executeInternal();
}
finally
{
- isExecuting.set(false);
+ executionLock.unlock();
}
}
private boolean shouldSkip()
{
- boolean shouldSkip = !sidecarSchema.isInitialized();
- if (shouldSkip)
+ if (!sidecarSchema.isInitialized())
{
LOGGER.trace("Skipping restore job discovering due to sidecarSchema not initialized");
+ return true;
}
- boolean executing = isExecuting.get();
- shouldSkip = shouldSkip || executing;
- if (executing)
- {
- LOGGER.trace("Skipping restore job discovering due to overlapping execution of this task");
- }
-
- // skip the task
- return shouldSkip;
+ return false;
}
private void executeInternal()
@@ -221,10 +229,8 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
LOGGER.debug("Discovering restore jobs. " +
"inflightJobsCount={} jobDiscoveryRecencyDays={}",
- inflightJobsCount, delay(), jobDiscoveryRecencyDays);
+ inflightJobsCount, jobDiscoveryRecencyDays);
- // reset in-flight jobs
- inflightJobsCount = 0;
RunContext context = new RunContext();
List<RestoreJob> restoreJobs = restoreJobDatabaseAccessor.findAllRecent(context.nowMillis, jobDiscoveryRecencyDays);
RestoreJobManagerGroup restoreJobManagers = restoreJobManagerGroupSingleton.get();
@@ -242,6 +248,9 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
jobIdsByDay.cleanupMaybe();
// resize to the earliestInDays with the minimum days as defined in jobDiscoveryMinimumRecencyDays
jobDiscoveryRecencyDays = Math.max(context.earliestInDays, restoreJobConfig.jobDiscoveryMinimumRecencyDays());
+ // Recount and publish under the lock so unsynchronized readers (scheduleDecision /
+ // hasInflightJobs) observe a consistent post-pass value.
+ inflightJobsCount = jobIdsByDay.inflightJobIds().size();
LOGGER.info("Exit job discovery. " +
"inflightJobsCount={} " +
"delay={} " +
@@ -252,6 +261,69 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
metrics.activeJobs.metric.setValue(inflightJobsCount);
}
+ /**
+ * Snapshot of restore job IDs currently considered in-flight (non-terminal). Acquires
+ * {@link #executionLock} so external callers (tests, the fast loop) can use it without
+ * holding the gate themselves; reentrant for callers already inside the lock.
+ */
+ Set<UUID> inflightJobIds()
+ {
+ executionLock.lock();
+ try
+ {
+ return jobIdsByDay.inflightJobIds();
+ }
+ finally
+ {
+ executionLock.unlock();
+ }
+ }
+
+ /**
+ * Applies the same transition handling as a full discovery pass, but only for a single freshly-read job.
+ * Called by the fast status-check loop when it detects a status change on a known in-flight job.
+ * A no-op when the job's status matches the last-known status cached during discovery.
+ */
+ void handleStatusTransition(RestoreJob currentJob)
+ {
+ executionLock.lock();
+ try
+ {
+ int day = currentJob.createdAt.getDaysSinceEpoch();
+ RestoreJobStatus previousStatus = jobIdsByDay.getKnownStatus(currentJob.jobId, day);
+ if (previousStatus == null)
+ {
+ // Job isn't tracked yet; let the next full discovery pass pick it up so that
+ // the in-flight set and derived state stay consistent.
+ return;
+ }
+ if (previousStatus == currentJob.status)
+ {
+ return;
+ }
+
+ RunContext context = new RunContext();
+ RestoreJobManagerGroup managers = restoreJobManagerGroupSingleton.get();
+ try
+ {
+ processOneJob(currentJob, managers, context);
+ }
+ catch (Exception e)
+ {
+ LOGGER.warn("Exception on processing status transition. jobId={} previousStatus={} newStatus={}",
+ currentJob.jobId, previousStatus, currentJob.status, e);
+ }
+ // Recount and publish under the lock so the volatile counter and the gauge reflect
+ // the fast-loop transition immediately, without waiting for the next slow-loop pass.
+ inflightJobsCount = jobIdsByDay.inflightJobIds().size();
+ metrics.activeJobs.metric.setValue(inflightJobsCount);
+ }
+ finally
+ {
+ executionLock.unlock();
+ }
+ }
+
@Override
public void onRingTopologyChanged(String keyspace, TokenRangeReplicasResponse oldTopology, TokenRangeReplicasResponse newTopology)
{
@@ -348,7 +420,6 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
context.earliestInDays = Math.max(context.earliestInDays, delta(context.today, job.createdAt));
restoreJobManagers.updateRestoreJob(job);
processSidecarManagedJobMaybe(job);
- inflightJobsCount += 1;
break;
case FAILED:
case ABORTED:
@@ -550,6 +621,15 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
return Math.abs(date1.getDaysSinceEpoch() - date2.getDaysSinceEpoch());
}
+ /**
+ * Per-day cache of the latest known {@link RestoreJobStatus} for each job, plus a per-day
+ * set of jobs whose slices have already been discovered.
+ *
+ * <p>Intentionally <b>not</b> internally synchronized. All callers must hold
+ * {@link RestoreJobDiscoverer#executionLock}; that single outer gate is the sole source of
+ * thread safety for this state. The asymmetric {@code lock()}/{@code tryLock()} pattern at
+ * the call sites guarantees exactly one thread is inside this class at a time.
+ */
static class JobIdsByDay
{
private final Map<Integer, Map<UUID, RestoreJobStatus>> jobsByDay = new HashMap<>();
@@ -593,6 +673,12 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
.contains(job.jobId);
}
+ RestoreJobStatus getKnownStatus(UUID jobId, int day)
+ {
+ Map<UUID, RestoreJobStatus> jobs = jobsByDay.get(day);
+ return jobs == null ? null : jobs.get(jobId);
+ }
+
void unsetSlicesDiscovered(RestoreJob job)
{
int day = populateDiscoveredDay(job);
@@ -614,6 +700,26 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
discoveredDays.clear();
}
+ /**
+ * Snapshot of jobIds whose latest known status is non-terminal. Caller must hold
+ * {@link RestoreJobDiscoverer#executionLock}.
+ */
+ Set<UUID> inflightJobIds()
+ {
+ Set<UUID> result = new HashSet<>();
+ for (Map<UUID, RestoreJobStatus> jobs : jobsByDay.values())
+ {
+ for (Map.Entry<UUID, RestoreJobStatus> entry : jobs.entrySet())
+ {
+ if (!entry.getValue().isFinal())
+ {
+ result.add(entry.getKey());
+ }
+ }
+ }
+ return result;
+ }
+
private int populateDiscoveredDay(RestoreJob job)
{
int day = job.createdAt.getDaysSinceEpoch();
@@ -634,23 +740,18 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
* 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.
*
- * <p>The body is gated by the same {@link #isExecuting} CAS used by
- * {@link #tryExecuteDiscovery()} so the discovery loop and the wake-up path are mutually
- * exclusive. The CAS provides the happens-before edge that makes the otherwise non-thread-safe
- * {@link JobIdsByDay} state safe across both entry points. If a slow-loop pass is already
- * running when a wake-up arrives, the wake-up is skipped — the slow loop or the next wake-up
- * will pick the transition up.
+ * <p>Acquires {@link #executionLock} via {@link ReentrantLock#tryLock()}: if the slow loop or
+ * the fast loop is already in, the wake-up is skipped — the slow loop reads the DB freshly
+ * on its next pass and will pick up the same transition, so dropping the wake-up is benign.
*
* @param restoreJob the restore job to process immediately
*/
public void processJobNow(RestoreJob restoreJob)
{
- // Resolve local DC outside the CAS gate; it can block on JMX/CQL and the assignment itself is atomic.
initLocalDatacenterMaybe();
- if (!isExecuting.compareAndSet(false, true))
+ if (!executionLock.tryLock())
{
- LOGGER.debug("Another thread is executing the restore job discovery already. Skipping wake-up. jobId={}",
- restoreJob.jobId);
+ LOGGER.debug("Discovery is already running. Skipping wake-up. jobId={}", restoreJob.jobId);
return;
}
try
@@ -661,7 +762,7 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
}
finally
{
- isExecuting.set(false);
+ executionLock.unlock();
}
}
@@ -677,6 +778,12 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
return jobDiscoveryRecencyDays;
}
+ @VisibleForTesting
+ StatusCheckTask statusCheckTask()
+ {
+ return statusCheckTask;
+ }
+
static class RunContext
{
long nowMillis = System.currentTimeMillis();
@@ -685,4 +792,83 @@ public class RestoreJobDiscoverer implements PeriodicTask, RingTopologyChangeLis
int abortedJobs = 0;
int expiredJobs = 0;
}
+
+ /**
+ * Fast status-check loop that complements the slow {@link RestoreJobDiscoverer} discovery loop.
+ *
+ * <p>Runs on {@link RestoreJobConfiguration#jobDiscoveryStatusCheckInterval()} (default ~1s) and
+ * performs cheap point-reads of {@code job.status} for the in-flight jobs the discoverer already knows
+ * about. When a status transition is detected the task delegates to
+ * {@link RestoreJobDiscoverer#handleStatusTransition(RestoreJob)} so peer Sidecar instances react to
+ * phase signals without waiting for the slow full-scan loop. The slow loop remains the correctness
+ * and recovery guarantee (new-job discovery, expired-job aborts, missed signals).
+ *
+ * <p>Implemented as a non-static inner class so it has direct access to the discoverer's state and
+ * shares its lifecycle — it isn't an independent component.
+ */
+ class StatusCheckTask implements PeriodicTask
+ {
+ @Override
+ public DurationSpec delay()
+ {
+ return restoreJobConfig.jobDiscoveryStatusCheckInterval();
+ }
+
+ @Override
+ public ScheduleDecision scheduleDecision()
+ {
+ if (!sidecarSchema.isInitialized())
+ {
+ return ScheduleDecision.SKIP;
+ }
+ if (inflightJobsCount == 0)
+ {
+ return ScheduleDecision.SKIP;
+ }
+ return ScheduleDecision.EXECUTE;
+ }
+
+ @Override
+ public void execute(Promise<Void> promise)
+ {
+ // Resolve local DC once per tick before acquiring the lock — the JMX/CQL call may
+ // block, and resolving it inside handleStatusTransition would re-attempt the call
+ // for every in-flight job whenever Cassandra is unavailable.
+ initLocalDatacenterMaybe();
+ // tryLock — skip this tick if the slow loop or processJobNow is in. Best-effort by
+ // design; the next tick (or the slow loop's next pass) will pick up any transition we
+ // miss.
+ if (!executionLock.tryLock())
+ {
+ promise.tryComplete();
+ return;
+ }
+
+ try
+ {
+ for (UUID jobId : jobIdsByDay.inflightJobIds())
+ {
+ try
+ {
+ RestoreJob current = restoreJobDatabaseAccessor.find(jobId);
+ if (current == null)
+ {
+ continue;
+ }
+ handleStatusTransition(current);
+ }
+ catch (Exception e)
+ {
+ // Do not fail the whole pass on one job; the slow loop will retry it.
+ LOGGER.warn("Exception on status check for jobId={}", jobId, e);
+ }
+ }
+ }
+ finally
+ {
+ executionLock.unlock();
+ promise.tryComplete();
+ }
+ }
+ }
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
index 9eec31fc..a0e946d6 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
@@ -25,6 +25,7 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -438,6 +439,298 @@ class RestoreJobDiscovererTest
assertThat(captured.jobId).isEqualTo(jobId);
}
+ @Test
+ void testInflightJobIdsReturnsNonTerminalJobsAfterDiscovery()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ UUID activeJobId = UUIDs.timeBased();
+ UUID terminalJobId = UUIDs.timeBased();
+ when(mockJobAccessor.findAllRecent(anyLong(), anyInt())).thenReturn(List.of(
+ RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(activeJobId))
+ .jobId(activeJobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.CREATED)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build(),
+ RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(terminalJobId))
+ .jobId(terminalJobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.SUCCEEDED)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build()));
+ executeBlocking();
+
+ assertThat(loop.inflightJobIds()).containsOnly(activeJobId);
+ }
+
+ @Test
+ void testHandleStatusTransitionIsNoOpWhenJobNotTracked()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ UUID jobId = UUIDs.timeBased();
+ RestoreJob job = RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(jobId))
+ .jobId(jobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.STAGE_READY)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build();
+
+ loop.handleStatusTransition(job);
+
+ verify(mockManagers, never()).updateRestoreJob(any());
+ verify(mockManagers, never()).removeJobInternal(any());
+ }
+
+ @Test
+ void testHandleStatusTransitionIsNoOpWhenStatusUnchanged()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ UUID jobId = UUIDs.timeBased();
+ RestoreJob initial = RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(jobId))
+ .jobId(jobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.CREATED)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build();
+ when(mockJobAccessor.findAllRecent(anyLong(), anyInt())).thenReturn(List.of(initial));
+ executeBlocking();
+ Mockito.reset(mockManagers);
+
+ loop.handleStatusTransition(initial);
+
+ verify(mockManagers, never()).updateRestoreJob(any());
+ verify(mockManagers, never()).removeJobInternal(any());
+ }
+
+ @Test
+ void testHandleStatusTransitionFinalizesOnTerminalTransition()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ UUID jobId = UUIDs.timeBased();
+ RestoreJob active = RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(jobId))
+ .jobId(jobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.IMPORT_READY)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build();
+ when(mockJobAccessor.findAllRecent(anyLong(), anyInt())).thenReturn(List.of(active));
+ executeBlocking();
+ Mockito.reset(mockManagers);
+
+ RestoreJob succeeded = active.unbuild().jobStatus(RestoreJobStatus.SUCCEEDED).build();
+ loop.handleStatusTransition(succeeded);
+
+ verify(mockManagers).removeJobInternal(succeeded);
+ }
+
+ @Test
+ void testHandleStatusTransitionUpdatesManagersOnActiveTransition()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ UUID jobId = UUIDs.timeBased();
+ RestoreJob created = RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(jobId))
+ .jobId(jobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.CREATED)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build();
+ when(mockJobAccessor.findAllRecent(anyLong(), anyInt())).thenReturn(List.of(created));
+ executeBlocking();
+ Mockito.reset(mockManagers);
+
+ RestoreJob stageReady = created.unbuild().jobStatus(RestoreJobStatus.STAGE_READY).build();
+ loop.handleStatusTransition(stageReady);
+
+ verify(mockManagers).updateRestoreJob(stageReady);
+ }
+
+ @Test
+ void testStatusCheckTaskSkipsWhenNoInflightJobs()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ assertThat(loop.statusCheckTask().scheduleDecision()).isEqualTo(ScheduleDecision.SKIP);
+ }
+
+ @Test
+ void testStatusCheckTaskSkipsWhenSchemaNotInitialized()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(false);
+ assertThat(loop.statusCheckTask().scheduleDecision()).isEqualTo(ScheduleDecision.SKIP);
+ }
+
+ @Test
+ void testStatusCheckTaskExecutesAgainstInflightJobs()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ UUID jobId = UUIDs.timeBased();
+ RestoreJob created = RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(jobId))
+ .jobId(jobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.CREATED)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build();
+ when(mockJobAccessor.findAllRecent(anyLong(), anyInt())).thenReturn(List.of(created));
+ executeBlocking();
+ assertThat(loop.statusCheckTask().scheduleDecision()).isEqualTo(ScheduleDecision.EXECUTE);
+
+ // Flip the DB-side status; the task should point-read each in-flight job and dispatch the transition
+ RestoreJob stageReady = created.unbuild().jobStatus(RestoreJobStatus.STAGE_READY).build();
+ when(mockJobAccessor.find(jobId)).thenReturn(stageReady);
+ Mockito.reset(mockManagers);
+
+ Promise<Void> promise = Promise.promise();
+ loop.statusCheckTask().execute(promise);
+
+ verify(mockJobAccessor).find(jobId);
+ verify(mockManagers).updateRestoreJob(stageReady);
+ assertThat(promise.future().succeeded()).isTrue();
+ }
+
+ @Test
+ void testStatusCheckTaskTolerantOfMissingOrFailingJobs()
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ UUID missingJobId = UUIDs.timeBased();
+ UUID failingJobId = UUIDs.timeBased();
+ RestoreJob missingJob = RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(missingJobId))
+ .jobId(missingJobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.CREATED)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build();
+ RestoreJob failingJob = RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(failingJobId))
+ .jobId(failingJobId)
+ .jobAgent("agent")
+ .jobStatus(RestoreJobStatus.CREATED)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build();
+ when(mockJobAccessor.findAllRecent(anyLong(), anyInt())).thenReturn(List.of(missingJob, failingJob));
+ executeBlocking();
+ when(mockJobAccessor.find(missingJobId)).thenReturn(null); // job vanished
+ when(mockJobAccessor.find(failingJobId)).thenThrow(new RuntimeException("db down")); // transient error
+
+ Promise<Void> promise = Promise.promise();
+ loop.statusCheckTask().execute(promise);
+
+ assertThat(promise.future().succeeded())
+ .describedAs("a missing or failing job should not abort the whole pass")
+ .isTrue();
+ }
+
+ // Hammers the slow discovery loop and the fast status-check task concurrently against
+ // shared JobIdsByDay state, with statuses flipping under their feet, to verify that
+ // per-method synchronization on JobIdsByDay is sufficient. Both loops mutate the map
+ // through processOneJob via different gates (isExecuting vs checkerExecuting), so they
+ // can interleave on every operation. The test asserts no exception leaks out of either
+ // loop and that the in-flight view converges to the latest committed state once both
+ // settle.
+ @Test
+ void testSlowAndFastLoopRaceDoesNotCorruptState() throws Exception
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
+ UUID a = UUIDs.timeBased();
+ UUID b = UUIDs.timeBased();
+ UUID c = UUIDs.timeBased();
+ AtomicReference<RestoreJobStatus> sa = new AtomicReference<>(RestoreJobStatus.CREATED);
+ AtomicReference<RestoreJobStatus> sb = new AtomicReference<>(RestoreJobStatus.STAGE_READY);
+ AtomicReference<RestoreJobStatus> sc = new AtomicReference<>(RestoreJobStatus.IMPORT_READY);
+
+ when(mockJobAccessor.findAllRecent(anyLong(), anyInt())).thenAnswer(inv ->
+ List.of(jobWith(a, sa.get()), jobWith(b, sb.get()), jobWith(c, sc.get())));
+ when(mockJobAccessor.find(a)).thenAnswer(inv -> jobWith(a, sa.get()));
+ when(mockJobAccessor.find(b)).thenAnswer(inv -> jobWith(b, sb.get()));
+ when(mockJobAccessor.find(c)).thenAnswer(inv -> jobWith(c, sc.get()));
+
+ // Prime the in-flight set so the fast loop has work on its first run.
+ executeBlocking();
+
+ int iterations = 200;
+ AtomicReference<Throwable> err = new AtomicReference<>();
+ CountDownLatch start = new CountDownLatch(1);
+ Thread slow = new Thread(() -> {
+ try
+ {
+ start.await();
+ for (int i = 0; i < iterations && err.get() == null; i++)
+ {
+ executeBlocking();
+ }
+ }
+ catch (Throwable t)
+ {
+ err.compareAndSet(null, t);
+ }
+ }, "race-slow-loop");
+ Thread fast = new Thread(() -> {
+ try
+ {
+ start.await();
+ for (int i = 0; i < iterations && err.get() == null; i++)
+ {
+ if ((i & 1) == 0)
+ {
+ sa.set(RestoreJobStatus.STAGE_READY);
+ sb.set(RestoreJobStatus.IMPORT_READY);
+ }
+ else
+ {
+ sa.set(RestoreJobStatus.CREATED);
+ sb.set(RestoreJobStatus.STAGE_READY);
+ }
+ Promise<Void> p = Promise.promise();
+ loop.statusCheckTask().execute(p);
+ }
+ }
+ catch (Throwable t)
+ {
+ err.compareAndSet(null, t);
+ }
+ }, "race-fast-loop");
+
+ slow.start();
+ fast.start();
+ start.countDown();
+ slow.join(TimeUnit.SECONDS.toMillis(10));
+ fast.join(TimeUnit.SECONDS.toMillis(10));
+
+ assertThat(err.get())
+ .describedAs("slow and fast loops must not corrupt jobIdsByDay under contention")
+ .isNull();
+
+ // Settle the world. After one more slow pass with all jobs terminal, the in-flight
+ // view must converge to empty regardless of how the race played out.
+ sa.set(RestoreJobStatus.SUCCEEDED);
+ sb.set(RestoreJobStatus.SUCCEEDED);
+ sc.set(RestoreJobStatus.SUCCEEDED);
+ executeBlocking();
+ assertThat(loop.inflightJobIds())
+ .describedAs("in-flight set converges after settling all jobs to SUCCEEDED")
+ .isEmpty();
+ assertThat(metrics.server().restore().activeJobs.metric.getValue())
+ .describedAs("active-jobs gauge converges to 0 after settling")
+ .isZero();
+ }
+
+ private static RestoreJob jobWith(UUID id, RestoreJobStatus status)
+ {
+ return RestoreJob.builder()
+ .createdAt(RestoreJob.toLocalDate(id))
+ .jobId(id)
+ .jobAgent("agent")
+ .jobStatus(status)
+ .expireAt(new Date(System.currentTimeMillis() + 10000L))
+ .build();
+ }
+
private RestoreJobConfiguration testConfig()
{
RestoreJobConfiguration restoreJobConfiguration = mock(RestoreJobConfiguration.class);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org