This is an automated email from the ASF dual-hosted git repository.
bbotella 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 e6b3ec9b3 CASSSIDECAR-508: Add local job execution and storage support
for coordinated cluster-wide operations (#392)
e6b3ec9b3 is described below
commit e6b3ec9b3945e4dc4702e227b7c941a4cc994bab
Author: Andrés Beck-Ruiz <[email protected]>
AuthorDate: Tue Oct 27 10:23:19 2026 +0000
CASSSIDECAR-508: Add local job execution and storage support for
coordinated cluster-wide operations (#392)
Updates operational job storage and tracking so that cluster-wide
operational jobs can be coordinated in CASSSIDECAR-378. Also adds LocalJob and
LocalJobManager, which the local coordinator uses to run individual node jobs
that are part of a larger cluster-wide operation.
The local job coordinator (StorageBackedLocalJobCoordinator) that uses
these changes is in this CASSSIDECAR-378 stacked PR.
Patch by Andrés Beck-Ruiz; Reviewed by Bernardo Botella, , Paulo Motta,
Arjun Ashok for CASSSIDECAR-508
---
CHANGES.txt | 1 +
.../handlers/ListOperationalJobsHandler.java | 39 ++--
.../sidecar/job/DurableOperationalJobTracker.java | 97 +++++++-
.../sidecar/job/InMemoryOperationalJobTracker.java | 14 ++
.../org/apache/cassandra/sidecar/job/LocalJob.java | 133 +++++++++++
.../cassandra/sidecar/job/LocalJobManager.java | 153 +++++++++++++
.../cassandra/sidecar/job/OperationalJob.java | 30 +++
.../sidecar/job/OperationalJobCoordinator.java | 4 +-
.../sidecar/job/OperationalJobManager.java | 6 +-
.../sidecar/job/OperationalJobTracker.java | 8 +
.../job/storage/CassandraStorageProvider.java | 40 ++--
.../sidecar/job/storage/OperationalJobRecord.java | 13 +-
.../sidecar/job/storage/StorageProvider.java | 7 +-
.../sidecar/modules/OperationalJobsModule.java | 8 +-
...urableOperationalJobTrackerIntegrationTest.java | 1 -
.../job/DurableOperationalJobTrackerTest.java | 247 ++++++++++++++++++++-
.../job/InMemoryOperationalJobTrackerTest.java | 17 ++
.../cassandra/sidecar/job/LocalJobManagerTest.java | 176 +++++++++++++++
.../apache/cassandra/sidecar/job/LocalJobTest.java | 94 ++++++++
.../sidecar/job/OperationalJobManagerTest.java | 12 +
.../cassandra/sidecar/job/OperationalJobTest.java | 98 ++++++++
.../job/storage/CassandraStorageProviderTest.java | 117 ++++++++++
22 files changed, 1254 insertions(+), 61 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 9987d8b5f..509e6e422 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Add local job execution and storage support for coordinated cluster-wide
operations (CASSSIDECAR-508)
* Use describe statement to fetch schema instead of Java driver
(CASSSIDECAR-505)
* Make the CDC CassandraBridge injectable (CASSSIDECAR-511)
* Introduce server-side configuration to enforce allowed credential modes for
restore jobs (CASSSIDECAR-501)
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandler.java
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandler.java
index ca8407031..57e655501 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandler.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandler.java
@@ -68,22 +68,27 @@ public class ListOperationalJobsHandler extends
AbstractHandler<Void> implements
@Override
protected void handleInternal(RoutingContext context, HttpServerRequest
httpRequest, @NotNull String host, SocketAddress remoteAddress, Void request)
{
- ListOperationalJobsResponse listResponse = new
ListOperationalJobsResponse();
- jobManager.allInflightJobs()
- .stream()
- .map(job -> OperationalJobResponse.builder()
- .jobId(job.jobId())
- .status(job.status())
- .operation(job.name())
- .startTime(job.startTime())
-
.nodesPending(job.nodesPending())
-
.nodesExecuting(job.nodesExecuting())
-
.nodesSucceeded(job.nodesSucceeded())
-
.nodesFailed(job.nodesFailed())
-
.lastUpdate(job.lastUpdate())
- .build()
- )
- .forEach(listResponse::addJob);
- context.json(listResponse);
+ // Durable tracker fetches active operations from storage, run off
event loop
+ executorPools.service()
+ .executeBlocking(jobManager::allInflightJobs)
+ .onFailure(cause -> processFailure(cause, context, host,
remoteAddress, request))
+ .onSuccess(jobs -> {
+ ListOperationalJobsResponse listResponse = new
ListOperationalJobsResponse();
+ jobs.stream()
+ .map(job -> OperationalJobResponse.builder()
+
.jobId(job.jobId())
+
.status(job.status())
+
.operation(job.name())
+
.startTime(job.startTime())
+
.nodesPending(job.nodesPending())
+
.nodesExecuting(job.nodesExecuting())
+
.nodesSucceeded(job.nodesSucceeded())
+
.nodesFailed(job.nodesFailed())
+
.lastUpdate(job.lastUpdate())
+ .build()
+ )
+ .forEach(listResponse::addJob);
+ context.json(listResponse);
+ });
}
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java
index d5aa8622b..9a4a9902e 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java
@@ -18,8 +18,10 @@
package org.apache.cassandra.sidecar.job;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -38,6 +40,7 @@ import
org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
import org.apache.cassandra.sidecar.config.ServiceConfiguration;
import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.apache.cassandra.sidecar.job.storage.StorageProviderException;
import org.apache.cassandra.sidecar.utils.InvocationTrackingFunction;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -88,28 +91,40 @@ public class DurableOperationalJobTracker implements
OperationalJobTracker
@Override
public OperationalJob computeIfAbsent(UUID jobId, Function<UUID,
OperationalJob> mappingFunction)
{
- if (!storageProvider.isAvailable())
- {
- throw new IllegalStateException("Storage provider is not
available");
- }
-
InvocationTrackingFunction<UUID, OperationalJob>
mappingFunctionTracker =
new InvocationTrackingFunction<>(mappingFunction);
OperationalJob job = liveJobs.computeIfAbsent(jobId,
mappingFunctionTracker);
if (mappingFunctionTracker.wasInvoked())
{
- persistInitialRecord(job, 1);
+ persistInitialRecord(job, job.shouldActivate(), 1);
}
return job;
}
- private void persistInitialRecord(OperationalJob job, int attempt)
+ private void persistInitialRecord(OperationalJob job, boolean
activatingCoordinatedJob, int attempt)
{
- executor.runBlocking(() ->
storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job)))
+ executor.runBlocking(() -> {
+ if (activatingCoordinatedJob)
+ {
+ seedNodeStatuses(job);
+ // Currently only support job submission that executes
immediately
+
storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job,
OperationalJobStatus.RUNNING,
+
Instant.now()));
+ }
+ else
+ {
+
storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job));
+ }
+ })
.onSuccess(v -> job.asyncResult().onComplete(ar -> {
- updateTerminalStatus(job);
+ // The coordinator owns updating the terminal status of
jobs that do not release
+ // the active operation lock on completion
+ if (job.releasesOnCompletion())
+ {
+ updateTerminalStatus(job);
+ }
liveJobs.remove(job.jobId());
}))
.onFailure(e -> {
@@ -120,7 +135,8 @@ public class DurableOperationalJobTracker implements
OperationalJobTracker
// Add jitter so that concurrent sidecar processes
retrying against the same transient
// Cassandra blip do not all retry in lockstep
long delay = RETRY_DELAY_MS * attempt +
ThreadLocalRandom.current().nextLong(RETRY_JITTER_MS);
- executor.setTimer(delay, id ->
persistInitialRecord(job, attempt + 1));
+ executor.setTimer(delay,
+ id -> persistInitialRecord(job,
activatingCoordinatedJob, attempt + 1));
}
else
{
@@ -170,6 +186,67 @@ public class DurableOperationalJobTracker implements
OperationalJobTracker
.collect(Collectors.toList());
}
+ @NotNull
+ @Override
+ public List<OperationalJobInfo> inflightJobs()
+ {
+ Map<UUID, OperationalJobInfo> merged = new LinkedHashMap<>(liveJobs);
+ if (storageProvider.isAvailable())
+ {
+ mergeActiveOperationsFromStorage(merged);
+ }
+
+ return merged.values()
+ .stream()
+ .filter(job -> !job.status().isCompleted())
+ .collect(Collectors.toList());
+ }
+
+ private void mergeActiveOperationsFromStorage(Map<UUID,
OperationalJobInfo> merged)
+ {
+ try
+ {
+ // Coordinated jobs are removed from liveJobs almost immediately
after handoff but might be still
+ // running. Therefore, we fetch those active operations from
storage
+ for (UUID activeJobId :
storageProvider.getActiveOperations().values())
+ {
+ if (merged.containsKey(activeJobId))
+ {
+ continue;
+ }
+ OperationalJobRecord record =
storageProvider.findJob(activeJobId);
+ if (record != null)
+ {
+ merged.put(activeJobId, enrichWithNodeStatuses(record));
+ }
+ }
+ }
+ catch (StorageProviderException e)
+ {
+ LOGGER.warn("Failed to read active operations from storage.
Returning in-memory jobs only", e);
+ }
+ }
+
+ /**
+ * Adds a {@link OperationalJobStatus#CREATED} for every node in the job's
execution order.
+ */
+ private void seedNodeStatuses(OperationalJob job)
+ {
+ List<List<UUID>> executionOrder = job.nodeExecutionOrder();
+ if (executionOrder == null)
+ {
+ return;
+ }
+ List<UUID> nodeIds = executionOrder.stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ if (nodeIds.isEmpty())
+ {
+ return;
+ }
+ storageProvider.updateNodeStatuses(job.jobId(), nodeIds,
OperationalJobStatus.CREATED);
+ }
+
/**
* Enriches an {@link OperationalJobRecord} with per-node status data from
storage.
* If the record already has non-empty node lists (e.g. populated by a
storage provider
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/InMemoryOperationalJobTracker.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/InMemoryOperationalJobTracker.java
index bbdd4777b..866cde23a 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/InMemoryOperationalJobTracker.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/InMemoryOperationalJobTracker.java
@@ -129,6 +129,20 @@ public class InMemoryOperationalJobTracker implements
OperationalJobTracker
.collect(Collectors.toList());
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @NotNull
+ public List<OperationalJobInfo> inflightJobs()
+ {
+ return jobsView().values()
+ .stream()
+ .filter(j -> !j.status().isCompleted())
+ .map(j -> (OperationalJobInfo) j)
+ .collect(Collectors.toList());
+ }
+
@VisibleForTesting
OperationalJob put(OperationalJob job)
{
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/LocalJob.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/LocalJob.java
new file mode 100644
index 000000000..608e9aa7f
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/LocalJob.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.job;
+
+import java.util.UUID;
+
+import io.vertx.core.Future;
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+
+/**
+ * Represents a job executed on a local Cassandra node as part of a
cluster-wide operation.
+ */
+public abstract class LocalJob
+{
+ private final UUID operationId;
+ private final UUID nodeId;
+ private final String instanceHost;
+ private final OperationType operationType;
+ private volatile OperationalJobStatus status;
+ private volatile boolean cancelled;
+
+ /**
+ * @param operationId the cluster-wide operation this job belongs to
+ * @param nodeId the host ID of the Cassandra node
+ * @param instanceHost the hostname or IP address of the Cassandra
instance
+ * @param operationType the type of operation being performed
+ */
+ protected LocalJob(UUID operationId, UUID nodeId, String instanceHost,
OperationType operationType)
+ {
+ this.operationId = operationId;
+ this.nodeId = nodeId;
+ this.instanceHost = instanceHost;
+ this.operationType = operationType;
+ this.status = OperationalJobStatus.CREATED;
+ }
+
+ /**
+ * Performs the job's work. Called on a worker thread, so implementations
may block.
+ */
+ protected abstract void executeInternal();
+
+ /**
+ * Requests cancellation of this job. The running task is not interrupted.
Instead, a long-running
+ * {@link #executeInternal()} is expected to check {@link #isCancelled()}
at safe points and stop
+ * and recover when cancelled.
+ */
+ public void cancel()
+ {
+ cancelled = true;
+ }
+
+ /**
+ * @return whether cancellation has been requested for this job
+ */
+ protected boolean isCancelled()
+ {
+ return cancelled;
+ }
+
+ /**
+ * Performs reconciliation when the overall cluster-wide job terminates
with FAILED
+ * status.
+ *
+ * @return a future that completes when reconciliation is done
+ */
+ public Future<Void> onJobFailed()
+ {
+ return Future.succeededFuture();
+ }
+
+ /**
+ * Execute the job, tracking status transitions
+ *
+ * @return null (for compatibility with {@code Callable} in {@code
executeBlocking})
+ */
+ public Void execute()
+ {
+ status = OperationalJobStatus.RUNNING;
+ try
+ {
+ executeInternal();
+ status = OperationalJobStatus.SUCCEEDED;
+ }
+ catch (Exception e)
+ {
+ status = OperationalJobStatus.FAILED;
+ throw e;
+ }
+ return null;
+ }
+
+ public UUID operationId()
+ {
+ return operationId;
+ }
+
+ public UUID nodeId()
+ {
+ return nodeId;
+ }
+
+ public String instanceHost()
+ {
+ return instanceHost;
+ }
+
+ public OperationType operationType()
+ {
+ return operationType;
+ }
+
+ public OperationalJobStatus status()
+ {
+ return status;
+ }
+}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/LocalJobManager.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/LocalJobManager.java
new file mode 100644
index 000000000..bc91be543
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/LocalJobManager.java
@@ -0,0 +1,153 @@
+/*
+ * 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.job;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.Future;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Manages the lifecycle of {@link LocalJob} instances for this Sidecar.
+ */
+@Singleton
+public class LocalJobManager
+{
+ private final Map<OperationNodeKey, LocalJob> activeJobs = new
ConcurrentHashMap<>();
+ private final TaskExecutorPool internalExecutorPool;
+
+ @Inject
+ public LocalJobManager(ExecutorPools executorPools)
+ {
+ this.internalExecutorPool = executorPools.internal();
+ }
+
+ /**
+ * Submit a {@link LocalJob} for execution
+ *
+ * @param job the job to submit
+ * @return a future that completes when the job finishes, or succeeds
immediately if already tracked
+ */
+ public Future<Void> submitJob(LocalJob job)
+ {
+ LocalJob existing = activeJobs.putIfAbsent(new
OperationNodeKey(job.operationId(), job.nodeId()), job);
+ if (existing != null)
+ {
+ return Future.succeededFuture();
+ }
+ return internalExecutorPool.executeBlocking(job::execute).mapEmpty();
+ }
+
+ /**
+ * Get the active job for a node within a specific operation, or {@code
null} if none.
+ *
+ * @param operationId the cluster-wide operation identifier
+ * @param nodeId the node identifier
+ * @return the active job, or {@code null}
+ */
+ @Nullable
+ public LocalJob getJob(UUID operationId, UUID nodeId)
+ {
+ return activeJobs.get(new OperationNodeKey(operationId, nodeId));
+ }
+
+ /**
+ * Get all active local jobs.
+ *
+ * @return an unmodifiable view of active jobs
+ */
+ public Collection<LocalJob> activeJobs()
+ {
+ return Collections.unmodifiableCollection(activeJobs.values());
+ }
+
+ /**
+ * Remove a job from tracking, only if it is still the tracked one
+ *
+ * @param operationId the cluster-wide operation identifier
+ * @param nodeId the node whose job should be removed
+ * @param expected the job the caller is finished with
+ * @return whether the job was removed
+ */
+ public boolean removeJob(UUID operationId, UUID nodeId, LocalJob expected)
+ {
+ return activeJobs.remove(new OperationNodeKey(operationId, nodeId),
expected);
+ }
+
+ /**
+ * Composite key identifying per-node state within a specific cluster-wide
operation
+ */
+ static final class OperationNodeKey
+ {
+ private final UUID operationId;
+ private final UUID nodeId;
+
+ OperationNodeKey(UUID operationId, UUID nodeId)
+ {
+ this.operationId = operationId;
+ this.nodeId = nodeId;
+ }
+
+ UUID operationId()
+ {
+ return operationId;
+ }
+
+ UUID nodeId()
+ {
+ return nodeId;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (!(o instanceof OperationNodeKey))
+ {
+ return false;
+ }
+ OperationNodeKey that = (OperationNodeKey) o;
+ return operationId.equals(that.operationId) &&
nodeId.equals(that.nodeId);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(operationId, nodeId);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "OperationNodeKey{operationId=" + operationId + ", nodeId="
+ nodeId + '}';
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java
index bdfa377c8..d086450f7 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java
@@ -22,6 +22,7 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import org.slf4j.Logger;
@@ -201,6 +202,35 @@ public abstract class OperationalJob implements
Task<Void>, OperationalJobInfo
return true;
}
+ /**
+ * @return the ordered list of parallel node groups for execution, or null
if not applicable
+ */
+ @Nullable
+ public List<List<UUID>> nodeExecutionOrder()
+ {
+ return null;
+ }
+
+ /**
+ * Seed node statuses and mark the job as RUNNING, only for a coordinated
job that successfully started.
+ * A job that failed to start (e.g. it did not acquire the active
operation lock) is already complete.
+ */
+ final boolean shouldActivate()
+ {
+ return requiresCoordination()
+ && nodeExecutionOrder() != null
+ && !asyncResult().isComplete();
+ }
+
+ /**
+ * @return metadata associated with this operation, or null if not
applicable
+ */
+ @Nullable
+ public Map<String, String> operationMetadata()
+ {
+ return null;
+ }
+
@Override
public final Void result()
{
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobCoordinator.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobCoordinator.java
index 90be1a3f8..3e52943c8 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobCoordinator.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobCoordinator.java
@@ -34,12 +34,12 @@ public interface OperationalJobCoordinator
{
/**
* Attempt to set an operation as active. Implementations must ensure
mutual exclusion
- * so that only one operation of the given type is active at a time.
+ * so that only one operation of the given type is active at a time within
the local datacenter.
*
* @param operationType the type of operation
* @param operationId the unique identifier for this operation
* @return {@code true} if the operation was successfully set as active,
- * {@code false} if an operation of the same type is already active
+ * {@code false} if an operation of the same type is already
active in the local datacenter
*/
boolean trySetActive(OperationType operationType, UUID operationId);
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java
index c348c86ae..4d1c1206a 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java
@@ -21,7 +21,6 @@ package org.apache.cassandra.sidecar.job;
import java.util.List;
import java.util.UUID;
import java.util.function.BiConsumer;
-import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -71,10 +70,7 @@ public class OperationalJobManager
*/
public List<OperationalJobInfo> allInflightJobs()
{
- return jobTracker.jobsView().values()
- .stream()
- .filter(j -> !j.status().isCompleted())
- .collect(Collectors.toList());
+ return jobTracker.inflightJobs();
}
/**
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobTracker.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobTracker.java
index 5dcb4283c..bed0f10f1 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobTracker.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobTracker.java
@@ -70,4 +70,12 @@ public interface OperationalJobTracker
*/
@NotNull
List<OperationalJob> inflightJobsByOperation(String operation);
+
+ /**
+ * Returns all inflight (CREATED or RUNNING) jobs tracked by this Sidecar.
+ *
+ * @return list of inflight job info
+ */
+ @NotNull
+ List<OperationalJobInfo> inflightJobs();
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/storage/CassandraStorageProvider.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/storage/CassandraStorageProvider.java
index 47781845b..e9667154a 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/storage/CassandraStorageProvider.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/storage/CassandraStorageProvider.java
@@ -18,22 +18,21 @@
package org.apache.cassandra.sidecar.job.storage;
-import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.function.Supplier;
-import com.datastax.driver.core.Host;
-import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.DriverException;
import org.apache.cassandra.sidecar.common.data.OperationType;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.response.NodeSettings;
import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
import org.apache.cassandra.sidecar.db.ActiveClusterOpsDatabaseAccessor;
import org.apache.cassandra.sidecar.db.ClusterOpsDatabaseAccessor;
import org.apache.cassandra.sidecar.db.ClusterOpsNodeStateDatabaseAccessor;
import org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -48,6 +47,7 @@ public class CassandraStorageProvider implements
StorageProvider
private final ClusterOpsDatabaseAccessor clusterOpsAccessor;
private final ClusterOpsNodeStateDatabaseAccessor nodeStateAccessor;
private final ActiveClusterOpsDatabaseAccessor activeOpsAccessor;
+ private final InstanceMetadataFetcher instanceMetadataFetcher;
private volatile String clusterName;
private volatile String datacenter;
@@ -61,12 +61,14 @@ public class CassandraStorageProvider implements
StorageProvider
ClusterOpsDatabaseAccessor
clusterOpsAccessor,
ClusterOpsNodeStateDatabaseAccessor
nodeStateAccessor,
ActiveClusterOpsDatabaseAccessor
activeOpsAccessor,
+ InstanceMetadataFetcher
instanceMetadataFetcher,
@Nullable String clusterName)
{
this.sessionProvider = sessionProvider;
this.clusterOpsAccessor = clusterOpsAccessor;
this.nodeStateAccessor = nodeStateAccessor;
this.activeOpsAccessor = activeOpsAccessor;
+ this.instanceMetadataFetcher = instanceMetadataFetcher;
this.clusterName = clusterName;
}
@@ -176,20 +178,13 @@ public class CassandraStorageProvider implements
StorageProvider
// TTL on all tables handles record pruning.
try
{
- Session session = sessionProvider.get();
if (clusterName == null)
{
- clusterName =
session.getCluster().getMetadata().getClusterName();
+ clusterName =
sessionProvider.get().getCluster().getMetadata().getClusterName();
}
if (datacenter == null)
{
- Collection<Host> connectedHosts =
session.getState().getConnectedHosts();
- if (connectedHosts.isEmpty())
- {
- throw new StorageProviderException(
- "Failed to resolve local datacenter: no connected hosts
available");
- }
- datacenter = connectedHosts.iterator().next().getDatacenter();
+ datacenter = resolveLocalDatacenter();
}
}
catch (CassandraUnavailableException e)
@@ -198,6 +193,23 @@ public class CassandraStorageProvider implements
StorageProvider
}
}
+ /**
+ * Resolves this Sidecar's local datacenter from the managed instance's
node settings
+ */
+ @NotNull
+ private String resolveLocalDatacenter()
+ {
+ NodeSettings nodeSettings =
instanceMetadataFetcher.callOnFirstAvailableInstance(
+ instance -> instance.delegate().nodeSettings());
+ String nodeSettingsDatacenter = nodeSettings.datacenter();
+ if (nodeSettingsDatacenter == null || nodeSettingsDatacenter.isEmpty())
+ {
+ throw new StorageProviderException("Failed to resolve local
datacenter for operational job coordination: "
+ + "node settings did not
provide a datacenter");
+ }
+ return nodeSettingsDatacenter;
+ }
+
@Override
public boolean isAvailable()
{
@@ -216,13 +228,13 @@ public class CassandraStorageProvider implements
StorageProvider
{
if (clusterName == null || datacenter == null)
{
- throw new StorageProviderException("StorageProvider has not been
initialized. Call initialize() first.");
+ initialize();
}
try
{
return action.get();
}
- catch (DriverException e)
+ catch (DriverException | CassandraUnavailableException e)
{
throw new StorageProviderException("Failed to execute " +
operation, e);
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java
index 42761634b..f66f8ca04 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java
@@ -229,11 +229,22 @@ public class OperationalJobRecord implements
OperationalJobInfo
* @return a new record capturing the job's current state
*/
public static OperationalJobRecord fromOperationalJob(OperationalJob job)
+ {
+ return fromOperationalJob(job, job.status(), job.startTime());
+ }
+
+ public static OperationalJobRecord fromOperationalJob(OperationalJob job,
+ OperationalJobStatus
status,
+ @Nullable Instant
startTime)
{
return builder().jobId(job.jobId())
.operationType(job.operationType())
- .status(job.status())
+ .status(status)
+ .startTime(startTime)
.lastUpdate(Instant.now())
+ .failureReason(job.failureReason())
+ .nodeExecutionOrder(job.nodeExecutionOrder())
+ .operationMetadata(job.operationMetadata())
.build();
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/storage/StorageProvider.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/storage/StorageProvider.java
index 0f931bc5e..3309ef2d0 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/storage/StorageProvider.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/storage/StorageProvider.java
@@ -89,17 +89,18 @@ public interface StorageProvider extends Closeable
// --- Active Operation Coordination ---
/**
- * Set an operation as active if no other operation of the same type is
currently active.
+ * Set an operation as active if no other operation of the same type is
currently active
+ * in the local datacenter.
* <p>
* Implementations must provide compare-and-set (CAS) semantics to ensure
only one active
- * operation of a given type runs at a time across the cluster. This lock
is per operation
+ * operation of a given type runs at a time within the local datacenter.
This lock is per operation
* plan, not per node; within a single operation, multiple nodes may be
operated on
* concurrently as defined by the node execution order.
*
* @param operationType the operation type
* @param operationId the unique identifier for this operation
* @return {@code true} if the operation was successfully set as active,
{@code false} if
- * an operation of the same type is already active (including the
same operation ID)
+ * an operation of the same type is already active in the local
datacenter
*/
boolean trySetActiveOperation(OperationType operationType, UUID
operationId);
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/modules/OperationalJobsModule.java
b/server/src/main/java/org/apache/cassandra/sidecar/modules/OperationalJobsModule.java
index 820e7c388..cfaa1d488 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/modules/OperationalJobsModule.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/modules/OperationalJobsModule.java
@@ -35,6 +35,7 @@ import
org.apache.cassandra.sidecar.job.storage.CassandraStorageProvider;
import org.apache.cassandra.sidecar.job.storage.StorageProvider;
import org.apache.cassandra.sidecar.modules.multibindings.KeyClassMapKey;
import org.apache.cassandra.sidecar.modules.multibindings.TableSchemaMapKeys;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
/**
* Guice module for operational job storage, providing schema registrations
@@ -71,9 +72,10 @@ public class OperationalJobsModule extends AbstractModule
StorageProvider cassandraStorageProvider(CQLSessionProvider
sessionProvider,
ClusterOpsDatabaseAccessor
clusterOpsAccessor,
ClusterOpsNodeStateDatabaseAccessor nodeStateAccessor,
- ActiveClusterOpsDatabaseAccessor
activeOpsAccessor)
+ ActiveClusterOpsDatabaseAccessor
activeOpsAccessor,
+ InstanceMetadataFetcher
instanceMetadataFetcher)
{
- return new CassandraStorageProvider(sessionProvider,
- clusterOpsAccessor,
nodeStateAccessor, activeOpsAccessor, null);
+ return new CassandraStorageProvider(sessionProvider,
clusterOpsAccessor, nodeStateAccessor,
+ activeOpsAccessor,
instanceMetadataFetcher, null);
}
}
diff --git
a/server/src/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java
b/server/src/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java
index 155161d9c..5b9a5c69c 100644
---
a/server/src/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java
+++
b/server/src/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java
@@ -47,7 +47,6 @@ class DurableOperationalJobTrackerIntegrationTest extends
IntegrationTestBase
waitForSchemaReady(10, TimeUnit.SECONDS);
StorageProvider storageProvider =
injector.getInstance(StorageProvider.class);
- storageProvider.initialize();
ExecutorPools executorPools =
injector.getInstance(ExecutorPools.class);
TaskExecutorPool executorPool = executorPools.internal();
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java
index 58158a78b..f8acd3c3e 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java
@@ -42,11 +42,13 @@ import
org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
import org.apache.cassandra.sidecar.job.storage.StorageProvider;
import org.apache.cassandra.sidecar.job.storage.StorageProviderException;
+import org.mockito.InOrder;
import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.CREATED;
import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.RUNNING;
import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.SUCCEEDED;
+import static
org.apache.cassandra.sidecar.job.OperationalJobTest.createCoordinatedOperationalJob;
import static
org.apache.cassandra.sidecar.job.OperationalJobTest.createOperationalJob;
import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
import static org.assertj.core.api.Assertions.assertThat;
@@ -57,6 +59,7 @@ import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -324,14 +327,31 @@ class DurableOperationalJobTrackerTest
}
@Test
- void testComputeIfAbsentFailsWhenStorageUnavailable()
+ void
testComputeIfAbsentDegradesForNonCoordinatedJobWhenStorageUnavailable()
{
- when(storageProvider.isAvailable()).thenReturn(false);
+ doThrow(new StorageProviderException("Storage
unavailable")).when(storageProvider).persistJob(any());
OperationalJob job = createOperationalJob(CREATED);
- assertThatThrownBy(() -> tracker.computeIfAbsent(job.jobId(), id ->
job))
- .isInstanceOf(IllegalStateException.class)
- .hasMessage("Storage provider is not available");
+ OperationalJob result = tracker.computeIfAbsent(job.jobId(), id ->
job);
+
+ assertThat(result).isSameAs(job);
+ loopAssert(2, () -> verify(storageProvider,
times(3)).persistJob(any()));
+ assertThat(tracker.jobsView()).containsKey(job.jobId());
+ }
+
+ @Test
+ void testFailedToStartCoordinatedJobIsNotSeededOrMarkedRunning()
+ {
+ UUID jobId = UUIDs.timeBased();
+ UUID node = UUIDs.timeBased();
+ OperationalJob job = createCoordinatedOperationalJob(jobId,
List.of(List.of(node)));
+ job.failToStart(new RuntimeException("An active operation already
exists"));
+
+ tracker.computeIfAbsent(jobId, id -> job);
+
+ loopAssert(2, () -> verify(storageProvider).persistJob(any()));
+ verify(storageProvider, never()).updateNodeStatuses(any(), any(),
any());
+ verify(storageProvider, never()).updateJobStatus(any(), any(),
eq(RUNNING), any());
}
@Test
@@ -439,6 +459,96 @@ class DurableOperationalJobTrackerTest
.hasMessage("Node state unavailable");
}
+ @Test
+ void
testCoordinatedJobThatDoesNotReleaseOnCompletionSkipsTerminalStatusUpdate()
throws InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = createCoordinatedOperationalJob(jobId,
List.of(List.of(UUIDs.timeBased())));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ assertThat(tracker.jobsView()).doesNotContainKey(jobId);
+ verify(storageProvider, never()).updateJobStatus(eq(jobId), any(),
eq(SUCCEEDED), any());
+ });
+ }
+
+ @Test
+ void testCoordinatedJobThatReleasesOnCompletionUpdatesTerminalStatus()
throws InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = createCoordinatedOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DRAIN), eq(SUCCEEDED), isNull());
+ });
+ }
+
+ @Test
+ void testNonCoordinatedJobStillUpdatesTerminalStatus() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DRAIN), eq(SUCCEEDED), isNull());
+ });
+ }
+
+ @Test
+ void testCoordinatedJobSeedsNodeStatusesBeforePersistingRunningRecord()
+ {
+ UUID jobId = UUIDs.timeBased();
+ UUID node1 = UUIDs.timeBased();
+ UUID node2 = UUIDs.timeBased();
+ UUID node3 = UUIDs.timeBased();
+ OperationalJob job = createCoordinatedOperationalJob(jobId,
+
List.of(List.of(node1, node2), List.of(node3)));
+
+ tracker.computeIfAbsent(jobId, id -> job);
+
+ loopAssert(2, () -> verify(storageProvider).persistJob(any()));
+ InOrder inOrder = inOrder(storageProvider);
+ inOrder.verify(storageProvider).updateNodeStatuses(jobId,
List.of(node1, node2, node3), CREATED);
+ inOrder.verify(storageProvider).persistJob(argThat(record ->
record.status() == RUNNING
+ &&
record.startTime() != null));
+ verify(storageProvider, never()).updateJobStatus(any(), any(), any(),
any());
+ }
+
+ @Test
+ void testNonCoordinatedJobDoesNotSeedNodeStatuses()
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ tracker.computeIfAbsent(jobId, id -> job);
+
+ loopAssert(2, () -> {
+ verify(storageProvider).persistJob(any());
+ verify(storageProvider, never()).updateNodeStatuses(any(), any(),
any());
+ verify(storageProvider, never()).updateJobStatus(any(), any(),
eq(RUNNING), any());
+ });
+ }
+
@Test
void testOnCompleteRetrySucceedsAfterTransientFailure() throws
InterruptedException
{
@@ -524,4 +634,131 @@ class DurableOperationalJobTrackerTest
assertThat(tracker.jobsView()).doesNotContainKey(jobId);
});
}
+
+ @Test
+ void testInflightJobsMergesLiveJobsWithActiveStorageRecords()
+ {
+ UUID coordinatedJobId = UUIDs.timeBased();
+ OperationalJobRecord record = OperationalJobRecord.builder()
+
.jobId(coordinatedJobId)
+
.operationType(OperationType.DRAIN)
+ .status(RUNNING)
+ .build();
+
when(storageProvider.getActiveOperations()).thenReturn(Map.of(OperationType.DRAIN,
coordinatedJobId));
+ when(storageProvider.findJob(coordinatedJobId)).thenReturn(record);
+
when(storageProvider.getNodeStatusesForOperation(coordinatedJobId)).thenReturn(Collections.emptyMap());
+
+ OperationalJob liveJob = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(liveJob.jobId(), id -> liveJob);
+
+ List<OperationalJobInfo> inflight = tracker.inflightJobs();
+
+ assertThat(inflight).hasSize(2);
+ assertThat(inflight).extracting(OperationalJobInfo::jobId)
+ .containsExactlyInAnyOrder(liveJob.jobId(),
coordinatedJobId);
+ }
+
+ @Test
+ void testInflightJobsPrefersLiveJobAndSkipsStorageLookupForSameId()
+ {
+ UUID jobId = UUIDs.timeBased();
+ UUID node = UUIDs.timeBased();
+
when(storageProvider.getActiveOperations()).thenReturn(Map.of(OperationType.DRAIN,
jobId));
+ OperationalJob liveJob = createCoordinatedOperationalJob(jobId,
List.of(List.of(node)));
+ tracker.computeIfAbsent(jobId, id -> liveJob);
+
+ List<OperationalJobInfo> inflight = tracker.inflightJobs();
+
+ assertThat(inflight).hasSize(1);
+ assertThat(inflight.get(0)).isSameAs(liveJob);
+ verify(storageProvider, never()).findJob(jobId);
+ }
+
+ @Test
+ void testInflightJobsEnrichesStorageRecordsWithNodeStatuses()
+ {
+ UUID coordinatedJobId = UUIDs.timeBased();
+ UUID pendingNode = UUIDs.timeBased();
+ UUID runningNode = UUIDs.timeBased();
+ OperationalJobRecord record = OperationalJobRecord.builder()
+
.jobId(coordinatedJobId)
+
.operationType(OperationType.DRAIN)
+ .status(RUNNING)
+ .build();
+
when(storageProvider.getActiveOperations()).thenReturn(Map.of(OperationType.DRAIN,
coordinatedJobId));
+ when(storageProvider.findJob(coordinatedJobId)).thenReturn(record);
+ Map<UUID, OperationalJobStatus> nodeStatuses = new HashMap<>();
+ nodeStatuses.put(pendingNode, CREATED);
+ nodeStatuses.put(runningNode, RUNNING);
+
when(storageProvider.getNodeStatusesForOperation(coordinatedJobId)).thenReturn(nodeStatuses);
+
+ List<OperationalJobInfo> inflight = tracker.inflightJobs();
+
+ assertThat(inflight).hasSize(1);
+
assertThat(inflight.get(0).nodesPending()).containsExactly(pendingNode);
+
assertThat(inflight.get(0).nodesExecuting()).containsExactly(runningNode);
+ }
+
+ @Test
+ void testInflightJobsExcludesCompletedLiveJobs()
+ {
+ OperationalJob completedJob = createOperationalJob(SUCCEEDED);
+ tracker.computeIfAbsent(completedJob.jobId(), id -> completedJob);
+
+ List<OperationalJobInfo> inflight = tracker.inflightJobs();
+
+ assertThat(inflight).isEmpty();
+ }
+
+ @Test
+ void testInflightJobsSkipsActiveOperationMissingFromStorage()
+ {
+ UUID coordinatedJobId = UUIDs.timeBased();
+
when(storageProvider.getActiveOperations()).thenReturn(Map.of(OperationType.DRAIN,
coordinatedJobId));
+ when(storageProvider.findJob(coordinatedJobId)).thenReturn(null);
+
+ List<OperationalJobInfo> inflight = tracker.inflightJobs();
+
+ assertThat(inflight).isEmpty();
+ verify(storageProvider).findJob(coordinatedJobId);
+ }
+
+ @Test
+ void testInflightJobsReturnsLiveJobsWhenStorageUnavailable()
+ {
+ when(storageProvider.isAvailable()).thenReturn(false);
+ OperationalJob liveJob = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(liveJob.jobId(), id -> liveJob);
+
+ List<OperationalJobInfo> inflight = tracker.inflightJobs();
+
+ assertThat(inflight).containsExactly(liveJob);
+ verify(storageProvider, never()).getActiveOperations();
+ }
+
+ @Test
+ void testInflightJobsReturnsLiveJobsWhenGetActiveOperationsFails()
+ {
+ when(storageProvider.getActiveOperations()).thenThrow(new
StorageProviderException("read failed"));
+ OperationalJob liveJob = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(liveJob.jobId(), id -> liveJob);
+
+ List<OperationalJobInfo> inflight = tracker.inflightJobs();
+
+ assertThat(inflight).containsExactly(liveJob);
+ }
+
+ @Test
+ void testInflightJobsReturnsLiveJobsWhenFindJobFails()
+ {
+ UUID coordinatedJobId = UUIDs.timeBased();
+
when(storageProvider.getActiveOperations()).thenReturn(Map.of(OperationType.DRAIN,
coordinatedJobId));
+ when(storageProvider.findJob(coordinatedJobId)).thenThrow(new
StorageProviderException("read failed"));
+ OperationalJob liveJob = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(liveJob.jobId(), id -> liveJob);
+
+ List<OperationalJobInfo> inflight = tracker.inflightJobs();
+
+ assertThat(inflight).containsExactly(liveJob);
+ }
}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/job/InMemoryOperationalJobTrackerTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/job/InMemoryOperationalJobTrackerTest.java
index cf5e5f590..59316302d 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/job/InMemoryOperationalJobTrackerTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/job/InMemoryOperationalJobTrackerTest.java
@@ -229,4 +229,21 @@ class InMemoryOperationalJobTrackerTest
assertThat(inflightJobs).isEmpty();
}
+
+ @Test
+ void testInflightJobsReturnsInflightJobsFromMap()
+ {
+ OperationalJob createdJob = createOperationalJob(CREATED);
+ OperationalJob runningJob = createOperationalJob(RUNNING);
+ jobTracker.put(createdJob);
+ jobTracker.put(runningJob);
+ jobTracker.put(job1);
+
+ List<OperationalJobInfo> inflightJobs = jobTracker.inflightJobs();
+
+ assertThat(inflightJobs)
+ .describedAs("Only CREATED and RUNNING jobs are inflight")
+ .extracting(OperationalJobInfo::jobId)
+ .containsExactlyInAnyOrder(createdJob.jobId(), runningJob.jobId());
+ }
}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/job/LocalJobManagerTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/job/LocalJobManagerTest.java
new file mode 100644
index 000000000..cb166dfa9
--- /dev/null
+++
b/server/src/test/java/org/apache/cassandra/sidecar/job/LocalJobManagerTest.java
@@ -0,0 +1,176 @@
+/*
+ * 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.job;
+
+import java.util.Collection;
+import java.util.UUID;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import org.apache.cassandra.sidecar.TestResourceReaper;
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
+
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link LocalJobManager}
+ */
+class LocalJobManagerTest
+{
+ private static final UUID OPERATION_ID = UUID.randomUUID();
+ private static final UUID NODE_ID_1 = UUID.randomUUID();
+ private static final UUID NODE_ID_2 = UUID.randomUUID();
+
+ private Vertx vertx;
+ private ExecutorPools executorPools;
+ private LocalJobManager manager;
+
+ @BeforeEach
+ void setup()
+ {
+ vertx = Vertx.vertx();
+ executorPools = new ExecutorPools(vertx, new
ServiceConfigurationImpl());
+ manager = new LocalJobManager(executorPools);
+ }
+
+ @AfterEach
+ void cleanup()
+ {
+ TestResourceReaper.create().with(vertx).with(executorPools).close();
+ }
+
+ @Test
+ void testSubmitJobExecutesAsync()
+ {
+ LocalJob job = createTestJob(NODE_ID_1);
+ Future<Void> result = manager.submitJob(job);
+ loopAssert(2, () -> {
+ assertThat(job.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ assertThat(result.succeeded()).isTrue();
+ });
+ }
+
+ @Test
+ void testSubmitDuplicateDoesNotReplaceExisting()
+ {
+ LocalJob first = createTestJob(NODE_ID_1);
+ LocalJob second = createTestJob(NODE_ID_1);
+
+ manager.submitJob(first);
+ Future<Void> duplicate = manager.submitJob(second);
+
+ assertThat(manager.getJob(OPERATION_ID, NODE_ID_1)).isSameAs(first);
+ assertThat(duplicate.succeeded()).isTrue();
+ assertThat(second.status()).isEqualTo(OperationalJobStatus.CREATED);
+ }
+
+ @Test
+ void testGetJobReturnsActiveJob()
+ {
+ LocalJob job = createTestJob(NODE_ID_1);
+ manager.submitJob(job);
+ assertThat(manager.getJob(OPERATION_ID, NODE_ID_1)).isSameAs(job);
+ }
+
+ @Test
+ void testGetJobReturnsNullWhenNone()
+ {
+ assertThat(manager.getJob(OPERATION_ID, NODE_ID_1)).isNull();
+ }
+
+ @Test
+ void testCompletedJobIsStillTrackedUntilRemoveJob()
+ {
+ LocalJob job = createTestJob(NODE_ID_1);
+ manager.submitJob(job);
+ loopAssert(2, () ->
assertThat(job.status()).isEqualTo(OperationalJobStatus.SUCCEEDED));
+
+ assertThat(manager.getJob(OPERATION_ID, NODE_ID_1)).isSameAs(job);
+
+ assertThat(manager.removeJob(OPERATION_ID, NODE_ID_1, job)).isTrue();
+ assertThat(manager.getJob(OPERATION_ID, NODE_ID_1)).isNull();
+ }
+
+ @Test
+ void testRemoveJobWithDuplicateHandleKeepsTrackedJob()
+ {
+ LocalJob tracked = createTestJob(NODE_ID_1);
+ LocalJob duplicate = createTestJob(NODE_ID_1);
+ manager.submitJob(tracked);
+ manager.submitJob(duplicate);
+
+ assertThat(manager.removeJob(OPERATION_ID, NODE_ID_1,
duplicate)).isFalse();
+ assertThat(manager.getJob(OPERATION_ID, NODE_ID_1)).isSameAs(tracked);
+ }
+
+ @Test
+ void testActiveJobsReturnsAllTracked()
+ {
+ LocalJob job1 = createTestJob(NODE_ID_1);
+ LocalJob job2 = createTestJob(NODE_ID_2);
+ manager.submitJob(job1);
+ manager.submitJob(job2);
+
+ Collection<LocalJob> active = manager.activeJobs();
+ assertThat(active).containsExactlyInAnyOrder(job1, job2);
+ }
+
+ @Test
+ void testSameNodeAcrossOperationsTrackedIndependently()
+ {
+ UUID otherOperationId = UUID.randomUUID();
+ LocalJob job1 = createTestJob(OPERATION_ID, NODE_ID_1);
+ LocalJob job2 = createTestJob(otherOperationId, NODE_ID_1);
+
+ manager.submitJob(job1);
+ manager.submitJob(job2);
+
+ assertThat(manager.getJob(OPERATION_ID, NODE_ID_1)).isSameAs(job1);
+ assertThat(manager.getJob(otherOperationId, NODE_ID_1)).isSameAs(job2);
+ assertThat(manager.activeJobs()).containsExactlyInAnyOrder(job1, job2);
+
+ assertThat(manager.removeJob(OPERATION_ID, NODE_ID_1, job1)).isTrue();
+ assertThat(manager.getJob(OPERATION_ID, NODE_ID_1)).isNull();
+ assertThat(manager.getJob(otherOperationId, NODE_ID_1)).isSameAs(job2);
+ }
+
+ private static LocalJob createTestJob(UUID nodeId)
+ {
+ return createTestJob(OPERATION_ID, nodeId);
+ }
+
+ private static LocalJob createTestJob(UUID operationId, UUID nodeId)
+ {
+ return new LocalJob(operationId, nodeId, "127.0.0.1",
OperationType.DRAIN)
+ {
+ @Override
+ protected void executeInternal()
+ {
+ }
+ };
+ }
+}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/job/LocalJobTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/job/LocalJobTest.java
new file mode 100644
index 000000000..6e0457cf7
--- /dev/null
+++ b/server/src/test/java/org/apache/cassandra/sidecar/job/LocalJobTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.job;
+
+import java.util.UUID;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for {@link LocalJob}
+ */
+class LocalJobTest
+{
+ private static final UUID OPERATION_ID = UUID.randomUUID();
+ private static final UUID NODE_ID = UUID.randomUUID();
+ private static final String INSTANCE_HOST = "127.0.0.1";
+ private static final OperationType OPERATION_TYPE = OperationType.DRAIN;
+
+ @Test
+ void testStatusTransitionsOnSuccess()
+ {
+ LocalJob job = createSucceedingJob();
+ assertThat(job.status()).isEqualTo(OperationalJobStatus.CREATED);
+
+ job.execute();
+ assertThat(job.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ }
+
+ @Test
+ void testStatusTransitionsOnFailure()
+ {
+ LocalJob job = createFailingJob();
+ assertThat(job.status()).isEqualTo(OperationalJobStatus.CREATED);
+
+ assertThatThrownBy(job::execute).isInstanceOf(RuntimeException.class);
+ assertThat(job.status()).isEqualTo(OperationalJobStatus.FAILED);
+ }
+
+ @Test
+ void testCancelDoesNotChangeStatus()
+ {
+ LocalJob job = createSucceedingJob();
+
+ job.cancel();
+
+ assertThat(job.isCancelled()).isTrue();
+ assertThat(job.status()).isEqualTo(OperationalJobStatus.CREATED);
+ }
+
+ private static LocalJob createSucceedingJob()
+ {
+ return new LocalJob(OPERATION_ID, NODE_ID, INSTANCE_HOST,
OPERATION_TYPE)
+ {
+ @Override
+ protected void executeInternal()
+ {
+ }
+ };
+ }
+
+ private static LocalJob createFailingJob()
+ {
+ return new LocalJob(OPERATION_ID, NODE_ID, INSTANCE_HOST,
OPERATION_TYPE)
+ {
+ @Override
+ protected void executeInternal()
+ {
+ throw new RuntimeException("test failure");
+ }
+ };
+ }
+}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java
index 797743927..f652db475 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java
@@ -326,6 +326,18 @@ class OperationalJobManagerTest
verify(coordinator, never()).trySetActive(any(), any());
}
+ @Test
+ void testAllInflightJobsDelegatesToTracker()
+ {
+ OperationalJobTracker tracker = mock(OperationalJobTracker.class);
+ List<OperationalJobInfo> inflight =
List.of(mock(OperationalJobInfo.class));
+ when(tracker.inflightJobs()).thenReturn(inflight);
+ OperationalJobManager manager = new OperationalJobManager(tracker, new
DisabledOperationalJobCoordinator(), executorPool);
+
+ assertThat(manager.allInflightJobs()).isEqualTo(inflight);
+ verify(tracker).inflightJobs();
+ }
+
private static OperationalJob createCoordinatedJob(UUID jobId)
{
return new OperationalJob(jobId)
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobTest.java
index 485a64819..31f603c3d 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobTest.java
@@ -129,6 +129,104 @@ class OperationalJobTest
};
}
+ public static OperationalJob createCoordinatedOperationalJob(UUID jobId,
DurationSpec jobDuration)
+ {
+ return createCoordinatedOperationalJob(jobId, jobDuration, null);
+ }
+
+ public static OperationalJob createCoordinatedOperationalJob(UUID jobId,
DurationSpec jobDuration, OperationalJobException jobFailure)
+ {
+ return new OperationalJob(jobId)
+ {
+ @Override
+ public boolean hasConflict(List<OperationalJob> jobs)
+ {
+ return false;
+ }
+
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
+ @Override
+ public boolean requiresCoordination()
+ {
+ return true;
+ }
+
+ @Override
+ protected Future<Void> executeInternal() throws
OperationalJobException
+ {
+ if (jobDuration != null)
+ {
+
Uninterruptibles.sleepUninterruptibly(jobDuration.quantity(),
jobDuration.unit());
+ }
+
+ if (jobFailure != null)
+ {
+ throw jobFailure;
+ }
+ return Future.succeededFuture();
+ }
+
+ @Override
+ public String name()
+ {
+ return "Coordinated Operation";
+ }
+ };
+ }
+
+ public static OperationalJob createCoordinatedOperationalJob(UUID jobId,
List<List<UUID>> nodeExecutionOrder)
+ {
+ return new OperationalJob(jobId)
+ {
+ @Override
+ public boolean hasConflict(List<OperationalJob> jobs)
+ {
+ return false;
+ }
+
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
+ @Override
+ public boolean requiresCoordination()
+ {
+ return true;
+ }
+
+ @Override
+ public boolean releasesOnCompletion()
+ {
+ return false;
+ }
+
+ @Override
+ public List<List<UUID>> nodeExecutionOrder()
+ {
+ return nodeExecutionOrder;
+ }
+
+ @Override
+ protected Future<Void> executeInternal()
+ {
+ return Future.succeededFuture();
+ }
+
+ @Override
+ public String name()
+ {
+ return "Coordinated Operation";
+ }
+ };
+ }
+
public static OperationalJob createOperationalJob(UUID jobId, DurationSpec
jobDuration)
{
return createOperationalJob(jobId, jobDuration, null);
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/job/storage/CassandraStorageProviderTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/job/storage/CassandraStorageProviderTest.java
new file mode 100644
index 000000000..96002efaa
--- /dev/null
+++
b/server/src/test/java/org/apache/cassandra/sidecar/job/storage/CassandraStorageProviderTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.job.storage;
+
+import java.util.UUID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.datastax.driver.core.utils.UUIDs;
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import org.apache.cassandra.sidecar.common.response.NodeSettings;
+import org.apache.cassandra.sidecar.common.server.CQLSessionProvider;
+import org.apache.cassandra.sidecar.db.ActiveClusterOpsDatabaseAccessor;
+import org.apache.cassandra.sidecar.db.ClusterOpsDatabaseAccessor;
+import org.apache.cassandra.sidecar.db.ClusterOpsNodeStateDatabaseAccessor;
+import org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static
org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException.Service.CQL;
+import static
org.apache.cassandra.sidecar.exceptions.CassandraUnavailableException.Service.CQL_AND_JMX;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link CassandraStorageProvider}
+ */
+class CassandraStorageProviderTest
+{
+ private static final String CLUSTER_NAME = "test-cluster";
+ private static final UUID OPERATION_ID = UUIDs.timeBased();
+
+ private ActiveClusterOpsDatabaseAccessor activeOpsAccessor;
+ private InstanceMetadataFetcher instanceMetadataFetcher;
+ private CassandraStorageProvider provider;
+
+ @BeforeEach
+ void setup()
+ {
+ activeOpsAccessor = mock(ActiveClusterOpsDatabaseAccessor.class);
+ instanceMetadataFetcher = mock(InstanceMetadataFetcher.class);
+ provider = new CassandraStorageProvider(mock(CQLSessionProvider.class),
+
mock(ClusterOpsDatabaseAccessor.class),
+
mock(ClusterOpsNodeStateDatabaseAccessor.class),
+ activeOpsAccessor,
+ instanceMetadataFetcher,
+ CLUSTER_NAME);
+ }
+
+ @Test
+ void testResolvesDatacenterFromNodeSettings()
+ {
+
when(instanceMetadataFetcher.<NodeSettings>callOnFirstAvailableInstance(any()))
+ .thenReturn(NodeSettings.builder().datacenter("dc-from-node").build());
+
+ provider.trySetActiveOperation(OperationType.DRAIN, OPERATION_ID);
+
+ verify(activeOpsAccessor).trySetActiveOperation(eq(CLUSTER_NAME),
eq("dc-from-node"),
+
eq(OperationType.DRAIN), eq(OPERATION_ID));
+ }
+
+ @Test
+ void testThrowsWhenNodeSettingsUnavailable()
+ {
+
when(instanceMetadataFetcher.<NodeSettings>callOnFirstAvailableInstance(any()))
+ .thenThrow(new CassandraUnavailableException(CQL_AND_JMX, "node
settings unavailable"));
+
+ assertThatThrownBy(() ->
provider.trySetActiveOperation(OperationType.DRAIN, OPERATION_ID))
+ .isInstanceOf(StorageProviderException.class)
+ .hasMessageContaining("Failed to initialize storage provider");
+ }
+
+ @Test
+ void testThrowsWhenNodeSettingsDatacenterMissing()
+ {
+
when(instanceMetadataFetcher.<NodeSettings>callOnFirstAvailableInstance(any()))
+ .thenReturn(NodeSettings.builder().build());
+
+ assertThatThrownBy(() ->
provider.trySetActiveOperation(OperationType.DRAIN, OPERATION_ID))
+ .isInstanceOf(StorageProviderException.class)
+ .hasMessageContaining("Failed to resolve local datacenter");
+ }
+
+ @Test
+ void testWrapsCassandraUnavailableFromAccessor()
+ {
+
when(instanceMetadataFetcher.<NodeSettings>callOnFirstAvailableInstance(any()))
+ .thenReturn(NodeSettings.builder().datacenter("dc-from-node").build());
+ when(activeOpsAccessor.getActiveOperations(CLUSTER_NAME,
"dc-from-node"))
+ .thenThrow(new CassandraUnavailableException(CQL, "session
unavailable"));
+
+ assertThatThrownBy(() -> provider.getActiveOperations())
+ .isInstanceOf(StorageProviderException.class)
+ .hasMessageContaining("Failed to execute getActiveOperations")
+ .hasCauseInstanceOf(CassandraUnavailableException.class);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
