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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
