Implement releaseActiveOperationLock

Author: andresbeckruiz

server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java

@@ -136,6 +136,7 @@ public void trySubmitJob(OperationalJob job,
         {
             if (ar.succeeded() && Boolean.TRUE.equals(ar.result()))
             {
+                job.asyncResult().onComplete(result -> releaseActiveOperationLock(job));
                 trackAndExecute(job, onComplete, serviceExecutorPool, waitTime);
             }
             else
@@ -207,6 +208,24 @@ 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)
+    {
+        if (coordinator == null)
+        {
+            return;
+        }
+        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.
      *

server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java

@@ -47,6 +47,7 @@
 import static org.mockito.ArgumentMatchers.any;
 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;
 
@@ -199,6 +200,7 @@ void testCoordinatorCalledWhenJobRequiresCoordination() throws InterruptedExcept
         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
@@ -224,6 +226,7 @@ void testConflictWhenCoordinatorReturnsFalse() throws InterruptedException
         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