diff --git a/CHANGES.txt b/CHANGES.txt index 265a4a997..591667151 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * Implement job coordination for cluster-wide operations (CASSSIDECAR-377) * Implement durable operational job tracker (CASSSIDECAR-374) * Remove filesystem path from Http response (CASSSIDECAR-477) * Add basic configuration retrieval logic to ConfigurationManager (CASSSIDECAR-427) diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/DisabledOperationalJobCoordinator.java b/server/src/main/java/org/apache/cassandra/sidecar/job/DisabledOperationalJobCoordinator.java new file mode 100644 index 000000000..994e5f6ea --- /dev/null +++ b/server/src/main/java/org/apache/cassandra/sidecar/job/DisabledOperationalJobCoordinator.java @@ -0,0 +1,68 @@ +/* + * 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.Map; +import java.util.UUID; + +import org.apache.cassandra.sidecar.common.data.OperationType; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * An {@link OperationalJobCoordinator} used when a Sidecar instance is not configured to support + * coordinated cluster-wide operations. + *

+ * Jobs that do not require coordination ({@link OperationalJob#requiresCoordination()} returns + * {@code false}) never reach this coordinator, so uncoordinated operations (e.g. decommission) are + * unaffected. + */ +public class DisabledOperationalJobCoordinator implements OperationalJobCoordinator +{ + private static final String NOT_SUPPORTED_MESSAGE = + "Operational job coordination is not supported by this Sidecar instance. " + + "Configure a coordinator to enable coordinated cluster-wide operations."; + + @Override + public boolean trySetActive(OperationType operationType, UUID operationId) + { + throw new UnsupportedOperationException(NOT_SUPPORTED_MESSAGE); + } + + @Override + public boolean clearActive(OperationType operationType, UUID operationId) + { + throw new UnsupportedOperationException(NOT_SUPPORTED_MESSAGE); + } + + @Override + @Nullable + public UUID getActiveOperation(OperationType operationType) + { + return null; + } + + @Override + @NotNull + public Map getActiveOperations() + { + return Collections.emptyMap(); + } +} 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 f3c15c6ac..bdfa377c8 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 @@ -183,6 +183,24 @@ public boolean requiresCoordination() return false; } + /** + * Whether the manager should release the active operation lock (via + * {@link OperationalJobCoordinator#clearActive}) when this job completes locally. Only consulted for jobs that + * {@link #requiresCoordination() require coordination}. + *

+ * Single-node coordinated jobs return {@code true} (the default): the Sidecar that acquires the lock also + * finishes the work, so releasing on local completion is correct. Distributed cluster-wide jobs whose work + * finishes on other nodes return {@code false} and rely on the orchestration layer to call + * {@link OperationalJobCoordinator#clearActive} once all nodes reach a terminal state. + * + * @return {@code true} if the manager should release the lock on local completion; {@code false} to defer + * release to the orchestration layer + */ + public boolean releasesOnCompletion() + { + return true; + } + @Override public final Void result() { @@ -343,6 +361,19 @@ public Future asyncResult(TaskExecutorPool executorPool, DurationSpec wait }); } + /** + * Marks the job as failed before it begins executing, for example when it cannot be started due to a + * coordination conflict. The execution result is completed exceptionally so the job reports + * {@link OperationalJobStatus#FAILED} with the supplied reason, and {@link #executeInternal()} is never invoked. + * + * @param cause the reason the job could not be started + */ + public void failToStart(Throwable cause) + { + lastUpdate = Instant.now(); + executionPromise.tryFail(cause); + } + /** * OperationalJob body. The implementation returns a Future representing the job execution. */ diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobCoordinator.java b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobCoordinator.java new file mode 100644 index 000000000..90be1a3f8 --- /dev/null +++ b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobCoordinator.java @@ -0,0 +1,79 @@ +/* + * 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.Map; +import java.util.UUID; + +import org.apache.cassandra.sidecar.common.data.OperationType; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * Coordinates cluster-wide operational jobs by managing active operation state. + * Ensures mutual exclusion so that only one operation of a given type is active + * at a time within a datacenter. + */ +public interface OperationalJobCoordinator +{ + /** + * Attempt to set an operation as active. Implementations must ensure mutual exclusion + * so that only one operation of the given type is active at a time. + * + * @param operationType the type of operation + * @param operationId the unique identifier for this operation + * @return {@code true} if the operation was successfully set as active, + * {@code false} if an operation of the same type is already active + */ + boolean trySetActive(OperationType operationType, UUID operationId); + + /** + * Clear the active operation lock, but only if the provided operation ID matches + * the currently active one. + *

+ * This method is not called by {@code OperationalJobManager} during job submission. + * For cluster-wide operations, the Sidecar that creates the job is not necessarily + * the one that finishes it. Clearing is the responsibility of the orchestration layer + * once all participating nodes have completed their work. + * + * @param operationType the type of operation + * @param operationId the operation ID to clear + * @return {@code true} if the active operation was cleared, + * {@code false} if the provided operation ID did not match the active one + */ + boolean clearActive(OperationType operationType, UUID operationId); + + /** + * Get the active operation ID for a given operation type. + * + * @param operationType the type of operation + * @return the active operation ID, or {@code null} if no operation of this type is active + */ + @Nullable + UUID getActiveOperation(OperationType operationType); + + /** + * Get all active operations. + * + * @return a map of operation type to operation ID for all currently active operations, + * or an empty map if no operations are active + */ + @NotNull + Map getActiveOperations(); +} 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 a1e4d406b..c348c86ae 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 @@ -28,10 +28,12 @@ import com.google.inject.Inject; import com.google.inject.Singleton; +import io.vertx.core.Future; import org.apache.cassandra.sidecar.common.server.utils.DurationSpec; import org.apache.cassandra.sidecar.concurrent.ExecutorPools; import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool; import org.apache.cassandra.sidecar.exceptions.OperationalJobConflictException; +import org.jetbrains.annotations.Nullable; /** * An abstraction of the management and tracking of long-running jobs running on the sidecar. @@ -41,18 +43,24 @@ public class OperationalJobManager { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); private final OperationalJobTracker jobTracker; - + private final OperationalJobCoordinator coordinator; private final TaskExecutorPool internalExecutorPool; /** - * Creates a manager instance with a default sized job-tracker. + * Creates a manager instance with a coordinator for cluster-wide operation mutual exclusion. Instances that + * do not support coordination bind a {@link DisabledOperationalJobCoordinator}, which fails coordination + * requests. * - * @param jobTracker the tracker for the operational jobs + * @param jobTracker the tracker for the operational jobs + * @param coordinator the coordinator for cluster-wide operations */ @Inject - public OperationalJobManager(OperationalJobTracker jobTracker, ExecutorPools executorPools) + public OperationalJobManager(OperationalJobTracker jobTracker, + OperationalJobCoordinator coordinator, + ExecutorPools executorPools) { this.jobTracker = jobTracker; + this.coordinator = coordinator; this.internalExecutorPool = executorPools.internal(); } @@ -99,13 +107,6 @@ public void trySubmitJob(OperationalJob job, try { checkConflict(job); - - // Track the job first, then start execution separately - OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job); - if (tracked == job) - { - internalExecutorPool.executeBlocking(job::execute); - } } catch (OperationalJobConflictException oje) { @@ -113,6 +114,58 @@ public void trySubmitJob(OperationalJob job, return; } + if (!job.requiresCoordination()) + { + trackAndExecute(job, onComplete, serviceExecutorPool, waitTime); + return; + } + + // Acquiring the active operation lock might perform blocking storage I/O so it must + // run off the event loop + acquireActiveOperationLock(job) + .onComplete(ar -> + { + if (ar.succeeded() && Boolean.TRUE.equals(ar.result())) + { + // Distributed jobs that finish on other nodes opt out of auto-release; the orchestration + // layer clears the lock once all nodes reach a terminal state. + if (job.releasesOnCompletion()) + { + job.asyncResult().onComplete(result -> releaseActiveOperationLock(job)); + } + trackAndExecute(job, onComplete, serviceExecutorPool, waitTime); + } + else + { + OperationalJobConflictException conflict = coordinationConflict(job, ar.cause()); + job.failToStart(conflict); + jobTracker.computeIfAbsent(job.jobId(), jobId -> job); + onComplete.accept(job, conflict); + } + }); + } + + /** + * Tracks the job and submits it for asynchronous execution on the internal executor pool, then arranges for + * {@code onComplete} to be invoked with the result (or after {@code waitTime} elapses). + * + * @param job the job to track and execute + * @param onComplete callback to invoke when the job completes + * @param serviceExecutorPool the executor pool to use for waiting on job completion + * @param waitTime the maximum time to wait for job completion before returning + */ + private void trackAndExecute(OperationalJob job, + BiConsumer onComplete, + TaskExecutorPool serviceExecutorPool, + DurationSpec waitTime) + { + // New job is submitted for all cases when we do not have a corresponding downstream job + OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job); + if (tracked == job) + { + internalExecutorPool.executeBlocking(job::execute); + } + // Get the result, waiting for the specified wait time for result job.asyncResult(serviceExecutorPool, waitTime) .onComplete(v -> onComplete.accept(job, null)); @@ -132,4 +185,51 @@ private void checkConflict(OperationalJob job) throws OperationalJobConflictExce throw new OperationalJobConflictException("The same operational job is already running on Cassandra. operationName='" + job.name() + '\''); } } + + /** + * For jobs that require cluster-wide coordination, attempts to acquire the active operation lock via the + * coordinator. The acquisition runs on the internal executor pool because it might performs blocking storage I/O + * + * @param job the job requiring coordination + * @return a future resolving to {@code true} if the lock was acquired, {@code false} if another operation + * already holds it, or a failed future if coordination could not be attempted (e.g. coordination is + * disabled on this instance or the storage call failed) + */ + private Future acquireActiveOperationLock(OperationalJob job) + { + return internalExecutorPool.executeBlocking(() -> coordinator.trySetActive(job.operationType(), job.jobId()), false); + } + + /** + * Releases the active operation lock previously acquired for the given job. The release performs + * blocking storage I/O, so it runs on the internal executor pool off the event loop. A failure to clear is logged + * rather than surfaced, since the job has already completed by this point. + * + * @param job the job whose active operation lock should be released + */ + private void releaseActiveOperationLock(OperationalJob job) + { + internalExecutorPool.executeBlocking(() -> coordinator.clearActive(job.operationType(), job.jobId()), false) + .onFailure(e -> logger.error("Failed to clear active operation lock. jobId={} operationType={}", + job.jobId(), job.operationType(), e)); + } + + /** + * Builds the conflict exception describing why the active operation lock could not be acquired. + * + * @param job the job that could not be coordinated + * @param cause the failure cause when coordination could not be attempted, or {@code null} when the lock is + * simply held by another active operation + * @return the conflict exception to report to the caller + */ + private OperationalJobConflictException coordinationConflict(OperationalJob job, @Nullable Throwable cause) + { + if (cause != null) + { + return new OperationalJobConflictException("Unable to coordinate operation. operationType='" + + job.operationType() + "', reason='" + cause.getMessage() + '\''); + } + return new OperationalJobConflictException("An active operation already exists. operationType='" + + job.operationType() + '\''); + } } diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/StorageBackedOperationalJobCoordinator.java b/server/src/main/java/org/apache/cassandra/sidecar/job/StorageBackedOperationalJobCoordinator.java new file mode 100644 index 000000000..152ba9b61 --- /dev/null +++ b/server/src/main/java/org/apache/cassandra/sidecar/job/StorageBackedOperationalJobCoordinator.java @@ -0,0 +1,71 @@ +/* + * 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.Map; +import java.util.UUID; + +import com.google.inject.Inject; +import com.google.inject.Singleton; +import org.apache.cassandra.sidecar.common.data.OperationType; +import org.apache.cassandra.sidecar.job.storage.StorageProvider; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * An {@link OperationalJobCoordinator} implementation that delegates to a {@link StorageProvider} + * for coordination of active operations. + */ +@Singleton +public class StorageBackedOperationalJobCoordinator implements OperationalJobCoordinator +{ + private final StorageProvider storageProvider; + + @Inject + public StorageBackedOperationalJobCoordinator(StorageProvider storageProvider) + { + this.storageProvider = storageProvider; + } + + @Override + public boolean trySetActive(OperationType operationType, UUID operationId) + { + return storageProvider.trySetActiveOperation(operationType, operationId); + } + + @Override + public boolean clearActive(OperationType operationType, UUID operationId) + { + return storageProvider.clearActiveOperation(operationType, operationId); + } + + @Override + @Nullable + public UUID getActiveOperation(OperationType operationType) + { + return storageProvider.getActiveOperation(operationType); + } + + @Override + @NotNull + public Map getActiveOperations() + { + return storageProvider.getActiveOperations(); + } +} diff --git a/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java b/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java index a2224e5ca..49c1af468 100644 --- a/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java +++ b/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java @@ -61,7 +61,9 @@ import org.apache.cassandra.sidecar.handlers.cassandra.NodeSettingsHandler; import org.apache.cassandra.sidecar.handlers.v2.cassandra.V2NodeSettingsHandler; import org.apache.cassandra.sidecar.handlers.validations.ValidateTableExistenceHandler; +import org.apache.cassandra.sidecar.job.DisabledOperationalJobCoordinator; import org.apache.cassandra.sidecar.job.InMemoryOperationalJobTracker; +import org.apache.cassandra.sidecar.job.OperationalJobCoordinator; import org.apache.cassandra.sidecar.job.OperationalJobTracker; import org.apache.cassandra.sidecar.modules.multibindings.KeyClassMapKey; import org.apache.cassandra.sidecar.modules.multibindings.TableSchemaMapKeys; @@ -83,6 +85,7 @@ public class CassandraOperationsModule extends AbstractModule protected void configure() { bind(OperationalJobTracker.class).to(InMemoryOperationalJobTracker.class); + bind(OperationalJobCoordinator.class).to(DisabledOperationalJobCoordinator.class); } @ProvidesIntoMap diff --git a/server/src/test/java/org/apache/cassandra/sidecar/job/DisabledOperationalJobCoordinatorTest.java b/server/src/test/java/org/apache/cassandra/sidecar/job/DisabledOperationalJobCoordinatorTest.java new file mode 100644 index 000000000..efebc6cf3 --- /dev/null +++ b/server/src/test/java/org/apache/cassandra/sidecar/job/DisabledOperationalJobCoordinatorTest.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.sidecar.job; + +import java.util.UUID; + +import org.junit.jupiter.api.Test; + +import com.datastax.driver.core.utils.UUIDs; +import org.apache.cassandra.sidecar.common.data.OperationType; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for {@link DisabledOperationalJobCoordinator}, the coordinator used when a Sidecar instance + * does not support coordinated cluster-wide operations. + */ +class DisabledOperationalJobCoordinatorTest +{ + private final OperationalJobCoordinator coordinator = new DisabledOperationalJobCoordinator(); + + @Test + void testTrySetActiveThrows() + { + UUID operationId = UUIDs.timeBased(); + assertThatThrownBy(() -> coordinator.trySetActive(OperationType.MOVE, operationId)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("coordination is not supported by this Sidecar instance"); + } + + @Test + void testClearActiveThrows() + { + UUID operationId = UUIDs.timeBased(); + assertThatThrownBy(() -> coordinator.clearActive(OperationType.MOVE, operationId)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("coordination is not supported by this Sidecar instance"); + } + + @Test + void testGetActiveOperationReturnsNull() + { + assertThat(coordinator.getActiveOperation(OperationType.MOVE)).isNull(); + } + + @Test + void testGetActiveOperationsReturnsEmptyMap() + { + assertThat(coordinator.getActiveOperations()).isEmpty(); + } +} 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 cf7492c2c..797743927 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 @@ -41,14 +41,20 @@ import org.apache.cassandra.sidecar.exceptions.OperationalJobConflictException; import org.apache.cassandra.sidecar.job.storage.StorageProvider; import org.apache.cassandra.sidecar.job.storage.StorageProviderException; +import org.jetbrains.annotations.NotNull; +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.testing.utils.AssertionUtils.loopAssert; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.after; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** @@ -82,7 +88,7 @@ void cleanup() void testWithNoDownstreamJob() throws InterruptedException { OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); - OperationalJobManager manager = new OperationalJobManager(tracker, executorPool); + OperationalJobManager manager = new OperationalJobManager(tracker, new DisabledOperationalJobCoordinator(), executorPool); CountDownLatch latch = new CountDownLatch(1); OperationalJob testJob = OperationalJobTest.createOperationalJob(SUCCEEDED); @@ -103,7 +109,7 @@ void testWithRunningDownstreamJob() throws InterruptedException { OperationalJob runningJob = OperationalJobTest.createOperationalJob(RUNNING); OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); - OperationalJobManager manager = new OperationalJobManager(tracker, executorPool); + OperationalJobManager manager = new OperationalJobManager(tracker, new DisabledOperationalJobCoordinator(), executorPool); CountDownLatch latch = new CountDownLatch(1); BiConsumer onComplete = (job, exception) -> { @@ -121,7 +127,7 @@ void testWithLongRunningJob() throws InterruptedException { UUID jobId = UUIDs.timeBased(); OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); - OperationalJobManager manager = new OperationalJobManager(tracker, executorPool); + OperationalJobManager manager = new OperationalJobManager(tracker, new DisabledOperationalJobCoordinator(), executorPool); CountDownLatch latch = new CountDownLatch(1); OperationalJob testJob = OperationalJobTest.createOperationalJob(jobId, SecondBoundConfiguration.parse("2s")); @@ -145,7 +151,7 @@ void testWithFailingJob() throws InterruptedException { UUID jobId = UUIDs.timeBased(); OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); - OperationalJobManager manager = new OperationalJobManager(tracker, executorPool); + OperationalJobManager manager = new OperationalJobManager(tracker, new DisabledOperationalJobCoordinator(), executorPool); CountDownLatch latch = new CountDownLatch(1); String msg = "Test Job failed"; @@ -193,7 +199,7 @@ void testJobRemovedFromTrackerWhenPersistenceFails() DurableOperationalJobTracker durableTracker = new DurableOperationalJobTracker(new ServiceConfigurationImpl(), storageProvider, executorPool.service()); - OperationalJobManager manager = new OperationalJobManager(durableTracker, executorPool); + OperationalJobManager manager = new OperationalJobManager(durableTracker, new DisabledOperationalJobCoordinator(), executorPool); UUID jobId = UUIDs.timeBased(); OperationalJob job = OperationalJobTest.createOperationalJob(jobId, MillisecondBoundConfiguration.parse("50ms")); @@ -207,4 +213,182 @@ void testJobRemovedFromTrackerWhenPersistenceFails() assertThat(durableTracker.jobsView()).doesNotContainKey(jobId); }); } + + void testCoordinatorCalledWhenJobRequiresCoordination() throws InterruptedException + { + OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); + OperationalJobCoordinator coordinator = mock(OperationalJobCoordinator.class); + when(coordinator.trySetActive(any(), any())).thenReturn(true); + OperationalJobManager manager = new OperationalJobManager(tracker, coordinator, executorPool); + CountDownLatch latch = new CountDownLatch(1); + + OperationalJob job = createCoordinatedJob(UUIDs.timeBased()); + BiConsumer onComplete = (j, ex) -> { + assertThat(ex).isNull(); + latch.countDown(); + }; + + manager.trySubmitJob(job, onComplete, executorPool.service(), SecondBoundConfiguration.parse("5s")); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + verify(coordinator).trySetActive(OperationType.MOVE, job.jobId()); + verify(coordinator, timeout(5000)).clearActive(OperationType.MOVE, job.jobId()); + } + + @Test + void testConflictWhenCoordinatorReturnsFalse() throws InterruptedException + { + OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); + OperationalJobCoordinator coordinator = mock(OperationalJobCoordinator.class); + when(coordinator.trySetActive(any(), any())).thenReturn(false); + OperationalJobManager manager = new OperationalJobManager(tracker, coordinator, executorPool); + CountDownLatch latch = new CountDownLatch(1); + + OperationalJob job = createCoordinatedJob(UUIDs.timeBased()); + BiConsumer onComplete = (j, ex) -> { + assertThat(ex).isInstanceOf(OperationalJobConflictException.class); + assertThat(ex.getMessage()).contains("An active operation already exists"); + latch.countDown(); + }; + + manager.trySubmitJob(job, onComplete, executorPool.service(), SecondBoundConfiguration.parse("5s")); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + OperationalJobInfo tracked = tracker.get(job.jobId()); + assertThat(tracked).isNotNull(); + assertThat(tracked.status()).isEqualTo(FAILED); + assertThat(tracked.failureReason()).contains("An active operation already exists"); + assertThat(tracker.inflightJobsByOperation(job.name())).doesNotContain(job); + verify(coordinator, never()).clearActive(any(), any()); + } + + @Test + void testCoordinationFailsWhenCoordinationDisabled() throws InterruptedException + { + OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); + // Coordination is disabled on this instance, yet the job requires coordination. + OperationalJobManager manager = new OperationalJobManager(tracker, new DisabledOperationalJobCoordinator(), executorPool); + CountDownLatch latch = new CountDownLatch(1); + + OperationalJob job = createCoordinatedJob(UUIDs.timeBased()); + BiConsumer onComplete = (j, ex) -> { + assertThat(ex).isInstanceOf(OperationalJobConflictException.class); + assertThat(ex.getMessage()).contains("coordination is not supported by this Sidecar instance"); + latch.countDown(); + }; + + manager.trySubmitJob(job, onComplete, executorPool.service(), SecondBoundConfiguration.parse("5s")); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + OperationalJobInfo tracked = tracker.get(job.jobId()); + assertThat(tracked).isNotNull(); + assertThat(tracked.status()).isEqualTo(FAILED); + assertThat(tracked.failureReason()).contains("coordination is not supported by this Sidecar instance"); + assertThat(tracker.inflightJobsByOperation(job.name())).doesNotContain(job); + } + + @Test + void testLockNotReleasedWhenJobOptsOut() throws InterruptedException + { + OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); + OperationalJobCoordinator coordinator = mock(OperationalJobCoordinator.class); + when(coordinator.trySetActive(any(), any())).thenReturn(true); + OperationalJobManager manager = new OperationalJobManager(tracker, coordinator, executorPool); + CountDownLatch latch = new CountDownLatch(1); + + // A distributed cluster-wide job that acquires the lock locally but relies on the orchestration + // layer to clear it once all nodes finish, so the manager must not auto-release on local completion. + OperationalJob job = createNonReleasingCoordinatedJob(UUIDs.timeBased()); + BiConsumer onComplete = (j, ex) -> { + assertThat(ex).isNull(); + latch.countDown(); + }; + + manager.trySubmitJob(job, onComplete, executorPool.service(), SecondBoundConfiguration.parse("5s")); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + verify(coordinator).trySetActive(OperationType.MOVE, job.jobId()); + verify(coordinator, after(1000).never()).clearActive(any(), any()); + } + + @Test + void testCoordinatorNotCalledWhenJobDoesNotRequireCoordination() throws InterruptedException + { + OperationalJobTracker tracker = new InMemoryOperationalJobTracker(4); + OperationalJobCoordinator coordinator = mock(OperationalJobCoordinator.class); + OperationalJobManager manager = new OperationalJobManager(tracker, coordinator, executorPool); + CountDownLatch latch = new CountDownLatch(1); + + OperationalJob job = OperationalJobTest.createOperationalJob(SUCCEEDED); + BiConsumer onComplete = (j, ex) -> { + assertThat(ex).isNull(); + latch.countDown(); + }; + + manager.trySubmitJob(job, onComplete, executorPool.service(), SecondBoundConfiguration.parse("5s")); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + verify(coordinator, never()).trySetActive(any(), any()); + } + + private static OperationalJob createCoordinatedJob(UUID jobId) + { + return new OperationalJob(jobId) + { + @Override + public boolean hasConflict(@NotNull List sameOperationJobs) + { + return false; + } + + @Override + public OperationType operationType() + { + return OperationType.MOVE; + } + + @Override + public boolean requiresCoordination() + { + return true; + } + + @Override + protected Future executeInternal() + { + return Future.succeededFuture(); + } + }; + } + + private static OperationalJob createNonReleasingCoordinatedJob(UUID jobId) + { + return new OperationalJob(jobId) + { + @Override + public boolean hasConflict(@NotNull List sameOperationJobs) + { + return false; + } + + @Override + public OperationType operationType() + { + return OperationType.MOVE; + } + + @Override + public boolean requiresCoordination() + { + return true; + } + + @Override + public boolean releasesOnCompletion() + { + return false; + } + + @Override + protected Future executeInternal() + { + return Future.succeededFuture(); + } + }; + } } diff --git a/server/src/test/java/org/apache/cassandra/sidecar/job/RepairJobTest.java b/server/src/test/java/org/apache/cassandra/sidecar/job/RepairJobTest.java index 1eb46601a..34f772aff 100644 --- a/server/src/test/java/org/apache/cassandra/sidecar/job/RepairJobTest.java +++ b/server/src/test/java/org/apache/cassandra/sidecar/job/RepairJobTest.java @@ -234,7 +234,7 @@ void testMultipleRepairJobsRunningInParallel() throws Exception { // Create a job tracker and manager OperationalJobTracker tracker = new InMemoryOperationalJobTracker(10); - OperationalJobManager manager = new OperationalJobManager(tracker, executorPool); + OperationalJobManager manager = new OperationalJobManager(tracker, new DisabledOperationalJobCoordinator(), executorPool); // Mock the storage operations StorageOperations storageOperations = mock(StorageOperations.class);