From c09846d5e1f81d979d8d32d0f6c921344c7dfa5d Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Tue, 26 May 2026 15:27:27 -0400 Subject: [PATCH 1/9] CASSSIDECAR-374: Implement durable operational job tracker --- CHANGES.txt | 1 + .../job/DurableOperationalJobTracker.java | 152 ++++++++ .../sidecar/job/NodeDecommissionJob.java | 10 + .../cassandra/sidecar/job/NodeDrainJob.java | 10 + .../cassandra/sidecar/job/NodeMoveJob.java | 10 + .../cassandra/sidecar/job/OperationalJob.java | 7 + .../sidecar/job/OperationalJobInfo.java | 6 + .../sidecar/job/OperationalJobManager.java | 9 +- .../cassandra/sidecar/job/RepairJob.java | 10 + .../job/storage/OperationalJobRecord.java | 72 +++- ...eOperationalJobTrackerIntegrationTest.java | 105 ++++++ .../ListOperationalJobsHandlerTest.java | 7 + .../job/DurableOperationalJobTrackerTest.java | 332 ++++++++++++++++++ .../sidecar/job/OperationalJobInfoTest.java | 19 + .../job/OperationalJobManagerTest.java | 39 ++ .../sidecar/job/OperationalJobTest.java | 25 ++ 16 files changed, 809 insertions(+), 5 deletions(-) create mode 100644 server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java create mode 100644 server/src/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java create mode 100644 server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 07239ad94..265a4a997 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/job/DurableOperationalJobTracker.java b/server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java new file mode 100644 index 000000000..51a77ebec --- /dev/null +++ b/server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java @@ -0,0 +1,152 @@ +/* + * 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.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +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.config.ServiceConfiguration; +import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord; +import org.apache.cassandra.sidecar.job.storage.StorageProvider; +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. + */ +@Singleton +public class DurableOperationalJobTracker implements OperationalJobTracker +{ + private static final Logger LOGGER = LoggerFactory.getLogger(DurableOperationalJobTracker.class); + private static final int MAX_STATUS_UPDATE_ATTEMPTS = 2; + private static final long RETRY_DELAY_MS = 100; + + private final ConcurrentHashMap liveJobs; + private final StorageProvider storageProvider; + + @Inject + public DurableOperationalJobTracker(ServiceConfiguration serviceConfiguration, + StorageProvider storageProvider) + { + this.liveJobs = new ConcurrentHashMap<>(serviceConfiguration.operationalJobTrackerSize()); + this.storageProvider = storageProvider; + } + + /** + * {@inheritDoc} + */ + @Override + public OperationalJob computeIfAbsent(UUID jobId, Function mappingFunction) + { + return liveJobs.computeIfAbsent(jobId, id -> { + OperationalJob job = mappingFunction.apply(id); + + storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job)); + + job.asyncResult().onComplete(ar -> { + liveJobs.remove(job.jobId()); + updateTerminalStatus(job); + }); + + return job; + }); + } + + @Nullable + @Override + public OperationalJobInfo get(UUID jobId) + { + OperationalJob liveJob = liveJobs.get(jobId); + if (liveJob != null) + { + return liveJob; + } + + return storageProvider.findJob(jobId); + } + + @NotNull + @Override + public Map jobsView() + { + return Collections.unmodifiableMap(liveJobs); + } + + @NotNull + @Override + public List inflightJobsByOperation(String operation) + { + return liveJobs.values() + .stream() + .filter(j -> j.name().equals(operation) && + (j.status() == OperationalJobStatus.RUNNING || + j.status() == OperationalJobStatus.CREATED)) + .collect(Collectors.toList()); + } + + /** + * Attempts to update the terminal status in storage with retry. + * If all attempts fail, logs a warning and continues. + */ + private void updateTerminalStatus(OperationalJob job) + { + OperationalJobStatus terminalStatus = job.status(); + String failureReason = job.failureReason(); + for (int attempt = 1; attempt <= MAX_STATUS_UPDATE_ATTEMPTS; attempt++) + { + try + { + storageProvider.updateJobStatus(job.jobId(), job.operationType(), terminalStatus, failureReason); + return; + } + catch (RuntimeException e) + { + LOGGER.warn("Failed to update terminal status for job {} (attempt {}/{}). error={}", + job.jobId(), attempt, MAX_STATUS_UPDATE_ATTEMPTS, e.getMessage()); + if (attempt < MAX_STATUS_UPDATE_ATTEMPTS) + { + try + { + Thread.sleep(RETRY_DELAY_MS * attempt); + } + catch (InterruptedException ie) + { + Thread.currentThread().interrupt(); + return; + } + } + } + } + } +} 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 e042200f6..c67026ac1 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.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 @@ protected Future executeInternal() 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 b5b1db924..52d44bfa0 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.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 @@ protected Future executeInternal() 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 c35f13ef2..5fa04f30b 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.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 @@ protected Future executeInternal() throws Exception 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 58320e4ec..f3c15c6ac 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 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 String name() 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. *

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 1748e3db6..a6fc8a951 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.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 ccaed3746..a1e4d406b 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 void trySubmitJob(OperationalJob job, { 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 a9fe4ccbc..156fa781e 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 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 OperationalJobStatus status() 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 7e3d508a4..8b6de0181 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,6 +19,7 @@ package org.apache.cassandra.sidecar.job.storage; import java.time.Instant; +import java.util.Collections; import java.util.List; import java.util.Map; @@ -28,12 +29,17 @@ 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; @@ -173,6 +179,70 @@ public Map operationMetadata() return operationMetadata; } + @Override + @Nullable + public UUID nodeId() + { + return null; + } + + @Override + public String name() + { + return operationType.name(); + } + + @Override + public long creationTime() + { + return creationTimeMillis; + } + + @Override + @NotNull + public List nodesPending() + { + return Collections.emptyList(); + } + + @Override + @NotNull + public List nodesExecuting() + { + return Collections.emptyList(); + } + + @Override + @NotNull + public List nodesSucceeded() + { + return Collections.emptyList(); + } + + @Override + @NotNull + public List nodesFailed() + { + return Collections.emptyList(); + } + + @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 new OperationalJobRecord(job.jobId(), job.operationType(), job.status()); + } + @Override public String toString() { 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 000000000..33a611f1e --- /dev/null +++ b/server/src/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java @@ -0,0 +1,105 @@ +/* + * 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(); + + DurableOperationalJobTracker tracker = new DurableOperationalJobTracker(new ServiceConfigurationImpl(), + storageProvider); + ExecutorPools executorPools = injector.getInstance(ExecutorPools.class); + TaskExecutorPool executorPool = executorPools.internal(); + + UUID jobId1 = UUIDs.timeBased(); + OperationalJob job1 = OperationalJobTest.createOperationalJob(jobId1, OperationalJobStatus.CREATED); + + tracker.computeIfAbsent(jobId1, id -> job1); + + 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 91159bd0c..c31f0bd3b 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.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 @@ public boolean hasConflict(List jobs) return false; } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override protected Future 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 000000000..75dcdc8bd --- /dev/null +++ b/server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java @@ -0,0 +1,332 @@ +/* + * 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.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.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.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); + tracker = new DurableOperationalJobTracker(new ServiceConfigurationImpl(), storageProvider); + vertx = Vertx.vertx(); + executorPools = new ExecutorPools(vertx, new ServiceConfigurationImpl()); + executorPool = executorPools.internal(); + } + + @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); + 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); + 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.DECOMMISSION), eq(SUCCEEDED), isNull()); + }); + } + + @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.DECOMMISSION), 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 = new OperationalJobRecord(jobId, OperationType.DECOMMISSION, SUCCEEDED); + when(storageProvider.findJob(jobId)).thenReturn(record); + + 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()); + 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 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 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 inflightDecommission = tracker.inflightJobsByOperation("decommission"); + + assertThat(inflightDecommission) + .hasSize(2) + .containsExactlyInAnyOrder(createdJob, runningJob); + verify(storageProvider, never()).findJob(any()); + } + + @Test + void testComputeIfAbsentPropagatesPersistenceFailure() + { + doThrow(new StorageProviderException("Storage unavailable")) + .when(storageProvider).persistJob(any()); + OperationalJob job = createOperationalJob(CREATED); + + assertThatThrownBy(() -> tracker.computeIfAbsent(job.jobId(), id -> job)) + .isInstanceOf(StorageProviderException.class) + .hasMessage("Storage unavailable"); + } + + @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 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.DECOMMISSION), 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.DECOMMISSION), 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.DECOMMISSION), 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.DECOMMISSION), eq(SUCCEEDED), isNull()); + 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 19b3a9ebe..ccc393293 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 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 @@ public boolean hasConflict(List sameOperationJobs) return false; } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override protected Future executeInternal() { @@ -113,6 +120,12 @@ public boolean hasConflict(List sameOperationJobs) return false; } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override protected Future executeInternal() throws OperationalJobException { @@ -132,6 +145,12 @@ public boolean hasConflict(List sameOperationJobs) return false; } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override protected Future 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 4007b6612..af959a0d5 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,23 @@ 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.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; /** * Tests to validate the Job submission behavior for scenarios which are a combination of values for @@ -148,6 +156,12 @@ public boolean hasConflict(List jobs) return false; } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override protected Future executeInternal() throws OperationalJobException { @@ -166,4 +180,29 @@ protected Future executeInternal() throws OperationalJobException assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(tracker.get(jobId)).isNotNull(); } + + @Test + void testJobNotExecutedWhenPersistenceFails() + { + StorageProvider storageProvider = mock(StorageProvider.class); + doThrow(new StorageProviderException("Storage unavailable")) + .when(storageProvider).persistJob(any()); + + DurableOperationalJobTracker durableTracker = new DurableOperationalJobTracker(new ServiceConfigurationImpl(), + storageProvider); + OperationalJobManager manager = new OperationalJobManager(durableTracker, executorPool); + + UUID jobId = UUIDs.timeBased(); + OperationalJob job = OperationalJobTest.createOperationalJob(jobId, MillisecondBoundConfiguration.parse("50ms")); + + assertThatThrownBy(() -> manager.trySubmitJob(job, + (j, ex) -> {}, + executorPool.service(), + SecondBoundConfiguration.parse("5s"))) + .isInstanceOf(StorageProviderException.class); + + assertThat(job.isExecuting()).isFalse(); + assertThat(job.asyncResult().isComplete()).isFalse(); + 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 c047d9dbc..21d10f0af 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.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 @@ public OperationalJobStatus status() return jobStatus; } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override public String name() { @@ -108,6 +115,12 @@ public OperationalJobStatus status() return jobStatus; } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override public String name() { @@ -146,6 +159,12 @@ protected Future executeInternal() throws OperationalJobException return Future.succeededFuture(); } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override public String name() { @@ -185,6 +204,12 @@ public boolean hasConflict(List jobs) return false; } + @Override + public OperationType operationType() + { + return OperationType.DECOMMISSION; + } + @Override protected Future executeInternal() throws OperationalJobException { From 31e787aca3b40c0261100404e59ad642d17c07fd Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Wed, 3 Jun 2026 15:46:38 -0400 Subject: [PATCH 2/9] Address PR review comments --- .../job/DurableOperationalJobTracker.java | 45 ++++++++++--------- ...eOperationalJobTrackerIntegrationTest.java | 5 ++- .../ListOperationalJobsHandlerTest.java | 2 +- .../job/DurableOperationalJobTrackerTest.java | 14 +++--- .../sidecar/job/OperationalJobInfoTest.java | 6 +-- .../job/OperationalJobManagerTest.java | 5 ++- .../sidecar/job/OperationalJobTest.java | 8 ++-- 7 files changed, 44 insertions(+), 41 deletions(-) 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 51a77ebec..1eef63d7f 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 @@ -32,6 +32,7 @@ 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; @@ -54,13 +55,16 @@ public class DurableOperationalJobTracker implements OperationalJobTracker private final ConcurrentHashMap liveJobs; private final StorageProvider storageProvider; + private final TaskExecutorPool executor; @Inject public DurableOperationalJobTracker(ServiceConfiguration serviceConfiguration, - StorageProvider storageProvider) + StorageProvider storageProvider, + TaskExecutorPool executor) { this.liveJobs = new ConcurrentHashMap<>(serviceConfiguration.operationalJobTrackerSize()); this.storageProvider = storageProvider; + this.executor = executor; } /** @@ -121,31 +125,28 @@ public List inflightJobsByOperation(String operation) */ private void updateTerminalStatus(OperationalJob job) { - OperationalJobStatus terminalStatus = job.status(); - String failureReason = job.failureReason(); - for (int attempt = 1; attempt <= MAX_STATUS_UPDATE_ATTEMPTS; attempt++) + updateTerminalStatus(job, 1); + } + + private void updateTerminalStatus(OperationalJob job, int attempt) + { + try + { + storageProvider.updateJobStatus(job.jobId(), job.operationType(), job.status(), job.failureReason()); + } + catch (RuntimeException e) { - try + LOGGER.warn("Failed to update terminal status for job {} (attempt {}/{}). error={}", + job.jobId(), attempt, MAX_STATUS_UPDATE_ATTEMPTS, e.getMessage()); + if (attempt < MAX_STATUS_UPDATE_ATTEMPTS) { - storageProvider.updateJobStatus(job.jobId(), job.operationType(), terminalStatus, failureReason); - return; + executor.setTimer(RETRY_DELAY_MS * attempt, + id -> updateTerminalStatus(job, attempt + 1)); } - catch (RuntimeException e) + else { - LOGGER.warn("Failed to update terminal status for job {} (attempt {}/{}). error={}", - job.jobId(), attempt, MAX_STATUS_UPDATE_ATTEMPTS, e.getMessage()); - if (attempt < MAX_STATUS_UPDATE_ATTEMPTS) - { - try - { - Thread.sleep(RETRY_DELAY_MS * attempt); - } - catch (InterruptedException ie) - { - Thread.currentThread().interrupt(); - return; - } - } + 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/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java b/server/src/test/integration/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerIntegrationTest.java index 33a611f1e..97b9424cc 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 @@ -49,10 +49,11 @@ void testDurableOperationalJobTrackerOperations() throws InterruptedException StorageProvider storageProvider = injector.getInstance(StorageProvider.class); storageProvider.initialize(); - DurableOperationalJobTracker tracker = new DurableOperationalJobTracker(new ServiceConfigurationImpl(), - storageProvider); 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); 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 c31f0bd3b..8edd2d82c 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 @@ -161,7 +161,7 @@ public boolean hasConflict(List jobs) @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override 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 75dcdc8bd..e57fdb2a3 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 @@ -74,10 +74,10 @@ class DurableOperationalJobTrackerTest void setUp() { storageProvider = mock(StorageProvider.class); - tracker = new DurableOperationalJobTracker(new ServiceConfigurationImpl(), storageProvider); vertx = Vertx.vertx(); executorPools = new ExecutorPools(vertx, new ServiceConfigurationImpl()); executorPool = executorPools.internal(); + tracker = new DurableOperationalJobTracker(new ServiceConfigurationImpl(), storageProvider, executorPool); } @AfterEach @@ -127,7 +127,7 @@ void testComputeIfAbsentUpdatesStatusOnCompletion() throws InterruptedException assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); loopAssert(2, () -> { - verify(storageProvider).updateJobStatus(eq(jobId), eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull()); + verify(storageProvider).updateJobStatus(eq(jobId), eq(OperationType.DRAIN), eq(SUCCEEDED), isNull()); }); } @@ -147,7 +147,7 @@ void testComputeIfAbsentUpdatesStatusOnFailure() throws InterruptedException assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); loopAssert(2, () -> { - verify(storageProvider).updateJobStatus(eq(jobId), eq(OperationType.DECOMMISSION), eq(FAILED), eq("test failure")); + verify(storageProvider).updateJobStatus(eq(jobId), eq(OperationType.DRAIN), eq(FAILED), eq("test failure")); }); } @@ -293,7 +293,7 @@ void testOnCompleteRetrySucceedsAfterTransientFailure() throws InterruptedExcept doThrow(new StorageProviderException("Transient failure")) .doNothing() - .when(storageProvider).updateJobStatus(eq(jobId), eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull()); + .when(storageProvider).updateJobStatus(eq(jobId), eq(OperationType.DRAIN), eq(SUCCEEDED), isNull()); CountDownLatch latch = new CountDownLatch(1); @@ -304,7 +304,7 @@ void testOnCompleteRetrySucceedsAfterTransientFailure() throws InterruptedExcept assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); loopAssert(2, () -> { - verify(storageProvider, times(2)).updateJobStatus(eq(jobId), eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull()); + verify(storageProvider, times(2)).updateJobStatus(eq(jobId), eq(OperationType.DRAIN), eq(SUCCEEDED), isNull()); }); } @@ -315,7 +315,7 @@ void testOnCompleteRemovesJobFromLocalMapWhenAllRetriesFail() throws Interrupted OperationalJob job = OperationalJobTest.createOperationalJob(jobId, MillisecondBoundConfiguration.parse("50ms")); doThrow(new StorageProviderException("Persistent failure")) - .when(storageProvider).updateJobStatus(eq(jobId), eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull()); + .when(storageProvider).updateJobStatus(eq(jobId), eq(OperationType.DRAIN), eq(SUCCEEDED), isNull()); CountDownLatch latch = new CountDownLatch(1); tracker.computeIfAbsent(jobId, id -> job); @@ -325,7 +325,7 @@ void testOnCompleteRemovesJobFromLocalMapWhenAllRetriesFail() throws Interrupted assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); loopAssert(2, () -> { - verify(storageProvider, times(2)).updateJobStatus(eq(jobId), eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull()); + verify(storageProvider, times(2)).updateJobStatus(eq(jobId), eq(OperationType.DRAIN), eq(SUCCEEDED), isNull()); 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 ccc393293..717b6201f 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 @@ -100,7 +100,7 @@ public boolean hasConflict(List sameOperationJobs) @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override @@ -123,7 +123,7 @@ public boolean hasConflict(List sameOperationJobs) @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override @@ -148,7 +148,7 @@ public boolean hasConflict(List sameOperationJobs) @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override 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 af959a0d5..e86d771a0 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 @@ -159,7 +159,7 @@ public boolean hasConflict(List jobs) @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override @@ -189,7 +189,8 @@ void testJobNotExecutedWhenPersistenceFails() .when(storageProvider).persistJob(any()); DurableOperationalJobTracker durableTracker = new DurableOperationalJobTracker(new ServiceConfigurationImpl(), - storageProvider); + storageProvider, + executorPool.service()); OperationalJobManager manager = new OperationalJobManager(durableTracker, executorPool); UUID jobId = UUIDs.timeBased(); 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 21d10f0af..485a64819 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 @@ -82,7 +82,7 @@ public OperationalJobStatus status() @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override @@ -118,7 +118,7 @@ public OperationalJobStatus status() @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override @@ -162,7 +162,7 @@ protected Future executeInternal() throws OperationalJobException @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override @@ -207,7 +207,7 @@ public boolean hasConflict(List jobs) @Override public OperationType operationType() { - return OperationType.DECOMMISSION; + return OperationType.DRAIN; } @Override From 4be1de3de7ddee33ef1dba775abb6d8e78806a74 Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Wed, 1 Jul 2026 14:34:24 -0400 Subject: [PATCH 3/9] Populate nodes list fields in OperationalJobRecord via DurableOperationalJobTracker --- .../job/DurableOperationalJobTracker.java | 65 +++++++++++++- .../job/storage/OperationalJobRecord.java | 59 +++++++++++-- .../job/DurableOperationalJobTrackerTest.java | 86 +++++++++++++++++++ 3 files changed, 204 insertions(+), 6 deletions(-) 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 1eef63d7f..4bae8ffcb 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,6 +18,7 @@ package org.apache.cassandra.sidecar.job; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; @@ -97,7 +98,12 @@ public OperationalJobInfo get(UUID jobId) return liveJob; } - return storageProvider.findJob(jobId); + OperationalJobRecord record = storageProvider.findJob(jobId); + if (record == null) + { + return null; + } + return enrichWithNodeStatuses(record); } @NotNull @@ -119,6 +125,63 @@ public List inflightJobsByOperation(String operation) .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 nodeStatuses = + storageProvider.getNodeStatusesForOperation(record.jobId()); + if (nodeStatuses.isEmpty()) + { + return record; + } + + List pending = new ArrayList<>(); + List executing = new ArrayList<>(); + List succeeded = new ArrayList<>(); + List failed = new ArrayList<>(); + + for (Map.Entry 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: + break; + } + } + + return new OperationalJobRecord(record.jobId(), record.operationType(), record.status(), + record.startTime(), record.lastUpdate(), record.failureReason(), + record.nodeExecutionOrder(), record.operationMetadata(), + Collections.unmodifiableList(pending), + Collections.unmodifiableList(executing), + Collections.unmodifiableList(succeeded), + Collections.unmodifiableList(failed)); + } + /** * Attempts to update the terminal status in storage with retry. * If all attempts fail, logs a warning and continues. 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 8b6de0181..8d4d338f0 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 @@ -55,6 +55,14 @@ public class OperationalJobRecord implements OperationalJobInfo private final List> nodeExecutionOrder; @Nullable private final Map operationMetadata; + @NotNull + private final List nodesPending; + @NotNull + private final List nodesExecuting; + @NotNull + private final List nodesSucceeded; + @NotNull + private final List nodesFailed; /** * Constructs an OperationalJobRecord with the given fields. @@ -69,7 +77,7 @@ public OperationalJobRecord(UUID jobId, OperationType operationType, Operational } /** - * Constructs an OperationalJobRecord with all fields. + * Constructs an OperationalJobRecord with all fields except per-node status lists. * * @param jobId time-based v1 UUID identifying the job * @param operationType the operation type @@ -86,6 +94,39 @@ public OperationalJobRecord(UUID jobId, OperationType operationType, Operational @Nullable String failureReason, @Nullable List> nodeExecutionOrder, @Nullable Map operationMetadata) + { + this(jobId, operationType, status, startTime, lastUpdate, failureReason, + nodeExecutionOrder, operationMetadata, + Collections.emptyList(), Collections.emptyList(), + Collections.emptyList(), Collections.emptyList()); + } + + /** + * Constructs an OperationalJobRecord with all fields including per-node status lists. + * + * @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 + * @param nodesPending node UUIDs with CREATED status + * @param nodesExecuting node UUIDs with RUNNING status + * @param nodesSucceeded node UUIDs with SUCCEEDED status + * @param nodesFailed node UUIDs with FAILED status + */ + public OperationalJobRecord(UUID jobId, OperationType operationType, OperationalJobStatus status, + @Nullable Instant startTime, + @Nullable Instant lastUpdate, + @Nullable String failureReason, + @Nullable List> nodeExecutionOrder, + @Nullable Map operationMetadata, + @NotNull List nodesPending, + @NotNull List nodesExecuting, + @NotNull List nodesSucceeded, + @NotNull List nodesFailed) { Preconditions.checkArgument(jobId != null, "jobId must not be null"); Preconditions.checkArgument(jobId.version() == 1, "jobId must be a time-based (v1) UUID"); @@ -100,6 +141,10 @@ public OperationalJobRecord(UUID jobId, OperationType operationType, Operational this.failureReason = failureReason; this.nodeExecutionOrder = nodeExecutionOrder; this.operationMetadata = operationMetadata; + this.nodesPending = nodesPending; + this.nodesExecuting = nodesExecuting; + this.nodesSucceeded = nodesSucceeded; + this.nodesFailed = nodesFailed; } /** @@ -202,28 +247,28 @@ public long creationTime() @NotNull public List nodesPending() { - return Collections.emptyList(); + return nodesPending; } @Override @NotNull public List nodesExecuting() { - return Collections.emptyList(); + return nodesExecuting; } @Override @NotNull public List nodesSucceeded() { - return Collections.emptyList(); + return nodesSucceeded; } @Override @NotNull public List nodesFailed() { - return Collections.emptyList(); + return nodesFailed; } @Override @@ -256,6 +301,10 @@ public String toString() ", failureReason=" + failureReason + ", nodeExecutionOrder=" + nodeExecutionOrder + ", operationMetadata=" + operationMetadata + + ", nodesPending=" + nodesPending + + ", nodesExecuting=" + nodesExecuting + + ", nodesSucceeded=" + nodesSucceeded + + ", nodesFailed=" + nodesFailed + '}'; } } 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 e57fdb2a3..cb6a644e0 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 @@ -18,6 +18,8 @@ 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; @@ -32,6 +34,7 @@ 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; @@ -189,6 +192,7 @@ void testGetFallsBackToStorage() UUID jobId = UUIDs.timeBased(); OperationalJobRecord record = new OperationalJobRecord(jobId, OperationType.DECOMMISSION, SUCCEEDED); when(storageProvider.findJob(jobId)).thenReturn(record); + when(storageProvider.getNodeStatusesForOperation(jobId)).thenReturn(Collections.emptyMap()); OperationalJobInfo result = tracker.get(jobId); @@ -285,6 +289,88 @@ void testGetPropagatesStorageFallbackFailure() .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 = new OperationalJobRecord(jobId, OperationType.DECOMMISSION, RUNNING); + when(storageProvider.findJob(jobId)).thenReturn(record); + + Map 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 = new OperationalJobRecord(jobId, OperationType.DECOMMISSION, SUCCEEDED, + null, null, null, null, null, + Collections.emptyList(), + Collections.emptyList(), + Collections.singletonList(succeededNode), + Collections.emptyList()); + 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 = new OperationalJobRecord(jobId, OperationType.DECOMMISSION, SUCCEEDED); + 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 = new OperationalJobRecord(jobId, OperationType.DECOMMISSION, SUCCEEDED); + 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 { From 3f52509433c9a0f40ff7a1efec086c9e5e301654 Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Wed, 1 Jul 2026 15:17:24 -0400 Subject: [PATCH 4/9] Return lower case name from OperationalJobRecord --- .../cassandra/sidecar/job/storage/OperationalJobRecord.java | 2 +- .../cassandra/sidecar/job/DurableOperationalJobTrackerTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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 8d4d338f0..2a68c2ea3 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 @@ -234,7 +234,7 @@ public UUID nodeId() @Override public String name() { - return operationType.name(); + return operationType.name().toLowerCase(); } @Override 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 cb6a644e0..2bfd462fe 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 @@ -199,7 +199,7 @@ void testGetFallsBackToStorage() assertThat(result).isInstanceOf(OperationalJobRecord.class); assertThat(result.jobId()).isEqualTo(jobId); assertThat(result.status()).isEqualTo(SUCCEEDED); - assertThat(result.name()).isEqualTo(OperationType.DECOMMISSION.name()); + assertThat(result.name()).isEqualTo(OperationType.DECOMMISSION.name().toLowerCase()); assertThat(result.operationType()).isEqualTo(OperationType.DECOMMISSION); verify(storageProvider).findJob(jobId); } From 6648515a924ba51b4ed5c5c9a116ecc90c227dad Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Wed, 1 Jul 2026 16:02:05 -0400 Subject: [PATCH 5/9] Remove from live jobs after updateTerminalStatus --- .../cassandra/sidecar/job/DurableOperationalJobTracker.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 4bae8ffcb..6ccabc40a 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 @@ -80,8 +80,8 @@ public OperationalJob computeIfAbsent(UUID jobId, Function storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job)); job.asyncResult().onComplete(ar -> { - liveJobs.remove(job.jobId()); updateTerminalStatus(job); + liveJobs.remove(job.jobId()); }); return job; From af8364c4f0082bda459bf8e10e59a1d92913f173 Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Wed, 1 Jul 2026 16:13:49 -0400 Subject: [PATCH 6/9] Move job persistence outside of CHM lock and event loop, add storage provider unavailable guard --- .../job/DurableOperationalJobTracker.java | 27 +++++++++++--- ...eOperationalJobTrackerIntegrationTest.java | 20 ++++++----- .../job/DurableOperationalJobTrackerTest.java | 36 ++++++++++++++----- 3 files changed, 60 insertions(+), 23 deletions(-) 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 6ccabc40a..1ac6d0624 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 @@ -74,18 +74,35 @@ public DurableOperationalJobTracker(ServiceConfiguration serviceConfiguration, @Override public OperationalJob computeIfAbsent(UUID jobId, Function mappingFunction) { - return liveJobs.computeIfAbsent(jobId, id -> { - OperationalJob job = mappingFunction.apply(id); + if (!storageProvider.isAvailable()) + { + throw new IllegalStateException("Storage provider is not available"); + } + + boolean[] created = {false}; + OperationalJob job = liveJobs.computeIfAbsent(jobId, id -> { + created[0] = true; + return mappingFunction.apply(id); + }); - storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job)); + if (created[0]) + { + executor.executeBlocking(() -> { + storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job)); + return null; + }).onFailure(e -> { + liveJobs.remove(jobId, job); + LOGGER.error("Failed to persist job {} to storage. Job will not be tracked durably.", + jobId, e); + }); job.asyncResult().onComplete(ar -> { updateTerminalStatus(job); liveJobs.remove(job.jobId()); }); + } - return job; - }); + return job; } @Nullable 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 97b9424cc..155161d9c 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 @@ -60,15 +60,17 @@ void testDurableOperationalJobTrackerOperations() throws InterruptedException tracker.computeIfAbsent(jobId1, id -> job1); - 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); - }); + 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")); 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 2bfd462fe..ac98e30fd 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 @@ -77,6 +77,7 @@ class DurableOperationalJobTrackerTest void setUp() { storageProvider = mock(StorageProvider.class); + when(storageProvider.isAvailable()).thenReturn(true); vertx = Vertx.vertx(); executorPools = new ExecutorPools(vertx, new ServiceConfigurationImpl()); executorPool = executorPools.internal(); @@ -97,11 +98,13 @@ void testComputeIfAbsentPersistsNewJob() OperationalJob result = tracker.computeIfAbsent(job.jobId(), id -> job); assertThat(result).isSameAs(job); - verify(storageProvider).persistJob(argThat(record -> - record.jobId().equals(job.jobId()) && - record.operationType() == job.operationType() && - record.status() == CREATED - )); + loopAssert(2, () -> { + verify(storageProvider).persistJob(argThat(record -> + record.jobId().equals(job.jobId()) && + record.operationType() == job.operationType() && + record.status() == CREATED + )); + }); } @Test @@ -113,7 +116,9 @@ void testComputeIfAbsentDoesNotPersistExistingJob() OperationalJob secondCall = tracker.computeIfAbsent(job.jobId(), id -> createOperationalJob(CREATED)); assertThat(secondCall).isSameAs(job); - verify(storageProvider, times(1)).persistJob(any()); + loopAssert(2, () -> { + verify(storageProvider, times(1)).persistJob(any()); + }); } @Test @@ -267,15 +272,28 @@ void testInflightJobsByOperationFiltersLocalMap() } @Test - void testComputeIfAbsentPropagatesPersistenceFailure() + void testComputeIfAbsentRemovesJobFromMapOnPersistenceFailure() { doThrow(new StorageProviderException("Storage unavailable")) .when(storageProvider).persistJob(any()); OperationalJob job = createOperationalJob(CREATED); + tracker.computeIfAbsent(job.jobId(), id -> job); + + loopAssert(2, () -> { + assertThat(tracker.jobsView()).doesNotContainKey(job.jobId()); + }); + } + + @Test + void testComputeIfAbsentFailsWhenStorageUnavailable() + { + when(storageProvider.isAvailable()).thenReturn(false); + OperationalJob job = createOperationalJob(CREATED); + assertThatThrownBy(() -> tracker.computeIfAbsent(job.jobId(), id -> job)) - .isInstanceOf(StorageProviderException.class) - .hasMessage("Storage unavailable"); + .isInstanceOf(IllegalStateException.class) + .hasMessage("Storage provider is not available"); } @Test From c7256ca5d62e8cb3f30ac691cb39d0df986812af Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Wed, 1 Jul 2026 16:32:28 -0400 Subject: [PATCH 7/9] Update MAX_STATUS_UPDATE_ATTEMPTS to 3 --- .../cassandra/sidecar/job/DurableOperationalJobTracker.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 1ac6d0624..e7c6e21e6 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 @@ -51,7 +51,7 @@ public class DurableOperationalJobTracker implements OperationalJobTracker { private static final Logger LOGGER = LoggerFactory.getLogger(DurableOperationalJobTracker.class); - private static final int MAX_STATUS_UPDATE_ATTEMPTS = 2; + private static final int MAX_STATUS_UPDATE_ATTEMPTS = 3; private static final long RETRY_DELAY_MS = 100; private final ConcurrentHashMap liveJobs; From a0a08ea925d5b7baa62c06d9b2ed7d2536ce408b Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Wed, 1 Jul 2026 17:17:42 -0400 Subject: [PATCH 8/9] Fix persistence failure test --- .../job/OperationalJobManagerTest.java | 21 ++++++++++--------- 1 file changed, 11 insertions(+), 10 deletions(-) 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 e86d771a0..cf7492c2c 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 @@ -44,11 +44,12 @@ 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.assertj.core.api.Assertions.assertThatThrownBy; 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 @@ -182,9 +183,10 @@ protected Future executeInternal() throws OperationalJobException } @Test - void testJobNotExecutedWhenPersistenceFails() + void testJobRemovedFromTrackerWhenPersistenceFails() { StorageProvider storageProvider = mock(StorageProvider.class); + when(storageProvider.isAvailable()).thenReturn(true); doThrow(new StorageProviderException("Storage unavailable")) .when(storageProvider).persistJob(any()); @@ -196,14 +198,13 @@ void testJobNotExecutedWhenPersistenceFails() UUID jobId = UUIDs.timeBased(); OperationalJob job = OperationalJobTest.createOperationalJob(jobId, MillisecondBoundConfiguration.parse("50ms")); - assertThatThrownBy(() -> manager.trySubmitJob(job, - (j, ex) -> {}, - executorPool.service(), - SecondBoundConfiguration.parse("5s"))) - .isInstanceOf(StorageProviderException.class); + manager.trySubmitJob(job, + (j, ex) -> {}, + executorPool.service(), + SecondBoundConfiguration.parse("5s")); - assertThat(job.isExecuting()).isFalse(); - assertThat(job.asyncResult().isComplete()).isFalse(); - assertThat(durableTracker.jobsView()).doesNotContainKey(jobId); + loopAssert(2, () -> { + assertThat(durableTracker.jobsView()).doesNotContainKey(jobId); + }); } } From 373e3670fba86cc68ad75df4c5c95b6b0100700c Mon Sep 17 00:00:00 2001 From: abeckruiz Date: Wed, 8 Jul 2026 17:24:47 -0400 Subject: [PATCH 9/9] Update terminal status after successful persist --- .../job/DurableOperationalJobTracker.java | 10 ++++------ .../job/DurableOperationalJobTrackerTest.java | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+), 6 deletions(-) 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 e7c6e21e6..a5c2d67e3 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 @@ -90,16 +90,14 @@ public OperationalJob computeIfAbsent(UUID jobId, Function executor.executeBlocking(() -> { storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job)); return null; - }).onFailure(e -> { + }).onSuccess(v -> job.asyncResult().onComplete(ar -> { + updateTerminalStatus(job); + liveJobs.remove(job.jobId()); + })).onFailure(e -> { liveJobs.remove(jobId, job); LOGGER.error("Failed to persist job {} to storage. Job will not be tracked durably.", jobId, e); }); - - job.asyncResult().onComplete(ar -> { - updateTerminalStatus(job); - liveJobs.remove(job.jobId()); - }); } return job; 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 ac98e30fd..a669bf7bb 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 @@ -139,6 +139,25 @@ void testComputeIfAbsentUpdatesStatusOnCompletion() throws InterruptedException }); } + @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 registers its completion handler before this test does, so a stale + // updateJobStatus would already have run once the job has completed. + verify(storageProvider, never()).updateJobStatus(any(), any(), any(), any()); + } + @Test void testComputeIfAbsentUpdatesStatusOnFailure() throws InterruptedException {