This is an automated email from the ASF dual-hosted git repository.
pauloricardomg pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
The following commit(s) were added to refs/heads/trunk by this push:
new 3c88f576 CASSSIDECAR-377: Implement job coordination for cluster-wide
operations (#360)
3c88f576 is described below
commit 3c88f57643990333097d313c984b90015bd84394
Author: Andrés Beck-Ruiz <[email protected]>
AuthorDate: Thu Aug 6 16:46:32 2026 -0400
CASSSIDECAR-377: Implement job coordination for cluster-wide operations
(#360)
Patch by Andres Beck-Ruiz; Reviewed by Arjun Ashok, Francisco Guerrero,
Paulo Motta, Shailaja Koppu for CASSSIDECAR-377
---
CHANGES.txt | 1 +
.../job/DisabledOperationalJobCoordinator.java | 68 ++++++++
.../cassandra/sidecar/job/OperationalJob.java | 31 ++++
.../sidecar/job/OperationalJobCoordinator.java | 79 +++++++++
.../sidecar/job/OperationalJobManager.java | 122 +++++++++++--
.../StorageBackedOperationalJobCoordinator.java | 71 ++++++++
.../sidecar/modules/CassandraOperationsModule.java | 3 +
.../job/DisabledOperationalJobCoordinatorTest.java | 68 ++++++++
.../sidecar/job/OperationalJobManagerTest.java | 194 ++++++++++++++++++++-
.../cassandra/sidecar/job/RepairJobTest.java | 2 +-
10 files changed, 622 insertions(+), 17 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 71c6953d..6c43ac33 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Implement job coordination for cluster-wide operations (CASSSIDECAR-377)
* Fix dead/dropped CDC lifecycle metrics and duplicate SidecarCdcStats
interface (CASSSIDECAR-489)
* Wire CDC configs in configs table to
SidecarCdcOptions/SidecarStatePersister (CASSSIDECAR-483)
* Implement durable operational job tracker (CASSSIDECAR-374)
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 00000000..994e5f6e
--- /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.
+ * <p>
+ * 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<OperationType, UUID> 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 f3c15c6a..bdfa377c 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 abstract class OperationalJob implements
Task<Void>, OperationalJobInfo
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}.
+ * <p>
+ * 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 abstract class OperationalJob implements
Task<Void>, OperationalJobInfo
});
}
+ /**
+ * 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 00000000..90be1a3f
--- /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.
+ * <p>
+ * 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<OperationType, UUID> 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 a1e4d406..c348c86a 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 org.slf4j.LoggerFactory;
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 class OperationalJobManager
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 class OperationalJobManager
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<OperationalJob,
OperationalJobConflictException> 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 @@ public class OperationalJobManager
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<Boolean> 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 00000000..152ba9b6
--- /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<OperationType, UUID> 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 a2224e5c..49c1af46 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.TokenRangeReplicaMapHandler;
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 00000000..efebc6cf
--- /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 cf7492c2..79774392 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.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 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 @@ class OperationalJobManagerTest
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 @@ class OperationalJobManagerTest
{
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<OperationalJob, OperationalJobConflictException> onComplete
= (job, exception) -> {
@@ -121,7 +127,7 @@ class OperationalJobManagerTest
{
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 @@ class OperationalJobManagerTest
{
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 @@ class OperationalJobManagerTest
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 @@ class OperationalJobManagerTest
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<OperationalJob, OperationalJobConflictException> 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<OperationalJob, OperationalJobConflictException> 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<OperationalJob, OperationalJobConflictException> 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<OperationalJob, OperationalJobConflictException> 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<OperationalJob, OperationalJobConflictException> 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<OperationalJob>
sameOperationJobs)
+ {
+ return false;
+ }
+
+ @Override
+ public OperationType operationType()
+ {
+ return OperationType.MOVE;
+ }
+
+ @Override
+ public boolean requiresCoordination()
+ {
+ return true;
+ }
+
+ @Override
+ protected Future<Void> executeInternal()
+ {
+ return Future.succeededFuture();
+ }
+ };
+ }
+
+ private static OperationalJob createNonReleasingCoordinatedJob(UUID jobId)
+ {
+ return new OperationalJob(jobId)
+ {
+ @Override
+ public boolean hasConflict(@NotNull List<OperationalJob>
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<Void> 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 1eb46601..34f772af 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 @@ class RepairJobTest
{
// 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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
