pauloricardomg commented on code in PR #359:
URL: https://github.com/apache/cassandra-sidecar/pull/359#discussion_r3404691981
##########
server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java:
##########
@@ -173,6 +179,70 @@ public Map<String, String> operationMetadata()
return operationMetadata;
}
+ @Override
+ @Nullable
+ public UUID nodeId()
+ {
+ return null;
+ }
+
+ @Override
+ public String name()
+ {
+ return operationType.name();
+ }
+
+ @Override
+ public long creationTime()
+ {
+ return creationTimeMillis;
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesPending()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesExecuting()
+ {
+ return Collections.emptyList();
Review Comment:
Is there any reason why we're not populating these nodes* fields from the
`cluster_ops_node_state` table ?
##########
server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java:
##########
@@ -173,6 +179,70 @@ public Map<String, String> operationMetadata()
return operationMetadata;
}
+ @Override
+ @Nullable
+ public UUID nodeId()
+ {
+ return null;
Review Comment:
Why did we not include this field in the cluster ops schema?
##########
server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java:
##########
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.job;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.datastax.driver.core.utils.UUIDs;
+import io.vertx.core.Vertx;
+import org.apache.cassandra.sidecar.TestResourceReaper;
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import
org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
+import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.apache.cassandra.sidecar.job.storage.StorageProviderException;
+
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.CREATED;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.RUNNING;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.SUCCEEDED;
+import static
org.apache.cassandra.sidecar.job.OperationalJobTest.createOperationalJob;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link DurableOperationalJobTracker}
+ */
+class DurableOperationalJobTrackerTest
+{
+ private StorageProvider storageProvider;
+ private DurableOperationalJobTracker tracker;
+ private Vertx vertx;
+ private ExecutorPools executorPools;
+ private TaskExecutorPool executorPool;
+
+ @BeforeEach
+ void setUp()
+ {
+ storageProvider = mock(StorageProvider.class);
+ vertx = Vertx.vertx();
+ executorPools = new ExecutorPools(vertx, new
ServiceConfigurationImpl());
+ executorPool = executorPools.internal();
+ tracker = new DurableOperationalJobTracker(new
ServiceConfigurationImpl(), storageProvider, executorPool);
+ }
+
+ @AfterEach
+ void cleanup()
+ {
+ TestResourceReaper.create().with(vertx).with(executorPools).close();
+ }
+
+ @Test
+ void testComputeIfAbsentPersistsNewJob()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+
+ OperationalJob result = tracker.computeIfAbsent(job.jobId(), id ->
job);
+
+ assertThat(result).isSameAs(job);
+ verify(storageProvider).persistJob(argThat(record ->
+ record.jobId().equals(job.jobId()) &&
+ record.operationType() == job.operationType() &&
+ record.status() == CREATED
+ ));
+ }
+
+ @Test
+ void testComputeIfAbsentDoesNotPersistExistingJob()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+
+ tracker.computeIfAbsent(job.jobId(), id -> job);
+ OperationalJob secondCall = tracker.computeIfAbsent(job.jobId(), id ->
createOperationalJob(CREATED));
+
+ assertThat(secondCall).isSameAs(job);
+ verify(storageProvider, times(1)).persistJob(any());
+ }
+
+ @Test
+ void testComputeIfAbsentUpdatesStatusOnCompletion() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull());
+ });
+ }
+
+ @Test
+ void testComputeIfAbsentUpdatesStatusOnFailure() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
+ MillisecondBoundConfiguration.parse("50ms"),
+ new
org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException("test
failure"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(FAILED), eq("test failure"));
Review Comment:
this assertion is failing
##########
server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java:
##########
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.job;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.datastax.driver.core.utils.UUIDs;
+import io.vertx.core.Vertx;
+import org.apache.cassandra.sidecar.TestResourceReaper;
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import
org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
+import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.apache.cassandra.sidecar.job.storage.StorageProviderException;
+
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.CREATED;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.RUNNING;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.SUCCEEDED;
+import static
org.apache.cassandra.sidecar.job.OperationalJobTest.createOperationalJob;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link DurableOperationalJobTracker}
+ */
+class DurableOperationalJobTrackerTest
+{
+ private StorageProvider storageProvider;
+ private DurableOperationalJobTracker tracker;
+ private Vertx vertx;
+ private ExecutorPools executorPools;
+ private TaskExecutorPool executorPool;
+
+ @BeforeEach
+ void setUp()
+ {
+ storageProvider = mock(StorageProvider.class);
+ vertx = Vertx.vertx();
+ executorPools = new ExecutorPools(vertx, new
ServiceConfigurationImpl());
+ executorPool = executorPools.internal();
+ tracker = new DurableOperationalJobTracker(new
ServiceConfigurationImpl(), storageProvider, executorPool);
+ }
+
+ @AfterEach
+ void cleanup()
+ {
+ TestResourceReaper.create().with(vertx).with(executorPools).close();
+ }
+
+ @Test
+ void testComputeIfAbsentPersistsNewJob()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+
+ OperationalJob result = tracker.computeIfAbsent(job.jobId(), id ->
job);
+
+ assertThat(result).isSameAs(job);
+ verify(storageProvider).persistJob(argThat(record ->
+ record.jobId().equals(job.jobId()) &&
+ record.operationType() == job.operationType() &&
+ record.status() == CREATED
+ ));
+ }
+
+ @Test
+ void testComputeIfAbsentDoesNotPersistExistingJob()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+
+ tracker.computeIfAbsent(job.jobId(), id -> job);
+ OperationalJob secondCall = tracker.computeIfAbsent(job.jobId(), id ->
createOperationalJob(CREATED));
+
+ assertThat(secondCall).isSameAs(job);
+ verify(storageProvider, times(1)).persistJob(any());
+ }
+
+ @Test
+ void testComputeIfAbsentUpdatesStatusOnCompletion() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull());
Review Comment:
this assertion is failing
##########
server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java:
##########
@@ -173,6 +179,70 @@ public Map<String, String> operationMetadata()
return operationMetadata;
}
+ @Override
+ @Nullable
+ public UUID nodeId()
+ {
+ return null;
+ }
+
+ @Override
+ public String name()
+ {
+ return operationType.name();
+ }
+
+ @Override
+ public long creationTime()
+ {
+ return creationTimeMillis;
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesPending()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesExecuting()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesSucceeded()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ @NotNull
+ public List<UUID> nodesFailed()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isExecuting()
+ {
+ return status == OperationalJobStatus.RUNNING;
+ }
+
+ /**
+ * Creates an {@link OperationalJobRecord} from a live {@link
OperationalJob}.
+ *
+ * @param job the operational job to convert
+ * @return a new record capturing the job's current state
+ */
+ public static OperationalJobRecord fromOperationalJob(OperationalJob job)
Review Comment:
When a job completes, the DurableOperationalJobTracker evicts it from the
live map. After that, get(jobId) returns
an OperationalJobRecord from storage. Let's trace what data is available
at each stage:
```
While the job is live (e.g., NodeDecommissionJob):
nodeId() → UUID of the node being decommissioned
startTime() → when execution began
nodesPending() → [nodeId] initially, then [] after execution starts
nodesExecuting()→ [nodeId] while running
nodesSucceeded()→ [nodeId] after success
nodesFailed() → [nodeId] after failure
```
The factory that persists the job only captures 3 fields:
```java
// OperationalJobRecord.fromOperationalJob(job)
return new OperationalJobRecord(job.jobId(), job.operationType(),
job.status());
// ^^^^^^^^ ^^^^^^^^^^^^^^^
^^^^^^^^^^
// that's it — nodeId, startTime, node lists
are all discarded
```
After completion, get() returns the record with stubbed implementations:
```
nodeId() → null (always)
startTime() → null (constructor sets it to null)
nodesPending() → emptyList (always)
nodesExecuting() → emptyList (always)
nodesSucceeded() → emptyList (always)
nodesFailed() → emptyList (always)
```
So an API client observing a decommission job would see:
```
DURING execution: AFTER completion (from storage):
───────────────────────────────── ─────────────────────────────────
{ {
"jobId": "abc-123", "jobId": "abc-123",
"operation": "decommission", "operation": "DECOMMISSION",
"jobStatus": "RUNNING", "jobStatus": "SUCCEEDED",
"startTime": "2026-06-12T...", "startTime": null,
"nodesExecuting": ["node-uuid"], "nodesExecuting": [],
"nodesPending": [], "nodesPending": [],
"nodesSucceeded": [], "nodesSucceeded": [],
"nodesFailed": [] "nodesFailed": []
} }
```
All the provenance — which node was decommissioned, when it started,
whether it succeeded per-node — is lost once the job leaves the
local map. The record in Cassandra only has (jobId, operationType, status).
##########
server/src/test/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTrackerTest.java:
##########
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.job;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import com.datastax.driver.core.utils.UUIDs;
+import io.vertx.core.Vertx;
+import org.apache.cassandra.sidecar.TestResourceReaper;
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import
org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
+import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.apache.cassandra.sidecar.job.storage.StorageProviderException;
+
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.CREATED;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.RUNNING;
+import static
org.apache.cassandra.sidecar.common.data.OperationalJobStatus.SUCCEEDED;
+import static
org.apache.cassandra.sidecar.job.OperationalJobTest.createOperationalJob;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link DurableOperationalJobTracker}
+ */
+class DurableOperationalJobTrackerTest
+{
+ private StorageProvider storageProvider;
+ private DurableOperationalJobTracker tracker;
+ private Vertx vertx;
+ private ExecutorPools executorPools;
+ private TaskExecutorPool executorPool;
+
+ @BeforeEach
+ void setUp()
+ {
+ storageProvider = mock(StorageProvider.class);
+ vertx = Vertx.vertx();
+ executorPools = new ExecutorPools(vertx, new
ServiceConfigurationImpl());
+ executorPool = executorPools.internal();
+ tracker = new DurableOperationalJobTracker(new
ServiceConfigurationImpl(), storageProvider, executorPool);
+ }
+
+ @AfterEach
+ void cleanup()
+ {
+ TestResourceReaper.create().with(vertx).with(executorPools).close();
+ }
+
+ @Test
+ void testComputeIfAbsentPersistsNewJob()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+
+ OperationalJob result = tracker.computeIfAbsent(job.jobId(), id ->
job);
+
+ assertThat(result).isSameAs(job);
+ verify(storageProvider).persistJob(argThat(record ->
+ record.jobId().equals(job.jobId()) &&
+ record.operationType() == job.operationType() &&
+ record.status() == CREATED
+ ));
+ }
+
+ @Test
+ void testComputeIfAbsentDoesNotPersistExistingJob()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+
+ tracker.computeIfAbsent(job.jobId(), id -> job);
+ OperationalJob secondCall = tracker.computeIfAbsent(job.jobId(), id ->
createOperationalJob(CREATED));
+
+ assertThat(secondCall).isSameAs(job);
+ verify(storageProvider, times(1)).persistJob(any());
+ }
+
+ @Test
+ void testComputeIfAbsentUpdatesStatusOnCompletion() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull());
+ });
+ }
+
+ @Test
+ void testComputeIfAbsentUpdatesStatusOnFailure() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
+ MillisecondBoundConfiguration.parse("50ms"),
+ new
org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException("test
failure"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(FAILED), eq("test failure"));
+ });
+ }
+
+ @Test
+ void testComputeIfAbsentRemovesJobFromLocalMapOnCompletion() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ assertThat(tracker.jobsView()).containsKey(jobId);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ assertThat(tracker.jobsView()).doesNotContainKey(jobId);
+ });
+ }
+
+ @Test
+ void testGetReturnsLiveJobFromLocalMap()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(job.jobId(), id -> job);
+
+ OperationalJobInfo result = tracker.get(job.jobId());
+
+ assertThat(result).isSameAs(job);
+ verify(storageProvider, never()).findJob(any());
+ }
+
+ @Test
+ void testGetFallsBackToStorage()
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJobRecord record = new OperationalJobRecord(jobId,
OperationType.DECOMMISSION, SUCCEEDED);
+ when(storageProvider.findJob(jobId)).thenReturn(record);
+
+ OperationalJobInfo result = tracker.get(jobId);
+
+ assertThat(result).isInstanceOf(OperationalJobRecord.class);
+ assertThat(result.jobId()).isEqualTo(jobId);
+ assertThat(result.status()).isEqualTo(SUCCEEDED);
+ assertThat(result.name()).isEqualTo(OperationType.DECOMMISSION.name());
+
assertThat(result.operationType()).isEqualTo(OperationType.DECOMMISSION);
+ verify(storageProvider).findJob(jobId);
+ }
+
+ @Test
+ void testGetReturnsNullWhenNotFoundAnywhere()
+ {
+ UUID jobId = UUIDs.timeBased();
+ when(storageProvider.findJob(jobId)).thenReturn(null);
+
+ OperationalJobInfo result = tracker.get(jobId);
+
+ assertThat(result).isNull();
+ verify(storageProvider).findJob(jobId);
+ }
+
+ @Test
+ void testJobsViewReturnsOnlyLocalMapContents()
+ {
+ OperationalJob job1 = createOperationalJob(CREATED);
+ OperationalJob job2 = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(job1.jobId(), id -> job1);
+ tracker.computeIfAbsent(job2.jobId(), id -> job2);
+
+ Map<UUID, OperationalJob> view = tracker.jobsView();
+
+ assertThat(view).hasSize(2);
+ assertThat(view).containsEntry(job1.jobId(), job1);
+ assertThat(view).containsEntry(job2.jobId(), job2);
+ verify(storageProvider, never()).findJob(any());
+ verify(storageProvider, never()).findAllJobs(any(int.class));
+ }
+
+ @Test
+ void testJobsViewIsImmutable()
+ {
+ OperationalJob job = createOperationalJob(CREATED);
+ tracker.computeIfAbsent(job.jobId(), id -> job);
+
+ Map<UUID, OperationalJob> view = tracker.jobsView();
+
+ assertThatThrownBy(() -> view.put(UUIDs.timeBased(), job))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ @Test
+ void testInflightJobsByOperationFiltersLocalMap()
+ {
+ OperationalJob createdJob = createOperationalJob("decommission",
CREATED);
+ OperationalJob runningJob = createOperationalJob("decommission",
RUNNING);
+ OperationalJob succeededJob = createOperationalJob("decommission",
SUCCEEDED);
+ OperationalJob drainJob = createOperationalJob("drain", RUNNING);
+
+ tracker.computeIfAbsent(createdJob.jobId(), id -> createdJob);
+ tracker.computeIfAbsent(runningJob.jobId(), id -> runningJob);
+ tracker.computeIfAbsent(succeededJob.jobId(), id -> succeededJob);
+ tracker.computeIfAbsent(drainJob.jobId(), id -> drainJob);
+
+ List<OperationalJob> inflightDecommission =
tracker.inflightJobsByOperation("decommission");
+
+ assertThat(inflightDecommission)
+ .hasSize(2)
+ .containsExactlyInAnyOrder(createdJob, runningJob);
+ verify(storageProvider, never()).findJob(any());
+ }
+
+ @Test
+ void testComputeIfAbsentPropagatesPersistenceFailure()
+ {
+ doThrow(new StorageProviderException("Storage unavailable"))
+ .when(storageProvider).persistJob(any());
+ OperationalJob job = createOperationalJob(CREATED);
+
+ assertThatThrownBy(() -> tracker.computeIfAbsent(job.jobId(), id ->
job))
+ .isInstanceOf(StorageProviderException.class)
+ .hasMessage("Storage unavailable");
+ }
+
+ @Test
+ void testGetPropagatesStorageFallbackFailure()
+ {
+ UUID jobId = UUIDs.timeBased();
+ when(storageProvider.findJob(jobId)).thenThrow(new
StorageProviderException("Storage unavailable"));
+
+ assertThatThrownBy(() -> tracker.get(jobId))
+ .isInstanceOf(StorageProviderException.class)
+ .hasMessage("Storage unavailable");
+ }
+
+ @Test
+ void testOnCompleteRetrySucceedsAfterTransientFailure() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ doThrow(new StorageProviderException("Transient failure"))
+ .doNothing()
+ .when(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull());
+
+ CountDownLatch latch = new CountDownLatch(1);
+
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider, times(2)).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull());
+ });
+ }
+
+ @Test
+ void testOnCompleteRemovesJobFromLocalMapWhenAllRetriesFail() throws
InterruptedException
+ {
+ UUID jobId = UUIDs.timeBased();
+ OperationalJob job = OperationalJobTest.createOperationalJob(jobId,
MillisecondBoundConfiguration.parse("50ms"));
+
+ doThrow(new StorageProviderException("Persistent failure"))
+ .when(storageProvider).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull());
+
+ CountDownLatch latch = new CountDownLatch(1);
+ tracker.computeIfAbsent(jobId, id -> job);
+ executorPool.executeBlocking(job::execute);
+
+ job.asyncResult().onComplete(ar -> latch.countDown());
+ assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ loopAssert(2, () -> {
+ verify(storageProvider, times(2)).updateJobStatus(eq(jobId),
eq(OperationType.DECOMMISSION), eq(SUCCEEDED), isNull());
Review Comment:
this assertion is failing
##########
server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.job;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.apache.cassandra.sidecar.config.ServiceConfiguration;
+import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A durable implementation of {@link OperationalJobTracker} that persists job
state
+ * via a {@link StorageProvider}. A local {@link ConcurrentHashMap} caches live
+ * {@link OperationalJob} references for the current process, since executing
jobs
+ * with Vert.x promises cannot be reconstituted from storage. Once a job
completes,
+ * it is removed from the local map, and subsequent lookups are served from
storage.
+ */
+@Singleton
+public class DurableOperationalJobTracker implements OperationalJobTracker
+{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(DurableOperationalJobTracker.class);
+ private static final int MAX_STATUS_UPDATE_ATTEMPTS = 2;
+ private static final long RETRY_DELAY_MS = 100;
+
+ private final ConcurrentHashMap<UUID, OperationalJob> liveJobs;
+ private final StorageProvider storageProvider;
+ private final TaskExecutorPool executor;
+
+ @Inject
+ public DurableOperationalJobTracker(ServiceConfiguration
serviceConfiguration,
+ StorageProvider storageProvider,
+ TaskExecutorPool executor)
+ {
+ this.liveJobs = new
ConcurrentHashMap<>(serviceConfiguration.operationalJobTrackerSize());
+ this.storageProvider = storageProvider;
+ this.executor = executor;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public OperationalJob computeIfAbsent(UUID jobId, Function<UUID,
OperationalJob> mappingFunction)
+ {
+ return liveJobs.computeIfAbsent(jobId, id -> {
+ OperationalJob job = mappingFunction.apply(id);
+
+
storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job));
Review Comment:
Can we do this outside the CHM lock ? it's a bad practice to perform I/O
inside the lock
##########
server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.job;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.apache.cassandra.sidecar.config.ServiceConfiguration;
+import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A durable implementation of {@link OperationalJobTracker} that persists job
state
+ * via a {@link StorageProvider}. A local {@link ConcurrentHashMap} caches live
+ * {@link OperationalJob} references for the current process, since executing
jobs
+ * with Vert.x promises cannot be reconstituted from storage. Once a job
completes,
+ * it is removed from the local map, and subsequent lookups are served from
storage.
+ */
+@Singleton
+public class DurableOperationalJobTracker implements OperationalJobTracker
+{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(DurableOperationalJobTracker.class);
+ private static final int MAX_STATUS_UPDATE_ATTEMPTS = 2;
+ private static final long RETRY_DELAY_MS = 100;
+
+ private final ConcurrentHashMap<UUID, OperationalJob> liveJobs;
+ private final StorageProvider storageProvider;
+ private final TaskExecutorPool executor;
+
+ @Inject
+ public DurableOperationalJobTracker(ServiceConfiguration
serviceConfiguration,
+ StorageProvider storageProvider,
+ TaskExecutorPool executor)
+ {
+ this.liveJobs = new
ConcurrentHashMap<>(serviceConfiguration.operationalJobTrackerSize());
+ this.storageProvider = storageProvider;
Review Comment:
who is expected to initialize the storage provider ? can this class assume
it's already initialized or do we want to initialize here?
##########
server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.job;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.apache.cassandra.sidecar.config.ServiceConfiguration;
+import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A durable implementation of {@link OperationalJobTracker} that persists job
state
+ * via a {@link StorageProvider}. A local {@link ConcurrentHashMap} caches live
+ * {@link OperationalJob} references for the current process, since executing
jobs
+ * with Vert.x promises cannot be reconstituted from storage. Once a job
completes,
+ * it is removed from the local map, and subsequent lookups are served from
storage.
+ */
+@Singleton
+public class DurableOperationalJobTracker implements OperationalJobTracker
+{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(DurableOperationalJobTracker.class);
+ private static final int MAX_STATUS_UPDATE_ATTEMPTS = 2;
+ private static final long RETRY_DELAY_MS = 100;
+
+ private final ConcurrentHashMap<UUID, OperationalJob> liveJobs;
+ private final StorageProvider storageProvider;
+ private final TaskExecutorPool executor;
+
+ @Inject
+ public DurableOperationalJobTracker(ServiceConfiguration
serviceConfiguration,
+ StorageProvider storageProvider,
+ TaskExecutorPool executor)
+ {
+ this.liveJobs = new
ConcurrentHashMap<>(serviceConfiguration.operationalJobTrackerSize());
Review Comment:
Unlike InMemoryOperationalJobTracker which has removeEldestEntry, the
durable tracker has no eviction. If
asyncResult() never completes (e.g., executeBlocking fails to submit due
to full pool), the entry stays in the map forever. Is this expected?
##########
server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java:
##########
@@ -173,6 +179,70 @@ public Map<String, String> operationMetadata()
return operationMetadata;
}
+ @Override
+ @Nullable
+ public UUID nodeId()
+ {
+ return null;
+ }
+
+ @Override
+ public String name()
+ {
+ return operationType.name();
Review Comment:
Live jobs return lowercase from name():
```java
// NodeDecommissionJob
private static final String OPERATION = "decommission";
public String name() { return OPERATION; }
```
OperationalJobRecord (from this patch) returns the Java enum's .name():
```java
// OperationalJobRecord.java
public String name() {
return operationType.name(); // → "DECOMMISSION" (Java enum names are
uppercase)
}
```
The handler builds the API response like this:
```java
// ListOperationalJobsHandler.java:77
.operation(job.name())
```
So a client polling a job sees:
1. While the job is live: "operation": "decommission"
2. After the job completes and is evicted from the local map (subsequent
calls served from storage): "operation": "DECOMMISSION"
The same job changes its operation field from lowercase to uppercase just
because the backing object switched from a live
NodeDecommissionJob to an OperationalJobRecord. That's a breaking change
for any client doing string comparison on the operation field.
##########
server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.job;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.apache.cassandra.sidecar.config.ServiceConfiguration;
+import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A durable implementation of {@link OperationalJobTracker} that persists job
state
+ * via a {@link StorageProvider}. A local {@link ConcurrentHashMap} caches live
+ * {@link OperationalJob} references for the current process, since executing
jobs
+ * with Vert.x promises cannot be reconstituted from storage. Once a job
completes,
+ * it is removed from the local map, and subsequent lookups are served from
storage.
+ */
+@Singleton
+public class DurableOperationalJobTracker implements OperationalJobTracker
+{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(DurableOperationalJobTracker.class);
+ private static final int MAX_STATUS_UPDATE_ATTEMPTS = 2;
+ private static final long RETRY_DELAY_MS = 100;
+
+ private final ConcurrentHashMap<UUID, OperationalJob> liveJobs;
+ private final StorageProvider storageProvider;
+ private final TaskExecutorPool executor;
+
+ @Inject
+ public DurableOperationalJobTracker(ServiceConfiguration
serviceConfiguration,
+ StorageProvider storageProvider,
+ TaskExecutorPool executor)
+ {
+ this.liveJobs = new
ConcurrentHashMap<>(serviceConfiguration.operationalJobTrackerSize());
+ this.storageProvider = storageProvider;
+ this.executor = executor;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public OperationalJob computeIfAbsent(UUID jobId, Function<UUID,
OperationalJob> mappingFunction)
+ {
+ return liveJobs.computeIfAbsent(jobId, id -> {
+ OperationalJob job = mappingFunction.apply(id);
+
+
storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job));
+
+ job.asyncResult().onComplete(ar -> {
+ liveJobs.remove(job.jobId());
Review Comment:
liveJobs removal should come after updateTerminalStatus to avoid `RUNNING →
CREATED → SUCCEEDED.` when polling status between removal and update
##########
server/src/main/java/org/apache/cassandra/sidecar/job/storage/OperationalJobRecord.java:
##########
@@ -173,6 +179,70 @@ public Map<String, String> operationMetadata()
return operationMetadata;
}
+ @Override
+ @Nullable
+ public UUID nodeId()
+ {
+ return null;
Review Comment:
Also why is this field needed, I don't see it being used anywhere.
##########
server/src/main/java/org/apache/cassandra/sidecar/job/DurableOperationalJobTracker.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.job;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
+import org.apache.cassandra.sidecar.config.ServiceConfiguration;
+import org.apache.cassandra.sidecar.job.storage.OperationalJobRecord;
+import org.apache.cassandra.sidecar.job.storage.StorageProvider;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A durable implementation of {@link OperationalJobTracker} that persists job
state
+ * via a {@link StorageProvider}. A local {@link ConcurrentHashMap} caches live
+ * {@link OperationalJob} references for the current process, since executing
jobs
+ * with Vert.x promises cannot be reconstituted from storage. Once a job
completes,
+ * it is removed from the local map, and subsequent lookups are served from
storage.
+ */
+@Singleton
+public class DurableOperationalJobTracker implements OperationalJobTracker
+{
+ private static final Logger LOGGER =
LoggerFactory.getLogger(DurableOperationalJobTracker.class);
+ private static final int MAX_STATUS_UPDATE_ATTEMPTS = 2;
+ private static final long RETRY_DELAY_MS = 100;
+
+ private final ConcurrentHashMap<UUID, OperationalJob> liveJobs;
+ private final StorageProvider storageProvider;
+ private final TaskExecutorPool executor;
+
+ @Inject
+ public DurableOperationalJobTracker(ServiceConfiguration
serviceConfiguration,
+ StorageProvider storageProvider,
+ TaskExecutorPool executor)
+ {
+ this.liveJobs = new
ConcurrentHashMap<>(serviceConfiguration.operationalJobTrackerSize());
+ this.storageProvider = storageProvider;
+ this.executor = executor;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public OperationalJob computeIfAbsent(UUID jobId, Function<UUID,
OperationalJob> mappingFunction)
+ {
+ return liveJobs.computeIfAbsent(jobId, id -> {
+ OperationalJob job = mappingFunction.apply(id);
+
+
storageProvider.persistJob(OperationalJobRecord.fromOperationalJob(job));
Review Comment:
what happens if `!storageProvider.isAvailable` ?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
