This is an automated email from the ASF dual-hosted git repository.
frankgh 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 951d091a CASSSIDECAR-374: Implement durable operational job tracker
(#359)
951d091a is described below
commit 951d091ae662e04d542981dd16b11d835726180f
Author: Andrés Beck-Ruiz <[email protected]>
AuthorDate: Mon Jul 13 23:17:12 2026 -0400
CASSSIDECAR-374: Implement durable operational job tracker (#359)
Patch by Andrés Beck-Ruiz; reviewed by Paulo Motta, Shailaja Koppu, Arjun
Ashok, Francisco Guerrero for CASSSIDECAR-374
---
CHANGES.txt | 1 +
.../sidecar/db/ClusterOpsDatabaseAccessor.java | 15 +-
.../sidecar/job/DurableOperationalJobTracker.java | 270 +++++++++++
.../cassandra/sidecar/job/NodeDecommissionJob.java | 10 +
.../apache/cassandra/sidecar/job/NodeDrainJob.java | 10 +
.../apache/cassandra/sidecar/job/NodeMoveJob.java | 10 +
.../cassandra/sidecar/job/OperationalJob.java | 7 +
.../cassandra/sidecar/job/OperationalJobInfo.java | 6 +
.../sidecar/job/OperationalJobManager.java | 9 +-
.../apache/cassandra/sidecar/job/RepairJob.java | 10 +
.../sidecar/job/storage/OperationalJobRecord.java | 252 ++++++++--
.../sidecar/utils/InvocationTrackingFunction.java | 50 ++
.../ClusterOpsDatabaseAccessorIntegrationTest.java | 33 +-
...urableOperationalJobTrackerIntegrationTest.java | 108 +++++
.../handlers/ListOperationalJobsHandlerTest.java | 7 +
.../job/DurableOperationalJobTrackerTest.java | 527 +++++++++++++++++++++
.../sidecar/job/OperationalJobInfoTest.java | 19 +
.../sidecar/job/OperationalJobManagerTest.java | 41 ++
.../cassandra/sidecar/job/OperationalJobTest.java | 25 +
19 files changed, 1354 insertions(+), 56 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 07239ad9..265a4a99 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Implement durable operational job tracker (CASSSIDECAR-374)
* Remove filesystem path from Http response (CASSSIDECAR-477)
* Add basic configuration retrieval logic to ConfigurationManager
(CASSSIDECAR-427)
* RPM package is created with wrong version (CASSSIDECAR-472)
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/db/ClusterOpsDatabaseAccessor.java
b/server/src/main/java/org/apache/cassandra/sidecar/db/ClusterOpsDatabaseAccessor.java
index eea6f769..9fb2ec4f 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/db/ClusterOpsDatabaseAccessor.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/db/ClusterOpsDatabaseAccessor.java
@@ -166,12 +166,19 @@ public class ClusterOpsDatabaseAccessor extends
DatabaseAccessor<ClusterOpsSchem
nodeExecutionOrder = null;
}
Map<String, String> operationMetadata =
row.getMap("operation_metadata", String.class, String.class);
- if (operationMetadata.isEmpty())
+ if (operationMetadata != null && operationMetadata.isEmpty())
{
operationMetadata = null;
}
- return new OperationalJobRecord(operationId, operationType, status,
- startTime, lastUpdate, failureReason,
- nodeExecutionOrder, operationMetadata);
+ return OperationalJobRecord.builder()
+ .jobId(operationId)
+ .operationType(operationType)
+ .status(status)
+ .startTime(startTime)
+ .lastUpdate(lastUpdate)
+ .failureReason(failureReason)
+ .nodeExecutionOrder(nodeExecutionOrder)
+ .operationMetadata(operationMetadata)
+ .build();
}
}
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
new file mode 100644
index 00000000..d5aa8622
--- /dev/null
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java
@@ -0,0 +1,270 @@
+/*
+ * 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.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+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.utils.InvocationTrackingFunction;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A durable implementation of {@link OperationalJobTracker} that persists job
state
+ * via a {@link StorageProvider}. A local {@link ConcurrentHashMap} caches live
+ * {@link OperationalJob} references for the current process, since executing
jobs
+ * with Vert.x promises cannot be reconstituted from storage. Once a job
completes,
+ * it is removed from the local map, and subsequent lookups are served from
storage.
+ *
+ * <p>Storage currently receives only two writes per job: the initial
+ * {@link OperationalJobStatus#CREATED CREATED} record on submission and the
terminal status on
+ * completion. As a result, if the sidecar restarts mid-operation, or if the
+ * terminal-status update exhausts its retries (see {@link
#updateTerminalStatus}), the persisted
+ * record can remain stuck at {@code CREATED} even though the operation has
since progressed or
+ * finished. There is currently no marker distinguishing a record that is
genuinely still
+ * {@code CREATED} from one whose true state was simply never recorded; adding
such a marker together
+ * with a reconciliation sweep (leveraging {@link
StorageProvider#findAllJobs(int)}) is tracked as
+ * follow-up work in
+ * <a
href="https://issues.apache.org/jira/browse/CASSSIDECAR-482";>CASSSIDECAR-482</a>.
+ */
+@Singleton
+public class DurableOperationalJobTracker implements OperationalJobTracker
+{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(DurableOperationalJobTracker.class);
+ private static final int MAX_STORAGE_WRITE_ATTEMPTS = 3;
+ private static final long RETRY_DELAY_MS = 100;
+ private static final long RETRY_JITTER_MS = 100;
+
+ private final ConcurrentHashMap<UUID, OperationalJob> liveJobs;
+ private final StorageProvider storageProvider;
+ private final TaskExecutorPool executor;
+
+ @Inject
+ public DurableOperationalJobTracker(ServiceConfiguration
serviceConfiguration,
+ StorageProvider storageProvider,
+ TaskExecutorPool executor)
+ {
+ this.liveJobs = new
ConcurrentHashMap<>(serviceConfiguration.operationalJobTrackerSize());
+ this.storageProvider = storageProvider;
+ this.executor = executor;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @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);
+ }
+
+ return job;
+ }
+
+ private void persistInitialRecord(OperationalJob job, int attempt)
+ {
+ executor.runBlocking(() ->
storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job)))
+ .onSuccess(v -> job.asyncResult().onComplete(ar -> {
+ updateTerminalStatus(job);
+ liveJobs.remove(job.jobId());
+ }))
+ .onFailure(e -> {
+ LOGGER.warn("Failed to persist job {} to storage (attempt
{}/{}). error={}",
+ job.jobId(), attempt,
MAX_STORAGE_WRITE_ATTEMPTS, e.getMessage());
+ if (attempt < MAX_STORAGE_WRITE_ATTEMPTS)
+ {
+ // 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));
+ }
+ else
+ {
+ // Persist exhausted its retries, but the job is
already executing on a separate executor.
+ // Keep it in liveJobs so in-process status queries
and conflict detection still see it.
+ LOGGER.error("Failed to persist job {} to storage
after {} attempts. "
+ + "Job will be tracked in-memory only.",
job.jobId(), MAX_STORAGE_WRITE_ATTEMPTS, e);
+ job.asyncResult().onComplete(ar ->
liveJobs.remove(job.jobId()));
+ }
+ });
+ }
+
+ @Nullable
+ @Override
+ public OperationalJobInfo get(UUID jobId)
+ {
+ OperationalJob liveJob = liveJobs.get(jobId);
+ if (liveJob != null)
+ {
+ return liveJob;
+ }
+
+ OperationalJobRecord record = storageProvider.findJob(jobId);
+ if (record == null)
+ {
+ return null;
+ }
+ return enrichWithNodeStatuses(record);
+ }
+
+ @NotNull
+ @Override
+ public Map<UUID, OperationalJob> jobsView()
+ {
+ return Collections.unmodifiableMap(liveJobs);
+ }
+
+ @NotNull
+ @Override
+ public List<OperationalJob> inflightJobsByOperation(String operation)
+ {
+ return liveJobs.values()
+ .stream()
+ .filter(j -> j.name().equals(operation) &&
+ (j.status() ==
OperationalJobStatus.RUNNING ||
+ j.status() ==
OperationalJobStatus.CREATED))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * 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
+ * that joins the data in a single query), the record is returned as is.
+ */
+ private OperationalJobRecord enrichWithNodeStatuses(OperationalJobRecord
record)
+ {
+ if (!record.nodesPending().isEmpty()
+ || !record.nodesExecuting().isEmpty()
+ || !record.nodesSucceeded().isEmpty()
+ || !record.nodesFailed().isEmpty())
+ {
+ return record;
+ }
+
+ Map<UUID, OperationalJobStatus> nodeStatuses =
+ storageProvider.getNodeStatusesForOperation(record.jobId());
+ if (nodeStatuses.isEmpty())
+ {
+ return record;
+ }
+
+ List<UUID> pending = new ArrayList<>();
+ List<UUID> executing = new ArrayList<>();
+ List<UUID> succeeded = new ArrayList<>();
+ List<UUID> failed = new ArrayList<>();
+
+ for (Map.Entry<UUID, OperationalJobStatus> entry :
nodeStatuses.entrySet())
+ {
+ switch (entry.getValue())
+ {
+ case CREATED:
+ pending.add(entry.getKey());
+ break;
+ case RUNNING:
+ executing.add(entry.getKey());
+ break;
+ case SUCCEEDED:
+ succeeded.add(entry.getKey());
+ break;
+ case FAILED:
+ failed.add(entry.getKey());
+ break;
+ default:
+ throw new IllegalStateException("Invalid state = " +
entry.getValue());
+ }
+ }
+
+ return OperationalJobRecord.builder()
+ .jobId(record.jobId())
+ .operationType(record.operationType())
+ .status(record.status())
+ .startTime(record.startTime())
+ .lastUpdate(record.lastUpdate())
+ .failureReason(record.failureReason())
+
.nodeExecutionOrder(record.nodeExecutionOrder())
+
.operationMetadata(record.operationMetadata())
+
.nodesPending(Collections.unmodifiableList(pending))
+
.nodesExecuting(Collections.unmodifiableList(executing))
+
.nodesSucceeded(Collections.unmodifiableList(succeeded))
+
.nodesFailed(Collections.unmodifiableList(failed))
+ .build();
+ }
+
+ /**
+ * Attempts to update the terminal status in storage with retry.
+ * If all attempts fail, logs a warning and continues.
+ */
+ private void updateTerminalStatus(OperationalJob job)
+ {
+ updateTerminalStatus(job, 1);
+ }
+
+ private void updateTerminalStatus(OperationalJob job, int attempt)
+ {
+ try
+ {
+ storageProvider.updateJobStatus(job.jobId(), job.operationType(),
job.status(), job.failureReason());
+ }
+ catch (RuntimeException e)
+ {
+ LOGGER.warn("Failed to update terminal status for job {} (attempt
{}/{}). error={}",
+ job.jobId(), attempt, MAX_STORAGE_WRITE_ATTEMPTS,
e.getMessage());
+ if (attempt < MAX_STORAGE_WRITE_ATTEMPTS)
+ {
+ // 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 -> updateTerminalStatus(job,
attempt + 1));
+ }
+ else
+ {
+ LOGGER.error("Exhausted retries when updating terminal status
for job {}. " +
+ "Manual intervention may be required to correct
job metadata.", job.jobId(), e);
+ }
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDecommissionJob.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDecommissionJob.java
index e042200f..c67026ac 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDecommissionJob.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDecommissionJob.java
@@ -25,6 +25,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.vertx.core.Future;
+import org.apache.cassandra.sidecar.common.data.OperationType;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
import org.jetbrains.annotations.NotNull;
@@ -103,6 +104,15 @@ public class NodeDecommissionJob extends OperationalJob
return Future.succeededFuture();
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DECOMMISSION;
+ }
+
/**
* {@inheritDoc}
*/
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDrainJob.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDrainJob.java
index b5b1db92..52d44bfa 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDrainJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDrainJob.java
@@ -27,6 +27,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.vertx.core.Future;
+import org.apache.cassandra.sidecar.common.data.OperationType;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
import
org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException;
@@ -136,6 +137,15 @@ public class NodeDrainJob extends OperationalJob
return Future.succeededFuture();
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
/**
* {@inheritDoc}
*/
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeMoveJob.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeMoveJob.java
index c35f13ef..5fa04f30 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeMoveJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeMoveJob.java
@@ -26,6 +26,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.vertx.core.Future;
+import org.apache.cassandra.sidecar.common.data.OperationType;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
import
org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException;
import org.jetbrains.annotations.NotNull;
@@ -83,6 +84,15 @@ public class NodeMoveJob extends OperationalJob
return Future.succeededFuture();
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.MOVE;
+ }
+
/**
* {@inheritDoc}
*/
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 58320e4e..f3c15c6a 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
@@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory;
import com.datastax.driver.core.utils.UUIDs;
import io.vertx.core.Future;
import io.vertx.core.Promise;
+import org.apache.cassandra.sidecar.common.data.OperationType;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
import
org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
@@ -239,6 +240,12 @@ public abstract class OperationalJob implements
Task<Void>, OperationalJobInfo
return simpleName.isEmpty() ? this.getClass().getName() : simpleName;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public abstract OperationType operationType();
+
/**
* Determines the status of the job. OperationalJob subclasses could
choose to override the method.
* <p>
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobInfo.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobInfo.java
index 1748e3db..a6fc8a95 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobInfo.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobInfo.java
@@ -22,6 +22,7 @@ import java.time.Instant;
import java.util.List;
import java.util.UUID;
+import org.apache.cassandra.sidecar.common.data.OperationType;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -48,6 +49,11 @@ public interface OperationalJobInfo
*/
String name();
+ /**
+ * @return the {@link OperationType} of this job
+ */
+ OperationType operationType();
+
/**
* @return the current status of the job
*/
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 ccaed374..a1e4d406 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
@@ -100,11 +100,12 @@ public class OperationalJobManager
{
checkConflict(job);
- // New job is submitted for all cases when we do not have a
corresponding downstream job
- jobTracker.computeIfAbsent(job.jobId(), jobId -> {
+ // Track the job first, then start execution separately
+ OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(),
jobId -> job);
+ if (tracked == job)
+ {
internalExecutorPool.executeBlocking(job::execute);
- return job;
- });
+ }
}
catch (OperationalJobConflictException oje)
{
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/job/RepairJob.java
b/server/src/main/java/org/apache/cassandra/sidecar/job/RepairJob.java
index a9fe4ccb..156fa781 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/RepairJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/RepairJob.java
@@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import org.apache.cassandra.sidecar.adapters.base.RepairOptions;
+import org.apache.cassandra.sidecar.common.data.OperationType;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
@@ -184,6 +185,15 @@ public class RepairJob extends OperationalJob
return currentStatus != null ? currentStatus : super.status();
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.REPAIR;
+ }
+
/**
* {@inheritDoc}
*/
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 7e3d508a..42761634 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
@@ -19,21 +19,29 @@
package org.apache.cassandra.sidecar.job.storage;
import java.time.Instant;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.UUID;
import com.datastax.driver.core.utils.UUIDs;
+import org.apache.cassandra.sidecar.common.DataObjectBuilder;
import org.apache.cassandra.sidecar.common.data.OperationType;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
import org.apache.cassandra.sidecar.common.utils.Preconditions;
+import org.apache.cassandra.sidecar.job.OperationalJob;
+import org.apache.cassandra.sidecar.job.OperationalJobInfo;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
* A data transfer object representing the persisted state of an operational
job.
+ * Implements {@link OperationalJobInfo} so that completed jobs stored in a
persistent storage
+ * can be returned directly through the tracker without requiring a live
{@link OperationalJob}.
*/
-public class OperationalJobRecord
+public class OperationalJobRecord implements OperationalJobInfo
{
private final UUID jobId;
private final OperationType operationType;
@@ -49,51 +57,39 @@ public class OperationalJobRecord
private final List<List<UUID>> nodeExecutionOrder;
@Nullable
private final Map<String, String> operationMetadata;
+ @NotNull
+ private final List<UUID> nodesPending;
+ @NotNull
+ private final List<UUID> nodesExecuting;
+ @NotNull
+ private final List<UUID> nodesSucceeded;
+ @NotNull
+ private final List<UUID> nodesFailed;
/**
- * Constructs an OperationalJobRecord with the given fields.
+ * Constructs an OperationalJobRecord from its {@link Builder}.
*
- * @param jobId time-based v1 UUID identifying the job
- * @param operationType the operation type
- * @param status the current status of the job
+ * @param builder the builder holding the field values
*/
- public OperationalJobRecord(UUID jobId, OperationType operationType,
OperationalJobStatus status)
+ private OperationalJobRecord(Builder builder)
{
- this(jobId, operationType, status, null, Instant.now(), null, null,
null);
- }
-
- /**
- * Constructs an OperationalJobRecord with all fields.
- *
- * @param jobId time-based v1 UUID identifying the job
- * @param operationType the operation type
- * @param status the current status of the job
- * @param startTime the timestamp when execution started, or null
if not yet started
- * @param lastUpdate the timestamp of the last status update, or
null for pre-existing rows
- * @param failureReason the failure reason if the job failed, or null
- * @param nodeExecutionOrder the ordered list of parallel node
groups for execution, or null
- * @param operationMetadata the operation parameters, or null
- */
- public OperationalJobRecord(UUID jobId, OperationType operationType,
OperationalJobStatus status,
- @Nullable Instant startTime,
- @Nullable Instant lastUpdate,
- @Nullable String failureReason,
- @Nullable List<List<UUID>> nodeExecutionOrder,
- @Nullable Map<String, String>
operationMetadata)
- {
- Preconditions.checkArgument(jobId != null, "jobId must not be null");
- Preconditions.checkArgument(jobId.version() == 1, "jobId must be a
time-based (v1) UUID");
- Preconditions.checkArgument(operationType != null, "operationType must
not be null");
- Preconditions.checkArgument(status != null, "status must not be null");
- this.jobId = jobId;
- this.operationType = operationType;
- this.status = status;
- this.creationTimeMillis = UUIDs.unixTimestamp(jobId);
- this.startTime = startTime;
- this.lastUpdate = lastUpdate;
- this.failureReason = failureReason;
- this.nodeExecutionOrder = nodeExecutionOrder;
- this.operationMetadata = operationMetadata;
+ Preconditions.checkArgument(builder.jobId != null, "jobId must not be
null");
+ Preconditions.checkArgument(builder.jobId.version() == 1, "jobId must
be a time-based (v1) UUID");
+ Preconditions.checkArgument(builder.operationType != null,
"operationType must not be null");
+ Preconditions.checkArgument(builder.status != null, "status must not
be null");
+ jobId = builder.jobId;
+ operationType = builder.operationType;
+ status = builder.status;
+ creationTimeMillis = UUIDs.unixTimestamp(builder.jobId);
+ startTime = builder.startTime;
+ lastUpdate = builder.lastUpdate;
+ failureReason = builder.failureReason;
+ nodeExecutionOrder = builder.nodeExecutionOrder;
+ operationMetadata = builder.operationMetadata;
+ nodesPending = builder.nodesPending;
+ nodesExecuting = builder.nodesExecuting;
+ nodesSucceeded = builder.nodesSucceeded;
+ nodesFailed = builder.nodesFailed;
}
/**
@@ -173,6 +169,74 @@ public class OperationalJobRecord
return operationMetadata;
}
+ @Override
+ @Nullable
+ public UUID nodeId()
+ {
+ return null;
+ }
+
+ @Override
+ public String name()
+ {
+ return operationType.name().toLowerCase();
+ }
+
+ @Override
+ public long creationTime()
+ {
+ return creationTimeMillis;
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesPending()
+ {
+ return nodesPending;
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesExecuting()
+ {
+ return nodesExecuting;
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesSucceeded()
+ {
+ return nodesSucceeded;
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesFailed()
+ {
+ return nodesFailed;
+ }
+
+ @Override
+ public boolean isExecuting()
+ {
+ return status == OperationalJobStatus.RUNNING;
+ }
+
+ /**
+ * Creates an {@link OperationalJobRecord} from a live {@link
OperationalJob}.
+ *
+ * @param job the operational job to convert
+ * @return a new record capturing the job's current state
+ */
+ public static OperationalJobRecord fromOperationalJob(OperationalJob job)
+ {
+ return builder().jobId(job.jobId())
+ .operationType(job.operationType())
+ .status(job.status())
+ .lastUpdate(Instant.now())
+ .build();
+ }
+
@Override
public String toString()
{
@@ -186,6 +250,112 @@ public class OperationalJobRecord
", failureReason=" + failureReason +
", nodeExecutionOrder=" + nodeExecutionOrder +
", operationMetadata=" + operationMetadata +
+ ", nodesPending=" + nodesPending +
+ ", nodesExecuting=" + nodesExecuting +
+ ", nodesSucceeded=" + nodesSucceeded +
+ ", nodesFailed=" + nodesFailed +
'}';
}
+
+ /**
+ * @return a new {@link Builder} for constructing an {@link
OperationalJobRecord}
+ */
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ /**
+ * {@link DataObjectBuilder} for {@link OperationalJobRecord}.
+ */
+ public static final class Builder implements DataObjectBuilder<Builder,
OperationalJobRecord>
+ {
+ private UUID jobId;
+ private OperationType operationType;
+ private OperationalJobStatus status;
+ private @Nullable Instant startTime;
+ private @Nullable Instant lastUpdate;
+ private @Nullable String failureReason;
+ private @Nullable List<List<UUID>> nodeExecutionOrder;
+ private @Nullable Map<String, String> operationMetadata;
+ private @NotNull List<UUID> nodesPending = Collections.emptyList();
+ private @NotNull List<UUID> nodesExecuting = Collections.emptyList();
+ private @NotNull List<UUID> nodesSucceeded = Collections.emptyList();
+ private @NotNull List<UUID> nodesFailed = Collections.emptyList();
+
+ private Builder()
+ {
+ }
+
+ @Override
+ public Builder self()
+ {
+ return this;
+ }
+
+ public Builder jobId(UUID jobId)
+ {
+ return update(b -> b.jobId = jobId);
+ }
+
+ public Builder operationType(OperationType operationType)
+ {
+ return update(b -> b.operationType = operationType);
+ }
+
+ public Builder status(OperationalJobStatus status)
+ {
+ return update(b -> b.status = status);
+ }
+
+ public Builder startTime(@Nullable Instant startTime)
+ {
+ return update(b -> b.startTime = startTime);
+ }
+
+ public Builder lastUpdate(@Nullable Instant lastUpdate)
+ {
+ return update(b -> b.lastUpdate = lastUpdate);
+ }
+
+ public Builder failureReason(@Nullable String failureReason)
+ {
+ return update(b -> b.failureReason = failureReason);
+ }
+
+ public Builder nodeExecutionOrder(@Nullable List<List<UUID>>
nodeExecutionOrder)
+ {
+ return update(b -> b.nodeExecutionOrder = nodeExecutionOrder);
+ }
+
+ public Builder operationMetadata(@Nullable Map<String, String>
operationMetadata)
+ {
+ return update(b -> b.operationMetadata = operationMetadata);
+ }
+
+ public Builder nodesPending(@NotNull List<UUID> nodesPending)
+ {
+ return update(b -> b.nodesPending =
Objects.requireNonNull(nodesPending, "nodesPending cannot be null"));
+ }
+
+ public Builder nodesExecuting(@NotNull List<UUID> nodesExecuting)
+ {
+ return update(b -> b.nodesExecuting =
Objects.requireNonNull(nodesExecuting, "nodesExecuting cannot be null"));
+ }
+
+ public Builder nodesSucceeded(@NotNull List<UUID> nodesSucceeded)
+ {
+ return update(b -> b.nodesSucceeded =
Objects.requireNonNull(nodesSucceeded, "nodesSucceeded cannot be null"));
+ }
+
+ public Builder nodesFailed(@NotNull List<UUID> nodesFailed)
+ {
+ return update(b -> b.nodesFailed =
Objects.requireNonNull(nodesFailed, "nodesFailed cannot be null"));
+ }
+
+ public OperationalJobRecord build()
+ {
+ return new OperationalJobRecord(this);
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/utils/InvocationTrackingFunction.java
b/server/src/main/java/org/apache/cassandra/sidecar/utils/InvocationTrackingFunction.java
new file mode 100644
index 00000000..5d2feef9
--- /dev/null
+++
b/server/src/main/java/org/apache/cassandra/sidecar/utils/InvocationTrackingFunction.java
@@ -0,0 +1,50 @@
+/*
+ * 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.utils;
+
+import java.util.function.Function;
+
+/**
+ * A {@link Function} decorator that records whether the wrapped function was
ever invoked.
+ *
+ * @param <T> the function input type
+ * @param <R> the function result type
+ */
+public class InvocationTrackingFunction<T, R> implements Function<T, R>
+{
+ private final Function<T, R> delegate;
+ private boolean invoked;
+
+ public InvocationTrackingFunction(Function<T, R> delegate)
+ {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public R apply(T input)
+ {
+ invoked = true;
+ return delegate.apply(input);
+ }
+
+ public boolean wasInvoked()
+ {
+ return invoked;
+ }
+}
diff --git
a/server/src/test/integration/org/apache/cassandra/sidecar/db/ClusterOpsDatabaseAccessorIntegrationTest.java
b/server/src/test/integration/org/apache/cassandra/sidecar/db/ClusterOpsDatabaseAccessorIntegrationTest.java
index 84720cc0..0942cd92 100644
---
a/server/src/test/integration/org/apache/cassandra/sidecar/db/ClusterOpsDatabaseAccessorIntegrationTest.java
+++
b/server/src/test/integration/org/apache/cassandra/sidecar/db/ClusterOpsDatabaseAccessorIntegrationTest.java
@@ -61,8 +61,14 @@ class ClusterOpsDatabaseAccessorIntegrationTest extends
IntegrationTestBase
Arrays.asList(UUID.randomUUID())
);
Map<String, String> metadata = Map.of("key1", "value1", "key2",
"value2");
- OperationalJobRecord job1 = new OperationalJobRecord(jobId1,
OperationType.REPAIR, OperationalJobStatus.CREATED,
- null,
Instant.now(), null, nodeOrder, metadata);
+ OperationalJobRecord job1 = OperationalJobRecord.builder()
+ .jobId(jobId1)
+
.operationType(OperationType.REPAIR)
+
.status(OperationalJobStatus.CREATED)
+
.lastUpdate(Instant.now())
+
.nodeExecutionOrder(nodeOrder)
+
.operationMetadata(metadata)
+ .build();
accessor.persistJob(clusterName, job1);
OperationalJobRecord found = accessor.findJob(clusterName, jobId1);
@@ -81,7 +87,11 @@ class ClusterOpsDatabaseAccessorIntegrationTest extends
IntegrationTestBase
});
UUID jobId2 = UUIDs.timeBased();
- OperationalJobRecord job2 = new OperationalJobRecord(jobId2,
OperationType.DECOMMISSION, OperationalJobStatus.CREATED);
+ OperationalJobRecord job2 = OperationalJobRecord.builder()
+ .jobId(jobId2)
+
.operationType(OperationType.DECOMMISSION)
+
.status(OperationalJobStatus.CREATED)
+ .build();
accessor.persistJob(clusterName, job2);
OperationalJobRecord found2 = accessor.findJob(clusterName, jobId2);
@@ -133,7 +143,11 @@ class ClusterOpsDatabaseAccessorIntegrationTest extends
IntegrationTestBase
});
UUID jobId3 = UUIDs.timeBased();
- accessor.persistJob(clusterName, new OperationalJobRecord(jobId3,
OperationType.REPAIR, OperationalJobStatus.CREATED));
+ accessor.persistJob(clusterName, OperationalJobRecord.builder()
+ .jobId(jobId3)
+
.operationType(OperationType.REPAIR)
+
.status(OperationalJobStatus.CREATED)
+ .build());
List<OperationalJobRecord> allJobs = accessor.findAllJobs(clusterName,
10);
assertThat(allJobs)
.withFailMessage("findAllJobs should return every persisted
job when the limit exceeds total count")
@@ -148,9 +162,14 @@ class ClusterOpsDatabaseAccessorIntegrationTest extends
IntegrationTestBase
.withFailMessage("findAllJobs should return the most recently
created job first")
.isEqualTo(jobId3);
- OperationalJobRecord job1Updated = new OperationalJobRecord(jobId1,
OperationType.REPAIR,
-
OperationalJobStatus.SUCCEEDED,
- null,
Instant.now(), null, nodeOrder, metadata);
+ OperationalJobRecord job1Updated = OperationalJobRecord.builder()
+ .jobId(jobId1)
+
.operationType(OperationType.REPAIR)
+
.status(OperationalJobStatus.SUCCEEDED)
+
.lastUpdate(Instant.now())
+
.nodeExecutionOrder(nodeOrder)
+
.operationMetadata(metadata)
+ .build();
accessor.persistJob(clusterName, job1Updated);
OperationalJobRecord afterUpsert = accessor.findJob(clusterName,
jobId1);
assertThat(afterUpsert)
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
new file mode 100644
index 00000000..155161d9
--- /dev/null
+++
b/server/src/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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 java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import com.datastax.driver.core.utils.UUIDs;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import
org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+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.testing.IntegrationTestBase;
+import org.apache.cassandra.testing.CassandraIntegrationTest;
+
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for {@link DurableOperationalJobTracker} with a
Cassandra-backed {@link StorageProvider}.
+ */
+class DurableOperationalJobTrackerIntegrationTest extends IntegrationTestBase
+{
+ @CassandraIntegrationTest
+ void testDurableOperationalJobTrackerOperations() throws
InterruptedException
+ {
+ waitForSchemaReady(10, TimeUnit.SECONDS);
+
+ StorageProvider storageProvider =
injector.getInstance(StorageProvider.class);
+ storageProvider.initialize();
+
+ ExecutorPools executorPools =
injector.getInstance(ExecutorPools.class);
+ TaskExecutorPool executorPool = executorPools.internal();
+ DurableOperationalJobTracker tracker = new
DurableOperationalJobTracker(new ServiceConfigurationImpl(),
+
storageProvider,
+
executorPool);
+
+ UUID jobId1 = UUIDs.timeBased();
+ OperationalJob job1 = OperationalJobTest.createOperationalJob(jobId1,
OperationalJobStatus.CREATED);
+
+ tracker.computeIfAbsent(jobId1, id -> job1);
+
+ loopAssert(2, () -> {
+ OperationalJobRecord record1 = storageProvider.findJob(jobId1);
+ assertThat(record1)
+ .withFailMessage("Job should be persisted to Cassandra on
creation")
+ .isNotNull()
+ .satisfies(r -> {
+ assertThat(r.jobId()).isEqualTo(jobId1);
+
assertThat(r.operationType()).isEqualTo(job1.operationType());
+
assertThat(r.status()).isEqualTo(OperationalJobStatus.CREATED);
+ });
+ });
+
+ UUID jobId2 = UUIDs.timeBased();
+ OperationalJob job2 = OperationalJobTest.createOperationalJob(jobId2,
MillisecondBoundConfiguration.parse("50ms"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId2, id -> job2);
+ executorPool.executeBlocking(job2::execute);
+
+ job2.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ OperationalJobRecord record2 = storageProvider.findJob(jobId2);
+ assertThat(record2)
+ .withFailMessage("Job status should be updated to SUCCEEDED in
Cassandra")
+ .isNotNull()
+ .satisfies(r ->
assertThat(r.status()).isEqualTo(OperationalJobStatus.SUCCEEDED));
+ });
+
+ loopAssert(2, () -> {
+ assertThat(tracker.jobsView())
+ .withFailMessage("Job should be removed from local map after
completion")
+ .doesNotContainKey(jobId2);
+ });
+
+ OperationalJobInfo retrieved = tracker.get(jobId2);
+ assertThat(retrieved)
+ .withFailMessage("get() should return an OperationalJobRecord from
storage after completion")
+ .isNotNull()
+ .isInstanceOf(OperationalJobRecord.class);
+ assertThat(retrieved.jobId()).isEqualTo(jobId2);
+
assertThat(retrieved.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ assertThat(retrieved.operationType()).isEqualTo(job2.operationType());
+ }
+}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandlerTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandlerTest.java
index 91159bd0..8edd2d82 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandlerTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandlerTest.java
@@ -47,6 +47,7 @@ import io.vertx.ext.web.client.predicate.ResponsePredicate;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.common.data.OperationType;
import
org.apache.cassandra.sidecar.common.response.ListOperationalJobsResponse;
import
org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException;
import org.apache.cassandra.sidecar.job.OperationalJob;
@@ -157,6 +158,12 @@ class ListOperationalJobsHandlerTest
return false;
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
protected Future<Void> executeInternal() throws OperationalJobException
{
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
new file mode 100644
index 00000000..58158a78
--- /dev/null
+++
b/server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java
@@ -0,0 +1,527 @@
+/*
+ * 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.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.datastax.driver.core.utils.UUIDs;
+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.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+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 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.createOperationalJob;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+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.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link DurableOperationalJobTracker}
+ */
+class DurableOperationalJobTrackerTest
+{
+ private StorageProvider storageProvider;
+ private DurableOperationalJobTracker tracker;
+ private Vertx vertx;
+ private ExecutorPools executorPools;
+ private TaskExecutorPool executorPool;
+
+ @BeforeEach
+ void setUp()
+ {
+ storageProvider = mock(StorageProvider.class);
+ when(storageProvider.isAvailable()).thenReturn(true);
+ vertx = Vertx.vertx();
+ executorPools = new ExecutorPools(vertx, new
ServiceConfigurationImpl());
+ executorPool = executorPools.internal();
+ tracker = new DurableOperationalJobTracker(new
ServiceConfigurationImpl(), storageProvider, executorPool);
+ }
+
+ @AfterEach
+ void cleanup()
+ {
+ TestResourceReaper.create().with(vertx).with(executorPools).close();
+ }
+
+ @Test
+ void testComputeIfAbsentPersistsNewJob()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+
+ OperationalJob result = tracker.computeIfAbsent(job.jobId(), id ->
job);
+
+ assertThat(result).isSameAs(job);
+ loopAssert(2, () -> {
+ verify(storageProvider).persistJob(argThat(record ->
+ record.jobId().equals(job.jobId()) &&
+ record.operationType() == job.operationType() &&
+ record.status() == CREATED
+ ));
+ });
+ }
+
+ @Test
+ void testComputeIfAbsentDoesNotPersistExistingJob()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+
+ tracker.computeIfAbsent(job.jobId(), id -> job);
+ OperationalJob secondCall = tracker.computeIfAbsent(job.jobId(), id ->
createOperationalJob(CREATED));
+
+ assertThat(secondCall).isSameAs(job);
+ loopAssert(2, () -> {
+ verify(storageProvider, times(1)).persistJob(any());
+ });
+ }
+
+ @Test
+ void testComputeIfAbsentUpdatesStatusOnCompletion() 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 testTerminalStatusNotUpdatedWhenPersistFails() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ doThrow(new StorageProviderException("persist
failed")).when(storageProvider).persistJob(any());
+
+ CountDownLatch completed = new CountDownLatch(1);
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+ job.asyncResult().onComplete(ar -> completed.countDown());
+
+ assertThat(completed.await(5, TimeUnit.SECONDS)).isTrue();
+ // The tracker only registers its completion handler after a
successful persist. Since persist
+ // failed here, the handler is never registered, so updateJobStatus
should not run.
+ verify(storageProvider, never()).updateJobStatus(any(), any(), any(),
any());
+ }
+
+ @Test
+ void testComputeIfAbsentUpdatesStatusOnFailure() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
+ MillisecondBoundConfiguration.parse("50ms"),
+ new
org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException("test
failure"));
+ 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(FAILED), eq("test failure"));
+ });
+ }
+
+ @Test
+ void testComputeIfAbsentRemovesJobFromLocalMapOnCompletion() 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);
+
+ assertThat(tracker.jobsView()).containsKey(jobId);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ assertThat(tracker.jobsView()).doesNotContainKey(jobId);
+ });
+ }
+
+ @Test
+ void testGetReturnsLiveJobFromLocalMap()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(job.jobId(), id -> job);
+
+ OperationalJobInfo result = tracker.get(job.jobId());
+
+ assertThat(result).isSameAs(job);
+ verify(storageProvider, never()).findJob(any());
+ }
+
+ @Test
+ void testGetFallsBackToStorage()
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJobRecord record = OperationalJobRecord.builder()
+ .jobId(jobId)
+
.operationType(OperationType.DECOMMISSION)
+ .status(SUCCEEDED)
+ .build();
+ when(storageProvider.findJob(jobId)).thenReturn(record);
+
when(storageProvider.getNodeStatusesForOperation(jobId)).thenReturn(Collections.emptyMap());
+
+ OperationalJobInfo result = tracker.get(jobId);
+
+ assertThat(result).isInstanceOf(OperationalJobRecord.class);
+ assertThat(result.jobId()).isEqualTo(jobId);
+ assertThat(result.status()).isEqualTo(SUCCEEDED);
+
assertThat(result.name()).isEqualTo(OperationType.DECOMMISSION.name().toLowerCase());
+
assertThat(result.operationType()).isEqualTo(OperationType.DECOMMISSION);
+ verify(storageProvider).findJob(jobId);
+ }
+
+ @Test
+ void testGetReturnsNullWhenNotFoundAnywhere()
+ {
+ UUID jobId = UUIDs.timeBased();
+ when(storageProvider.findJob(jobId)).thenReturn(null);
+
+ OperationalJobInfo result = tracker.get(jobId);
+
+ assertThat(result).isNull();
+ verify(storageProvider).findJob(jobId);
+ }
+
+ @Test
+ void testJobsViewReturnsOnlyLocalMapContents()
+ {
+ OperationalJob job1 = createOperationalJob(CREATED);
+ OperationalJob job2 = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(job1.jobId(), id -> job1);
+ tracker.computeIfAbsent(job2.jobId(), id -> job2);
+
+ Map<UUID, OperationalJob> view = tracker.jobsView();
+
+ assertThat(view).hasSize(2);
+ assertThat(view).containsEntry(job1.jobId(), job1);
+ assertThat(view).containsEntry(job2.jobId(), job2);
+ verify(storageProvider, never()).findJob(any());
+ verify(storageProvider, never()).findAllJobs(any(int.class));
+ }
+
+ @Test
+ void testJobsViewIsImmutable()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(job.jobId(), id -> job);
+
+ Map<UUID, OperationalJob> view = tracker.jobsView();
+
+ assertThatThrownBy(() -> view.put(UUIDs.timeBased(), job))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ @Test
+ void testInflightJobsByOperationFiltersLocalMap()
+ {
+ OperationalJob createdJob = createOperationalJob("decommission",
CREATED);
+ OperationalJob runningJob = createOperationalJob("decommission",
RUNNING);
+ OperationalJob succeededJob = createOperationalJob("decommission",
SUCCEEDED);
+ OperationalJob drainJob = createOperationalJob("drain", RUNNING);
+
+ tracker.computeIfAbsent(createdJob.jobId(), id -> createdJob);
+ tracker.computeIfAbsent(runningJob.jobId(), id -> runningJob);
+ tracker.computeIfAbsent(succeededJob.jobId(), id -> succeededJob);
+ tracker.computeIfAbsent(drainJob.jobId(), id -> drainJob);
+
+ List<OperationalJob> inflightDecommission =
tracker.inflightJobsByOperation("decommission");
+
+ assertThat(inflightDecommission)
+ .hasSize(2)
+ .containsExactlyInAnyOrder(createdJob, runningJob);
+ verify(storageProvider, never()).findJob(any());
+ }
+
+ @Test
+ void testComputeIfAbsentKeepsJobInMapOnPersistenceFailureUntilCompletion()
throws InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("200ms"));
+
+ doThrow(new StorageProviderException("Storage unavailable"))
+ .when(storageProvider).persistJob(any());
+
+ CountDownLatch completed = new CountDownLatch(1);
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+ job.asyncResult().onComplete(ar -> completed.countDown());
+
+ // The persist failed, but the job is already executing, so it must
remain in the live map
+ // (durability-degraded) rather than being dropped mid-flight.
+ loopAssert(2, () -> {
+ verify(storageProvider, atLeastOnce()).persistJob(any());
+ assertThat(tracker.jobsView()).containsKey(jobId);
+ });
+
+ // Once the job completes, the completion handler removes it from the
live map to avoid leaking.
+ assertThat(completed.await(5, TimeUnit.SECONDS)).isTrue();
+ loopAssert(2, () -> {
+ assertThat(tracker.jobsView()).doesNotContainKey(jobId);
+ });
+ }
+
+ @Test
+ void testComputeIfAbsentFailsWhenStorageUnavailable()
+ {
+ when(storageProvider.isAvailable()).thenReturn(false);
+ OperationalJob job = createOperationalJob(CREATED);
+
+ assertThatThrownBy(() -> tracker.computeIfAbsent(job.jobId(), id ->
job))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Storage provider is not available");
+ }
+
+ @Test
+ void testGetPropagatesStorageFallbackFailure()
+ {
+ UUID jobId = UUIDs.timeBased();
+ when(storageProvider.findJob(jobId)).thenThrow(new
StorageProviderException("Storage unavailable"));
+
+ assertThatThrownBy(() -> tracker.get(jobId))
+ .isInstanceOf(StorageProviderException.class)
+ .hasMessage("Storage unavailable");
+ }
+
+ @Test
+ void testGetEnrichesRecordWithNodeStatuses()
+ {
+ UUID jobId = UUIDs.timeBased();
+ UUID pendingNode = UUIDs.timeBased();
+ UUID runningNode = UUIDs.timeBased();
+ UUID succeededNode = UUIDs.timeBased();
+ UUID failedNode = UUIDs.timeBased();
+
+ OperationalJobRecord record = OperationalJobRecord.builder()
+ .jobId(jobId)
+
.operationType(OperationType.DECOMMISSION)
+ .status(RUNNING)
+ .build();
+ when(storageProvider.findJob(jobId)).thenReturn(record);
+
+ Map<UUID, OperationalJobStatus> nodeStatuses = new HashMap<>();
+ nodeStatuses.put(pendingNode, CREATED);
+ nodeStatuses.put(runningNode, RUNNING);
+ nodeStatuses.put(succeededNode, SUCCEEDED);
+ nodeStatuses.put(failedNode, FAILED);
+
when(storageProvider.getNodeStatusesForOperation(jobId)).thenReturn(nodeStatuses);
+
+ OperationalJobInfo result = tracker.get(jobId);
+
+ assertThat(result).isInstanceOf(OperationalJobRecord.class);
+ assertThat(result.nodesPending()).containsExactly(pendingNode);
+ assertThat(result.nodesExecuting()).containsExactly(runningNode);
+ assertThat(result.nodesSucceeded()).containsExactly(succeededNode);
+ assertThat(result.nodesFailed()).containsExactly(failedNode);
+ verify(storageProvider).getNodeStatusesForOperation(jobId);
+ }
+
+ @Test
+ void testGetSkipsEnrichmentWhenNodeListsAlreadyPopulated()
+ {
+ UUID jobId = UUIDs.timeBased();
+ UUID succeededNode = UUIDs.timeBased();
+
+ OperationalJobRecord record = OperationalJobRecord.builder()
+ .jobId(jobId)
+
.operationType(OperationType.DECOMMISSION)
+ .status(SUCCEEDED)
+
.nodesSucceeded(Collections.singletonList(succeededNode))
+ .build();
+ when(storageProvider.findJob(jobId)).thenReturn(record);
+
+ OperationalJobInfo result = tracker.get(jobId);
+
+ assertThat(result).isSameAs(record);
+ assertThat(result.nodesSucceeded()).containsExactly(succeededNode);
+ verify(storageProvider, never()).getNodeStatusesForOperation(any());
+ }
+
+ @Test
+ void testGetReturnsUnenrichedRecordWhenNoNodeStatuses()
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJobRecord record = OperationalJobRecord.builder()
+ .jobId(jobId)
+
.operationType(OperationType.DECOMMISSION)
+ .status(SUCCEEDED)
+ .build();
+ when(storageProvider.findJob(jobId)).thenReturn(record);
+
when(storageProvider.getNodeStatusesForOperation(jobId)).thenReturn(Collections.emptyMap());
+
+ OperationalJobInfo result = tracker.get(jobId);
+
+ assertThat(result).isSameAs(record);
+ assertThat(result.nodesPending()).isEmpty();
+ assertThat(result.nodesExecuting()).isEmpty();
+ assertThat(result.nodesSucceeded()).isEmpty();
+ assertThat(result.nodesFailed()).isEmpty();
+ verify(storageProvider).getNodeStatusesForOperation(jobId);
+ }
+
+ @Test
+ void testGetPropagatesNodeStatusQueryFailure()
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJobRecord record = OperationalJobRecord.builder()
+ .jobId(jobId)
+
.operationType(OperationType.DECOMMISSION)
+ .status(SUCCEEDED)
+ .build();
+ when(storageProvider.findJob(jobId)).thenReturn(record);
+ when(storageProvider.getNodeStatusesForOperation(jobId))
+ .thenThrow(new StorageProviderException("Node state unavailable"));
+
+ assertThatThrownBy(() -> tracker.get(jobId))
+ .isInstanceOf(StorageProviderException.class)
+ .hasMessage("Node state unavailable");
+ }
+
+ @Test
+ void testOnCompleteRetrySucceedsAfterTransientFailure() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ doThrow(new StorageProviderException("Transient failure"))
+ .doNothing()
+ .when(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DRAIN), eq(SUCCEEDED), isNull());
+
+ 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, times(2)).updateJobStatus(eq(jobId),
eq(OperationType.DRAIN), eq(SUCCEEDED), isNull());
+ });
+ }
+
+ @Test
+ void testOnCompleteRemovesJobFromLocalMapWhenAllRetriesFail() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ doThrow(new StorageProviderException("Persistent failure"))
+ .when(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DRAIN), eq(SUCCEEDED), isNull());
+
+ 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, times(3)).updateJobStatus(eq(jobId),
eq(OperationType.DRAIN), eq(SUCCEEDED), isNull());
+ assertThat(tracker.jobsView()).doesNotContainKey(jobId);
+ });
+ }
+
+ @Test
+ void testInitialPersistRetrySucceedsAfterTransientFailure() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ doThrow(new StorageProviderException("Transient failure"))
+ .doNothing()
+ .when(storageProvider).persistJob(any());
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ loopAssert(2, () -> {
+ verify(storageProvider, times(2)).persistJob(any());
+ verify(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DRAIN), eq(SUCCEEDED), isNull());
+ });
+ }
+
+ @Test
+ void testInitialPersistExhaustsRetriesAndTracksInMemoryOnly() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ doThrow(new StorageProviderException("Persistent
failure")).when(storageProvider).persistJob(any());
+
+ CountDownLatch completed = new CountDownLatch(1);
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+ job.asyncResult().onComplete(ar -> completed.countDown());
+
+ assertThat(completed.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider, times(3)).persistJob(any());
+ verify(storageProvider, never()).updateJobStatus(any(), any(),
any(), any());
+ assertThat(tracker.jobsView()).doesNotContainKey(jobId);
+ });
+ }
+}
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobInfoTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobInfoTest.java
index 19b3a9eb..717b6201 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobInfoTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobInfoTest.java
@@ -26,6 +26,7 @@ import org.junit.jupiter.api.Test;
import com.datastax.driver.core.utils.UUIDs;
import io.vertx.core.Future;
import io.vertx.core.Promise;
+import org.apache.cassandra.sidecar.common.data.OperationType;
import
org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException;
import static org.assertj.core.api.Assertions.assertThat;
@@ -96,6 +97,12 @@ class OperationalJobInfoTest
return false;
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
protected Future<Void> executeInternal()
{
@@ -113,6 +120,12 @@ class OperationalJobInfoTest
return false;
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
protected Future<Void> executeInternal() throws
OperationalJobException
{
@@ -132,6 +145,12 @@ class OperationalJobInfoTest
return false;
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
protected Future<Void> executeInternal()
{
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 4007b661..cf7492c2 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
@@ -32,15 +32,24 @@ import com.datastax.driver.core.utils.UUIDs;
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.server.exceptions.OperationalJobException;
+import
org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
import
org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
import org.apache.cassandra.sidecar.exceptions.OperationalJobConflictException;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.apache.cassandra.sidecar.job.storage.StorageProviderException;
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.testing.utils.AssertionUtils.loopAssert;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Tests to validate the Job submission behavior for scenarios which are a
combination of values for
@@ -148,6 +157,12 @@ class OperationalJobManagerTest
return false;
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
protected Future<Void> executeInternal() throws
OperationalJobException
{
@@ -166,4 +181,30 @@ class OperationalJobManagerTest
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(tracker.get(jobId)).isNotNull();
}
+
+ @Test
+ void testJobRemovedFromTrackerWhenPersistenceFails()
+ {
+ StorageProvider storageProvider = mock(StorageProvider.class);
+ when(storageProvider.isAvailable()).thenReturn(true);
+ doThrow(new StorageProviderException("Storage unavailable"))
+ .when(storageProvider).persistJob(any());
+
+ DurableOperationalJobTracker durableTracker = new
DurableOperationalJobTracker(new ServiceConfigurationImpl(),
+
storageProvider,
+
executorPool.service());
+ OperationalJobManager manager = new
OperationalJobManager(durableTracker, executorPool);
+
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ manager.trySubmitJob(job,
+ (j, ex) -> {},
+ executorPool.service(),
+ SecondBoundConfiguration.parse("5s"));
+
+ loopAssert(2, () -> {
+ assertThat(durableTracker.jobsView()).doesNotContainKey(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 c047d9db..485a6481 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
@@ -30,6 +30,7 @@ import io.vertx.core.Future;
import io.vertx.core.Promise;
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.common.server.exceptions.OperationalJobException;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
@@ -78,6 +79,12 @@ class OperationalJobTest
return jobStatus;
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
public String name()
{
@@ -108,6 +115,12 @@ class OperationalJobTest
return jobStatus;
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
public String name()
{
@@ -146,6 +159,12 @@ class OperationalJobTest
return Future.succeededFuture();
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
public String name()
{
@@ -185,6 +204,12 @@ class OperationalJobTest
return false;
}
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.DRAIN;
+ }
+
@Override
protected Future<Void> executeInternal() throws
OperationalJobException
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
