You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "andresbeckruiz (via GitHub)" <gi...@apache.org> on 2026/06/10 20:50:55 UTC
[PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
andresbeckruiz opened a new pull request, #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360
## [CASSSIDECAR-377](https://issues.apache.org/jira/browse/CASSSIDECAR-377)
Original PR made against [CASSSIDECAR-373](https://github.com/apache/cassandra-sidecar/pull/339) branch with review comments: https://github.com/andresbeckruiz/cassandra-sidecar/pull/2.
### Changes
- `OperationalJobCoordinator` interface
- `StorageOperationalJobCoordinator`: Implementation that delegates to `StorageProvider`'s compare and set based methods to acquire a lock for an active operation
- `OperationalJob.operationType()`: New abstract method returning `OperationType` enum, implemented by all concrete jobs
- `OperationalJobManager` integration: If a job requires coordination, the manager acquires the active lock via the coordinator before execution; throws `OperationalJobConflictException` if rejected
- Unit tests for `StorageOperationalJobCoordinator` and coordinator integration in `OperationalJobManagerTest`
This ticket only covers the job activation path when a job is submitted to be executed immediately. Clearing locks after completion and status update activation will be implemented in later tickets (CASSSIDECAR-378, CASSSIDECAR-379).
### Future Work
- [CASSSIDECAR-378](https://issues.apache.org/jira/browse/CASSSIDECAR-378): Add support for creation and coordination of local Sidecar Jobs
- [CASSSIDECAR-379](https://issues.apache.org/jira/browse/CASSSIDECAR-379): Add PATCH operational-jobs/{id} API to update status of existing operational jobs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538862656
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -101,10 +118,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 -> {
+ OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
Review Comment:
Coordination now happens before tracking and jobs that fail coordination are now marked as `FAILED`, addressed in [6ba25aa](https://github.com/apache/cassandra-sidecar/pull/360/commits/6ba25aa1fb1bc76c8d1809414681f5ad107a5318)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#issuecomment-5209005033
Rebased to prepare for merge ^
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "isaacreath (via GitHub)" <gi...@apache.org>.
isaacreath commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3544181050
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -41,18 +43,33 @@ public class OperationalJobManager
{
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
private final OperationalJobTracker jobTracker;
-
+ @Nullable
+ private final OperationalJobCoordinator coordinator;
Review Comment:
The nullable `OperationalJobCoordinator` feels like a bit of a smell to me. It seems like the goal here is to support backwards compatibility for the `OperationalJobManager` for users of the `OperationalJobManager` to not have to include coordination (e.g., for existing decommission jobs), am I right about that?
If so, I'd suggest that instead of having a null `OperationalJobCoordinator`, we have a second implementation, `DisabledOperationalJobCoordinator` or something similar, that fails all coordination tasks with an error saying that coordination is not currently supported by this sidecar instance (and maybe how to configure coordination if the user wants to have coordinated jobs).
This way we make the coordination enabled / disabled logic part of the behavior of the coordinator itself, as opposed to having a null dependency here. It can also clean up all accesses to the `OperationalJobCoordinator` as we no longer have to worry about it being null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3436977162
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
+ "Job requires coordination but no OperationalJobCoordinator is configured");
+ boolean activated = coordinator.trySetActive(job.operationType(), job.jobId());
Review Comment:
I am afraid to merge this PR without calling clearActive. If we delay adding that in a subsequent PR and if someone enables requiresCoordination() for any job meanwhile, then no more further operations will be allowed. I would recommend implementing calling clearActive in the same PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3436915257
##########
server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java:
##########
@@ -166,4 +179,93 @@ protected Future<Void> executeInternal() throws OperationalJobException
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(tracker.get(jobId)).isNotNull();
}
+
+ @Test
+ 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());
+ }
+
+ @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();
Review Comment:
Also verify that tracker doesn't have entry for this job after conflict detected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3539128353
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
+ "Job requires coordination but no OperationalJobCoordinator is configured");
+ boolean activated = coordinator.trySetActive(job.operationType(), job.jobId());
Review Comment:
Added in [ff1b8f2](https://github.com/apache/cassandra-sidecar/pull/360/commits/ff1b8f227a5102bb18ee7176cac82ce8f8cd0021)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538859540
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
Review Comment:
No more unhandled `IllegalStateException`, addressed in [3661c4a](https://github.com/apache/cassandra-sidecar/pull/360/commits/3661c4aaaedac3205088f1aa2d60d909357ae234)
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -101,10 +118,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 -> {
+ OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
Review Comment:
Coordination now happens before tracking and jobs that fail coordination are now marked as `FAILED`, addressed in [3661c4a](https://github.com/apache/cassandra-sidecar/pull/360/commits/3661c4aaaedac3205088f1aa2d60d909357ae234)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3539128353
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
+ "Job requires coordination but no OperationalJobCoordinator is configured");
+ boolean activated = coordinator.trySetActive(job.operationType(), job.jobId());
Review Comment:
Added in [097a9af](https://github.com/apache/cassandra-sidecar/pull/360/commits/097a9af44a5b85b2c4de5f945492176310878904)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3553152485
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -41,18 +43,33 @@ public class OperationalJobManager
{
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
private final OperationalJobTracker jobTracker;
-
+ @Nullable
+ private final OperationalJobCoordinator coordinator;
Review Comment:
Good idea. Replaced the `@Nullable OperationalJobCoordinator` with a `DisabledOperationalJobCoordinator` in [9868be7](https://github.com/apache/cassandra-sidecar/pull/360/commits/9868be7b9f4cc1b8348674b7bedbbf1d269f94f0)
To clarify, backwards compatibility for existing uncoordinated jobs (eg: decommission) is already handled by `requiresCoordination() == false`, so they short circuit before touching the coordinator. Therefore, the `DisabledOperationalJobCoordinator` only matters for a job requiring coordination submitted to an instance without coordination support.
Also updated this patch to bind the `OperationalJobCoordinator` to the `DisabledOperationalJobCoordinator`. Configuration to enable this class (as well as durable tracking) will be added in CASSSIDECAR-378.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3553152485
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -41,18 +43,33 @@ public class OperationalJobManager
{
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
private final OperationalJobTracker jobTracker;
-
+ @Nullable
+ private final OperationalJobCoordinator coordinator;
Review Comment:
Good idea. Replaced the `@Nullable OperationalJobCoordinator` with a `DisabledOperationalJobCoordinator` in [509b15b](https://github.com/apache/cassandra-sidecar/pull/360/commits/509b15b3d41bcc40c4ef46cd9c53439edc3654bd)
To clarify, backwards compatibility for existing uncoordinated jobs (eg: decommission) is already handled by `requiresCoordination() == false`, so they short circuit before touching the coordinator. Therefore, the `DisabledOperationalJobCoordinator` only matters for a job requiring coordination submitted to an instance without coordination support.
Also updated this patch to bind the `OperationalJobCoordinator` to the `DisabledOperationalJobCoordinator`. Configuration to enable this class (as well as durable tracking) will be added in CASSSIDECAR-378.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538859540
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
Review Comment:
No more unhandled `IllegalStateException`, addressed in [30a4292](https://github.com/apache/cassandra-sidecar/pull/360/commits/30a42928fa40df19d26a5e729d59f8b7c71fa6a0)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538859540
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
Review Comment:
No more unhandled `IllegalStateException`, addressed in [a2bd0e1](https://github.com/apache/cassandra-sidecar/pull/360/commits/a2bd0e15ce73f49a2223212df48a3daa6b2bd436)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "isaacreath (via GitHub)" <gi...@apache.org>.
isaacreath commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3437163382
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -101,10 +118,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 -> {
+ OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
Review Comment:
Would it make sense to mark the job as `FAILED` if coordination fails? And set the `failureReason` to indicate that the job was not able to start due to coordination failure?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538950842
##########
server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java:
##########
@@ -166,4 +179,93 @@ protected Future<Void> executeInternal() throws OperationalJobException
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(tracker.get(jobId)).isNotNull();
}
+
+ @Test
+ 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());
+ }
+
+ @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();
Review Comment:
Jobs are marked as `FAILED` if coordination fails, so no conflict will happen as `inflightJobsByOperation` does not return completed jobs. Added
`assertThat(tracker.inflightJobsByOperation(job.name())).doesNotContain(job);
`
in [30a4292](https://github.com/apache/cassandra-sidecar/pull/360/commits/30a42928fa40df19d26a5e729d59f8b7c71fa6a0) to confirm
##########
server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java:
##########
@@ -166,4 +179,93 @@ protected Future<Void> executeInternal() throws OperationalJobException
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(tracker.get(jobId)).isNotNull();
}
+
+ @Test
+ 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());
+ }
+
+ @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();
Review Comment:
Jobs are marked as `FAILED` if coordination fails, so no conflict will happen as `inflightJobsByOperation` does not return completed jobs. Added
`assertThat(tracker.inflightJobsByOperation(job.name())).doesNotContain(job);
`
in [30a4292](https://github.com/apache/cassandra-sidecar/pull/360/commits/30a42928fa40df19d26a5e729d59f8b7c71fa6a0) to confirm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538859540
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
Review Comment:
No more unhandled `IllegalStateException`, addressed in [6ba25aa](https://github.com/apache/cassandra-sidecar/pull/360/commits/6ba25aa1fb1bc76c8d1809414681f5ad107a5318)
##########
server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java:
##########
@@ -166,4 +179,93 @@ protected Future<Void> executeInternal() throws OperationalJobException
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(tracker.get(jobId)).isNotNull();
}
+
+ @Test
+ 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());
+ }
+
+ @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();
Review Comment:
Jobs are now marked as `FAILED` if coordination fails, so no conflict will happen as `inflightJobsByOperation` does not return completed jobs. Added
`assertThat(tracker.inflightJobsByOperation(job.name())).doesNotContain(job);
`
in [6ba25aa](https://github.com/apache/cassandra-sidecar/pull/360/commits/6ba25aa1fb1bc76c8d1809414681f5ad107a5318) to confirm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3436874511
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -101,10 +118,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 -> {
+ OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
Review Comment:
This leaves stale job entry in CREATED state if the coordination fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538950842
##########
server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java:
##########
@@ -166,4 +179,93 @@ protected Future<Void> executeInternal() throws OperationalJobException
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(tracker.get(jobId)).isNotNull();
}
+
+ @Test
+ 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());
+ }
+
+ @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();
Review Comment:
Jobs are marked as `FAILED` if coordination fails, so no conflict will happen as `inflightJobsByOperation` does not return completed jobs. Added
`assertThat(tracker.inflightJobsByOperation(job.name())).doesNotContain(job);`
in [30a4292](https://github.com/apache/cassandra-sidecar/pull/360/commits/30a42928fa40df19d26a5e729d59f8b7c71fa6a0) to confirm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3539128353
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
+ "Job requires coordination but no OperationalJobCoordinator is configured");
+ boolean activated = coordinator.trySetActive(job.operationType(), job.jobId());
Review Comment:
Added in [954454d](https://github.com/apache/cassandra-sidecar/pull/360/commits/954454d691b30637f260657f327dca522c97107b)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3555015640
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -99,19 +107,60 @@ public void trySubmitJob(OperationalJob job,
try
{
checkConflict(job);
-
- // New job is submitted for all cases when we do not have a corresponding downstream job
- jobTracker.computeIfAbsent(job.jobId(), jobId -> {
- internalExecutorPool.executeBlocking(job::execute);
- return job;
- });
}
catch (OperationalJobConflictException oje)
{
onComplete.accept(job, oje);
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()))
+ {
+ job.asyncResult().onComplete(result -> releaseActiveOperationLock(job));
Review Comment:
Good catch-- addressed in [5d5afc6](https://github.com/apache/cassandra-sidecar/pull/360/commits/5d5afc6f40244c8937050fcfa14f8978cb8aff3d)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3436894747
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -131,4 +150,26 @@ 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. Throws a conflict if another operation is already active.
+ *
+ * @param job instance of the job to coordinate
+ * @throws OperationalJobConflictException when the coordinator cannot activate the operation
+ */
+ private void tryCoordination(OperationalJob job) throws OperationalJobConflictException
+ {
+ if (job.requiresCoordination())
+ {
+ Preconditions.checkState(coordinator != null,
Review Comment:
Preconditions throws illegal state exception, which is not handled in the caller
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538862656
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -101,10 +118,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 -> {
+ OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
Review Comment:
Coordination now happens before tracking and jobs that fail coordination are now marked as `FAILED`, addressed in [30a4292](https://github.com/apache/cassandra-sidecar/pull/360/commits/30a42928fa40df19d26a5e729d59f8b7c71fa6a0)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#issuecomment-4908515645
CI: https://app.circleci.com/pipelines/github/andresbeckruiz/cassandra-sidecar/101/workflows/d27734e4-047e-4299-9029-ed0c9624184c
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3555015640
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -99,19 +107,60 @@ public void trySubmitJob(OperationalJob job,
try
{
checkConflict(job);
-
- // New job is submitted for all cases when we do not have a corresponding downstream job
- jobTracker.computeIfAbsent(job.jobId(), jobId -> {
- internalExecutorPool.executeBlocking(job::execute);
- return job;
- });
}
catch (OperationalJobConflictException oje)
{
onComplete.accept(job, oje);
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()))
+ {
+ job.asyncResult().onComplete(result -> releaseActiveOperationLock(job));
Review Comment:
Good catch-- addressed in [58bb7d4](https://github.com/apache/cassandra-sidecar/pull/360/commits/58bb7d45428d94dc3ea9b1c6c4aa0fd96a7bc9cf)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "pauloricardomg (via GitHub)" <gi...@apache.org>.
pauloricardomg commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3406169010
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -101,10 +118,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 -> {
+ OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
Review Comment:
do we need to remove from jobTracker if there's a coordination conflict ? Or perhaps should we just add to jobTracker in case the tryCoordination was successful, so tryCoordination should come before jobtracker.computeIfAbsent ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3553152485
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -41,18 +43,33 @@ public class OperationalJobManager
{
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
private final OperationalJobTracker jobTracker;
-
+ @Nullable
+ private final OperationalJobCoordinator coordinator;
Review Comment:
Good idea. Replaced the `@Nullable OperationalJobCoordinator` with a `DisabledOperationalJobCoordinator` in [405a93d](https://github.com/apache/cassandra-sidecar/pull/360/commits/405a93dad1e653cf6bf53cdc31eccfa6fb3b19e3)
To clarify, backwards compatibility for existing uncoordinated jobs (eg: decommission) is already handled by `requiresCoordination() == false`, so they short circuit before touching the coordinator. Therefore, the `DisabledOperationalJobCoordinator` only matters for a job requiring coordination submitted to an instance without coordination support.
Also updated this patch to bind the `OperationalJobCoordinator` to the `DisabledOperationalJobCoordinator`. Configuration to enable this class (as well as durable tracking) will be added in CASSSIDECAR-378.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3554488148
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -99,19 +107,60 @@ public void trySubmitJob(OperationalJob job,
try
{
checkConflict(job);
-
- // New job is submitted for all cases when we do not have a corresponding downstream job
- jobTracker.computeIfAbsent(job.jobId(), jobId -> {
- internalExecutorPool.executeBlocking(job::execute);
- return job;
- });
}
catch (OperationalJobConflictException oje)
{
onComplete.accept(job, oje);
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()))
+ {
+ job.asyncResult().onComplete(result -> releaseActiveOperationLock(job));
Review Comment:
`requiresCoordination()` is a clean job-level flag the manager just reads.
The release side isn't symmetric though, as `trySubmitJob` always auto-releases on local completion, with no way for a job to opt out. That's right for today's single-node jobs, but doesn't match clearActive's javadoc (orchestration-layer clearing for jobs that finish elsewhere).
Rather than just documenting that gap, could we add a parallel hook for release?
e.g. releasesOnCompletion(), default true; the manager just reads it instead of special-casing job types, so a future distributed job can skip the automatic release without any change to `OperationalJobManager`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3555015640
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -99,19 +107,60 @@ public void trySubmitJob(OperationalJob job,
try
{
checkConflict(job);
-
- // New job is submitted for all cases when we do not have a corresponding downstream job
- jobTracker.computeIfAbsent(job.jobId(), jobId -> {
- internalExecutorPool.executeBlocking(job::execute);
- return job;
- });
}
catch (OperationalJobConflictException oje)
{
onComplete.accept(job, oje);
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()))
+ {
+ job.asyncResult().onComplete(result -> releaseActiveOperationLock(job));
Review Comment:
Good catch-- addressed in [1bd9057](https://github.com/apache/cassandra-sidecar/pull/360/commits/1bd9057dbcf307c3c81d3fae046ec5f5b0ba39f1)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538950842
##########
server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java:
##########
@@ -166,4 +179,93 @@ protected Future<Void> executeInternal() throws OperationalJobException
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(tracker.get(jobId)).isNotNull();
}
+
+ @Test
+ 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());
+ }
+
+ @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();
Review Comment:
Jobs are now marked as `FAILED` if coordination fails, so no conflict will happen as `inflightJobsByOperation` does not return completed jobs. Added
`assertThat(tracker.inflightJobsByOperation(job.name())).doesNotContain(job);
`
in [30a4292](https://github.com/apache/cassandra-sidecar/pull/360/commits/30a42928fa40df19d26a5e729d59f8b7c71fa6a0) to confirm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538862656
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java:
##########
@@ -101,10 +118,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 -> {
+ OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
Review Comment:
Coordination now happens before tracking and jobs that fail coordination are now marked as `FAILED`, addressed in [a2bd0e1](https://github.com/apache/cassandra-sidecar/pull/360/commits/a2bd0e15ce73f49a2223212df48a3daa6b2bd436)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "andresbeckruiz (via GitHub)" <gi...@apache.org>.
andresbeckruiz commented on code in PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360#discussion_r3538950842
##########
server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java:
##########
@@ -166,4 +179,93 @@ protected Future<Void> executeInternal() throws OperationalJobException
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(tracker.get(jobId)).isNotNull();
}
+
+ @Test
+ 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());
+ }
+
+ @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();
Review Comment:
Jobs are now marked as `FAILED` if coordination fails, so no conflict will happen as `inflightJobsByOperation` does not return completed jobs. Added
`assertThat(tracker.inflightJobsByOperation(job.name())).doesNotContain(job);
`
in [3661c4a](https://github.com/apache/cassandra-sidecar/pull/360/commits/3661c4aaaedac3205088f1aa2d60d909357ae234) to confirm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-377: Implement job coordination for cluster-wide operations [cassandra-sidecar]
Posted by "pauloricardomg (via GitHub)" <gi...@apache.org>.
pauloricardomg merged PR #360:
URL: https://github.com/apache/cassandra-sidecar/pull/360
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org