You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by fr...@apache.org on 2026/01/28 22:32:00 UTC
(cassandra-sidecar) branch trunk updated: CASSSIDECAR-268 Adds endpoint to perform repair for a given keyspace and options (#231)
This is an automated email from the ASF dual-hosted git repository.
frankgh pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
The following commit(s) were added to refs/heads/trunk by this push:
new 52aed8fc CASSSIDECAR-268 Adds endpoint to perform repair for a given keyspace and options (#231)
52aed8fc is described below
commit 52aed8fc37d03ee80a88ee4d09d97d46a1165a5c
Author: Arjun Ashok <ar...@apple.com>
AuthorDate: Wed Jan 28 14:31:55 2026 -0800
CASSSIDECAR-268 Adds endpoint to perform repair for a given keyspace and options (#231)
Patch by Arjun Ashok; reviewed by Yifan Cai, N V Harikrishna, Francisco Guerrero for CASSSIDECAR-268
---
.../adapters/base/CassandraStorageOperations.java | 20 +
.../sidecar/adapters/base/RepairOptions.java | 79 ++++
.../jmx/GossipDependentStorageJmxOperations.java | 12 +
.../adapters/base/jmx/StorageJmxOperations.java | 16 +
.../cassandra/sidecar/common/ApiEndpointsV1.java | 2 +
.../sidecar/common/request/RepairRequest.java | 63 +++
.../sidecar/common/request/data/RepairPayload.java | 257 ++++++++++++
.../cassandra/sidecar/client/SidecarClient.java | 19 +
.../sidecar/client/SidecarClientTest.java | 24 ++
conf/sidecar.yaml | 10 +
.../testing/SharedClusterIntegrationTestBase.java | 1 -
.../routes/ClearSnapshotIntegrationTest.java | 11 +-
.../sidecar/routes/RepairIntegrationTest.java | 265 ++++++++++++
.../routes/RepairMultiDCIntegrationTest.java | 278 +++++++++++++
.../sidecar/common/server/StorageOperations.java | 18 +
.../acl/authorization/BasicPermissions.java | 1 +
.../sidecar/config/RepairJobsConfiguration.java | 37 ++
.../sidecar/config/ServiceConfiguration.java | 5 +
.../config/yaml/RepairJobsConfigurationImpl.java | 73 ++++
.../config/yaml/ServiceConfigurationImpl.java | 50 ++-
.../cassandra/sidecar/handlers/RepairHandler.java | 179 ++++++++
.../sidecar/handlers/data/RepairRequestParam.java | 59 +++
.../validations/ValidateTableExistenceHandler.java | 1 +
.../cassandra/sidecar/job/NodeDecommissionJob.java | 25 +-
.../apache/cassandra/sidecar/job/NodeDrainJob.java | 15 +-
.../apache/cassandra/sidecar/job/NodeMoveJob.java | 25 +-
.../cassandra/sidecar/job/OperationalJob.java | 38 +-
.../sidecar/job/OperationalJobManager.java | 6 +-
.../apache/cassandra/sidecar/job/RepairJob.java | 437 +++++++++++++++++++
.../sidecar/modules/CassandraOperationsModule.java | 32 +-
.../modules/multibindings/VertxRouteMapKeys.java | 5 +
.../sidecar/utils/OperationalJobUtils.java | 6 +-
.../sidecar/config/SidecarConfigurationTest.java | 28 ++
.../cassandra/sidecar/handlers/CommonTest.java | 56 ++-
.../handlers/ListOperationalJobsHandlerTest.java | 6 +-
.../handlers/NodeDecommissionHandlerTest.java | 41 +-
.../sidecar/handlers/RepairHandlerTest.java | 463 +++++++++++++++++++++
.../cassandra/sidecar/job/NodeDrainJobTest.java | 26 +-
.../cassandra/sidecar/job/NodeMoveJobTest.java | 7 +-
.../sidecar/job/OperationalJobManagerTest.java | 8 +-
.../cassandra/sidecar/job/OperationalJobTest.java | 15 +-
.../cassandra/sidecar/job/RepairJobTest.java | 349 ++++++++++++++++
42 files changed, 2932 insertions(+), 136 deletions(-)
diff --git a/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraStorageOperations.java b/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraStorageOperations.java
index 56dfca31..5ba5e59f 100644
--- a/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraStorageOperations.java
+++ b/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraStorageOperations.java
@@ -257,6 +257,26 @@ public class CassandraStorageOperations implements StorageOperations
.drain();
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public List<String> getParentRepairStatus(int cmd)
+ {
+ return jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME)
+ .getParentRepairStatus(cmd);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int repairAsync(String keyspace, Map<String, String> repairOptions)
+ {
+ StorageJmxOperations ssProxy = jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME);
+ return ssProxy.repairAsync(keyspace, repairOptions);
+ }
+
/**
* {@inheritDoc}
*/
diff --git a/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/RepairOptions.java b/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/RepairOptions.java
new file mode 100644
index 00000000..22045815
--- /dev/null
+++ b/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/RepairOptions.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.adapters.base;
+
+/**
+ * Enum representing the repair options supported.
+ * This class mimics the options supported by the repair operation in Apache Cassandra
+ * as defined in org.apache.cassandra.repair.messages.RepairOption
+ * (<a href="https://github.com/apache/cassandra/blob/trunk/src/java/org/apache/cassandra/repair/messages/RepairOption.java">...</a>)
+ */
+public enum RepairOptions
+{
+ /**
+ * Whether to repair only the primary range of the node (true/false)
+ */
+ PRIMARY_RANGE("primaryRange"),
+ /**
+ * Whether to perform an incremental repair (true/false)
+ * If false, a full repair is performed
+ */
+ INCREMENTAL("incremental"),
+ /**
+ * Specific token ranges to repair
+ */
+ RANGES("ranges"),
+ /**
+ * List of column families (tables) to repair (comma-separated)
+ */
+ COLUMN_FAMILIES("columnFamilies"),
+ /**
+ * Restrict repair to specific data centers (comma-separated)
+ */
+ DATACENTERS("dataCenters"),
+ /**
+ * Restrict repair to specific hosts (comma-separated IPs or hostnames)
+ */
+ HOSTS("hosts"),
+ /**
+ * Force the repair operation
+ */
+ FORCE_REPAIR("forceRepair"),
+ /**
+ * Type of preview repair to run before actual repair
+ * Options: "none", "auto", "running", "full"
+ * Mainly used to assess data consistency without actually repairing
+ */
+ PREVIEW("previewKind");
+
+ private final String value;
+
+ RepairOptions(String value)
+ {
+ this.value = value;
+ }
+
+ /**
+ * @return Name corresponding to the repair option
+ */
+ public String optionName()
+ {
+ return value;
+ }
+}
diff --git a/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/GossipDependentStorageJmxOperations.java b/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/GossipDependentStorageJmxOperations.java
index ae7f561d..b14ac789 100644
--- a/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/GossipDependentStorageJmxOperations.java
+++ b/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/GossipDependentStorageJmxOperations.java
@@ -231,4 +231,16 @@ public class GossipDependentStorageJmxOperations implements StorageJmxOperations
{
delegate.move(newToken);
}
+
+ @Override
+ public int repairAsync(String keyspace, Map<String, String> options)
+ {
+ return delegate.repairAsync(keyspace, options);
+ }
+
+ @Override
+ public List<String> getParentRepairStatus(int cmd)
+ {
+ return delegate.getParentRepairStatus(cmd);
+ }
}
diff --git a/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/StorageJmxOperations.java b/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/StorageJmxOperations.java
index ced40624..a1e49217 100644
--- a/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/StorageJmxOperations.java
+++ b/adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/StorageJmxOperations.java
@@ -240,4 +240,20 @@ public interface StorageJmxOperations
* @param newToken the new token for the node to move to
*/
void move(String newToken) throws IOException;
+
+ /**
+ * Trigger an async repair operation for the given keyspace and options
+ * @param keyspace keyspace to be repaired
+ * @param options repair options
+ * @return repair command number, or 0 if nothing to repair
+ */
+ int repairAsync(String keyspace, Map<String, String> options);
+
+ /**
+ * Get the status of a given parent repair session.
+ * @param cmd the int reference returned when issuing the repair
+ * @return status of parent repair from ParentRepairStatus
+ * followed by final message or messages of the session
+ */
+ List<String> getParentRepairStatus(int cmd);
}
diff --git a/client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java b/client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java
index 8269bf04..024329f3 100644
--- a/client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java
+++ b/client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java
@@ -150,6 +150,8 @@ public final class ApiEndpointsV1
public static final String STREAM_STATS_ROUTE = API_V1 + CASSANDRA + "/stats/streams";
public static final String TABLE_STATS_ROUTE = API_V1 + CASSANDRA + PER_KEYSPACE + PER_TABLE + "/stats";
public static final String COMPACTION_STOP_ROUTE = API_V1 + CASSANDRA + OPERATION_ROUTE + "/compaction/stop";
+ public static final String REPAIR_ROUTE = API_V1 + CASSANDRA + PER_KEYSPACE + "/repair";
+
// Live Migration APIs
public static final String LIVE_MIGRATION_API_PREFIX = API_V1 + "/live-migration";
diff --git a/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/RepairRequest.java b/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/RepairRequest.java
new file mode 100644
index 00000000..f0ee9db6
--- /dev/null
+++ b/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/RepairRequest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.common.request;
+
+import io.netty.handler.codec.http.HttpMethod;
+import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+
+/**
+ * Class representing the repair API request
+ */
+public class RepairRequest extends JsonRequest<OperationalJobResponse>
+{
+ private final RepairPayload repairRequestPayload;
+
+
+ /**
+ * Constructs a Sidecar repair request with the given parameters.
+ *
+ * @param keyspace name of the keyspace in the cluster
+ * @param repairRequestPayload request payload
+ */
+ public RepairRequest(String keyspace, RepairPayload repairRequestPayload)
+ {
+ super(requestURI(keyspace));
+ this.repairRequestPayload = repairRequestPayload;
+ }
+
+ @Override
+ public HttpMethod method()
+ {
+ return HttpMethod.PUT;
+ }
+
+ @Override
+ public Object requestBody()
+ {
+ return repairRequestPayload;
+ }
+
+ static String requestURI(String keyspace)
+ {
+ return ApiEndpointsV1.REPAIR_ROUTE
+ .replaceAll(ApiEndpointsV1.KEYSPACE_PATH_PARAM, keyspace);
+ }
+}
diff --git a/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/RepairPayload.java b/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/RepairPayload.java
new file mode 100644
index 00000000..be92504f
--- /dev/null
+++ b/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/RepairPayload.java
@@ -0,0 +1,257 @@
+/*
+ * 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.common.request.data;
+
+import java.util.List;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonValue;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.cassandra.sidecar.common.DataObjectBuilder;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Request payload for a repair job
+ */
+@JsonDeserialize(builder = RepairPayload.Builder.class)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class RepairPayload
+{
+ private static final String TABLES = "tables";
+ private static final String IS_PRIMARY_RANGE = "primaryRange";
+ private static final String DATACENTER = "datacenter";
+ private static final String HOSTS = "hosts";
+ private static final String START_TOKEN = "startToken";
+ private static final String END_TOKEN = "endToken";
+ private static final String REPAIR_TYPE = "repairType";
+ private static final String FORCE = "force";
+ private static final String VALIDATE = "validate";
+
+ private final List<String> tables;
+ private final Boolean isPrimaryRange;
+ private final String datacenter;
+ private final List<String> hosts;
+ private final String startToken;
+ private final String endToken;
+ private final RepairType repairType;
+ private final Boolean force;
+ private final Boolean validate;
+
+ public static RepairPayload.Builder builder()
+ {
+ return new RepairPayload.Builder();
+ }
+
+ /**
+ * Constructs a new {@link RepairPayload} from the configured {@link RepairPayload.Builder}.
+ *
+ * @param builder the builder used to create this object
+ */
+ protected RepairPayload(RepairPayload.Builder builder)
+ {
+ tables = builder.tables;
+ repairType = builder.repairType;
+ isPrimaryRange = builder.isPrimaryRange;
+ datacenter = builder.datacenter;
+ hosts = builder.hosts;
+ startToken = builder.startToken;
+ endToken = builder.endToken;
+ force = builder.force;
+ validate = builder.validate;
+ }
+
+ @Nullable @JsonProperty(TABLES)
+ public List<String> tables()
+ {
+ return tables;
+ }
+
+ @JsonProperty(IS_PRIMARY_RANGE)
+ public Boolean isPrimaryRange()
+ {
+ return isPrimaryRange;
+ }
+
+ @JsonProperty(DATACENTER)
+ public String datacenter()
+ {
+ return datacenter;
+ }
+
+ @JsonProperty(HOSTS)
+ public List<String> hosts()
+ {
+ return hosts;
+ }
+
+ @JsonProperty(START_TOKEN)
+ public String startToken()
+ {
+ return startToken;
+ }
+
+ @JsonProperty(END_TOKEN)
+ public String endToken()
+ {
+ return endToken;
+ }
+
+ @JsonProperty(REPAIR_TYPE)
+ public RepairType repairType()
+ {
+ return repairType;
+ }
+
+ @Nullable @JsonProperty(FORCE)
+ public Boolean force()
+ {
+ return force;
+ }
+
+ @JsonProperty(VALIDATE)
+ public Boolean shouldValidate()
+ {
+ return validate;
+ }
+
+ /**
+ * Enum representing types of repair supported
+ */
+ public enum RepairType
+ {
+ FULL,
+ INCREMENTAL;
+
+ @JsonValue
+ public String getValue()
+ {
+ return name().toLowerCase();
+ }
+
+ @JsonCreator
+ public static RepairType fromValue(@NotNull String text)
+ {
+ String trimmed = text.trim();
+ try
+ {
+ if (trimmed.isEmpty())
+ {
+ throw new IllegalArgumentException("Unexpected value: " + text);
+ }
+
+ return valueOf(trimmed.toUpperCase());
+ }
+ catch (IllegalArgumentException e)
+ {
+ throw new IllegalArgumentException("Unexpected value: " + text);
+ }
+ }
+ }
+
+ /**
+ * Builder implementation to construct a {@code RepairPayload}
+ */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static final class Builder implements DataObjectBuilder<RepairPayload.Builder, RepairPayload>
+ {
+ private List<String> tables;
+ private Boolean isPrimaryRange;
+ private String datacenter;
+ private List<String> hosts;
+ private String startToken;
+ private String endToken;
+ private RepairType repairType;
+ private Boolean force;
+ private Boolean validate;
+
+ private Builder()
+ {
+ }
+
+ @Override
+ public RepairPayload.Builder self()
+ {
+ return this;
+ }
+
+ @JsonProperty(TABLES)
+ public RepairPayload.Builder tables(List<String> tables)
+ {
+ return update(b -> b.tables = tables);
+ }
+
+ @JsonProperty(IS_PRIMARY_RANGE)
+ public RepairPayload.Builder isPrimaryRange(boolean isPrimaryRange)
+ {
+ return update(b -> b.isPrimaryRange = isPrimaryRange);
+ }
+
+ @JsonProperty(DATACENTER)
+ public RepairPayload.Builder datacenter(String datacenter)
+ {
+ return update(b -> b.datacenter = datacenter);
+ }
+
+ @JsonProperty(REPAIR_TYPE)
+ public RepairPayload.Builder repairType(RepairType type)
+ {
+ return update(b -> b.repairType = type);
+ }
+
+ @JsonProperty(HOSTS)
+ public RepairPayload.Builder hosts(List<String> hosts)
+ {
+ return update(b -> b.hosts = hosts);
+ }
+
+ @JsonProperty(START_TOKEN)
+ public RepairPayload.Builder startToken(String startToken)
+ {
+ return update(b -> b.startToken = startToken);
+ }
+
+ @JsonProperty(END_TOKEN)
+ public RepairPayload.Builder endToken(String endToken)
+ {
+ return update(b -> b.endToken = endToken);
+ }
+
+ @JsonProperty(FORCE)
+ public RepairPayload.Builder force(boolean force)
+ {
+ return update(b -> b.force = force);
+ }
+
+ @JsonProperty(VALIDATE)
+ public RepairPayload.Builder shouldValidate(boolean validate)
+ {
+ return update(b -> b.validate = validate);
+ }
+
+ @Override
+ public RepairPayload build()
+ {
+ return new RepairPayload(this);
+ }
+ }
+}
diff --git a/client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java b/client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java
index 15d0db85..6d6068c7 100644
--- a/client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java
+++ b/client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java
@@ -46,6 +46,7 @@ import org.apache.cassandra.sidecar.common.request.ImportSSTableRequest;
import org.apache.cassandra.sidecar.common.request.ListCdcSegmentsRequest;
import org.apache.cassandra.sidecar.common.request.LiveMigrationListInstanceFilesRequest;
import org.apache.cassandra.sidecar.common.request.LiveMigrationStatusRequest;
+import org.apache.cassandra.sidecar.common.request.RepairRequest;
import org.apache.cassandra.sidecar.common.request.RestoreJobProgressRequest;
import org.apache.cassandra.sidecar.common.request.RestoreJobSummaryRequest;
import org.apache.cassandra.sidecar.common.request.Service;
@@ -60,6 +61,7 @@ import org.apache.cassandra.sidecar.common.request.data.CreateRestoreJobRequestP
import org.apache.cassandra.sidecar.common.request.data.CreateSliceRequestPayload;
import org.apache.cassandra.sidecar.common.request.data.Digest;
import org.apache.cassandra.sidecar.common.request.data.NodeCommandRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
import org.apache.cassandra.sidecar.common.request.data.RestoreJobProgressRequestParams;
import org.apache.cassandra.sidecar.common.request.data.UpdateCdcServiceConfigPayload;
import org.apache.cassandra.sidecar.common.request.data.UpdateRestoreJobRequestPayload;
@@ -859,6 +861,23 @@ public class SidecarClient implements AutoCloseable, SidecarClientBlobRestoreExt
.build());
}
+ /**
+ * Executes a repair operation on the provided instance, keyspace and options
+ * @param instance the instance where the request will be executed
+ * @param keyspace keyspace for which the repair is being performed
+ * @param payload the repair request options as payload
+ * @return a completable future of the repair operation job response
+ */
+ public CompletableFuture<OperationalJobResponse> repair(SidecarInstance instance,
+ String keyspace,
+ RepairPayload payload)
+ {
+ return executor.executeRequestAsync(requestBuilder()
+ .singleInstanceSelectionPolicy(instance)
+ .request(new RepairRequest(keyspace, payload))
+ .build());
+ }
+
/**
* Executes the node drain request using the default retry policy and configured selection policy
*
diff --git a/client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java b/client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java
index 4eb570de..336f636f 100644
--- a/client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java
+++ b/client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java
@@ -77,6 +77,7 @@ import org.apache.cassandra.sidecar.common.request.data.CompactionStopRequestPay
import org.apache.cassandra.sidecar.common.request.data.CreateRestoreJobRequestPayload;
import org.apache.cassandra.sidecar.common.request.data.MD5Digest;
import org.apache.cassandra.sidecar.common.request.data.NodeCommandRequestPayload;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
import org.apache.cassandra.sidecar.common.request.data.UpdateCdcServiceConfigPayload;
import org.apache.cassandra.sidecar.common.request.data.XXHash32Digest;
import org.apache.cassandra.sidecar.common.response.CompactionStatsResponse;
@@ -1496,6 +1497,29 @@ abstract class SidecarClientTest
});
}
+ @Test
+ public void testRepair() throws Exception
+ {
+ UUID jobId = UUID.randomUUID();
+ String repairResponseString = "{\"jobId\":\"" + jobId + "\",\"jobStatus\":\"SUCCEEDED\",\"instance\":\"127.0.0.1\"}";
+
+ MockResponse response = new MockResponse()
+ .setResponseCode(OK.code())
+ .setHeader("content-type", "application/json")
+ .setBody(repairResponseString);
+ enqueue(response);
+ RepairPayload payload = RepairPayload.builder()
+ .tables(Collections.singletonList("test_table"))
+ .isPrimaryRange(true)
+ .build();
+
+ SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(servers.get(0));
+ OperationalJobResponse result = client.repair(sidecarInstance, "testkeyspace", payload).get(30, TimeUnit.SECONDS);
+ assertThat(result).isNotNull();
+ assertThat(result.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ validateResponseServed(ApiEndpointsV1.REPAIR_ROUTE.replaceAll(KEYSPACE_PATH_PARAM, "testkeyspace"));
+ }
+
@Test
void testFailsWithOneAttemptPerServer()
{
diff --git a/conf/sidecar.yaml b/conf/sidecar.yaml
index 631a6a27..1d6746fc 100644
--- a/conf/sidecar.yaml
+++ b/conf/sidecar.yaml
@@ -86,6 +86,10 @@ sidecar:
# values are 'default' and 'resolve_to_ip'. Default will resolve hostnames to addresses and addresses to
# hostnames whereas resolve_to_ip will only resolve hostnames to addresses.
dns_resolver: default
+ # The maximum wait time for operational job to run internally before returning the HTTP response;
+ # if the job finishes before the max wait time, it returns immediately on completion;
+ # otherwise, a response indicating the job is still running is returned after the maximum wait time.
+ operations_job_execution_max_wait_time: 5s
throttle:
stream_requests_per_sec: 5000
timeout: 10s
@@ -134,6 +138,12 @@ sidecar:
replication_factor: 1
# The TTL in seconds used to insert entries into the sidecar_lease schema
lease_schema_ttl: 5m
+ # Captures configuration parameters for repair jobs
+ repair:
+ # The maximum retry attempts for the repair job status to be valid
+ status_attempts: 5
+ # The polling interval for repair job status
+ status_polling_interval: 2s
coordination:
# Captures configuration parameters for the task that performs the cluster lease claim process
cluster_lease_claim:
diff --git a/integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java b/integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java
index 101a40ae..900fd513 100644
--- a/integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java
+++ b/integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java
@@ -162,7 +162,6 @@ public abstract class SharedClusterIntegrationTestBase
{
protected final Logger logger = LoggerFactory.getLogger(SharedClusterIntegrationTestBase.class);
private static final int MAX_CLUSTER_PROVISION_RETRIES = 5;
-
@TempDir
static Path secretsPath;
diff --git a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/ClearSnapshotIntegrationTest.java b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/ClearSnapshotIntegrationTest.java
index a8d198f8..c4f08e62 100644
--- a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/ClearSnapshotIntegrationTest.java
+++ b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/ClearSnapshotIntegrationTest.java
@@ -19,6 +19,8 @@
package org.apache.cassandra.sidecar.routes;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.List;
import java.util.UUID;
@@ -86,8 +88,8 @@ class ClearSnapshotIntegrationTest extends SharedClusterSidecarIntegrationTestBa
{
String snapshotName = "my-snapshot-" + UUID.randomUUID();
String testRoute = String.format(SNAPSHOT_ROUTE_TEMPLATE,
- tableName.maybeQuotedKeyspace(),
- tableName.maybeQuotedTable(),
+ urlEncode(tableName.maybeQuotedKeyspace()),
+ urlEncode(tableName.maybeQuotedTable()),
snapshotName);
// Create the snapshot
@@ -119,6 +121,11 @@ class ClearSnapshotIntegrationTest extends SharedClusterSidecarIntegrationTestBa
assertThat(response.statusCode()).isEqualTo(NOT_FOUND.code());
}
+ private String urlEncode(String value)
+ {
+ return URLEncoder.encode(value, StandardCharsets.UTF_8);
+ }
+
@Override
protected void initializeSchemaForTest()
{
diff --git a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/RepairIntegrationTest.java b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/RepairIntegrationTest.java
new file mode 100644
index 00000000..7c6b1776
--- /dev/null
+++ b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/RepairIntegrationTest.java
@@ -0,0 +1,265 @@
+/*
+ * 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.routes;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.IntStream;
+import java.util.stream.StreamSupport;
+
+import org.junit.jupiter.api.Test;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpResponseExpectation;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.web.client.HttpResponse;
+import io.vertx.ext.web.codec.BodyCodec;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.SimpleQueryResult;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.testing.QualifiedName;
+import org.apache.cassandra.sidecar.testing.SharedClusterSidecarIntegrationTestBase;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
+import org.apache.cassandra.testing.TestUtils;
+
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.RUNNING;
+import static org.apache.cassandra.testing.TestUtils.DC1_RF3;
+import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
+import static org.apache.cassandra.testing.TestUtils.TEST_TABLE_PREFIX;
+import static org.apache.cassandra.testing.utils.AssertionUtils.getBlocking;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+
+/**
+ * Integration tests for repair operations
+ */
+class RepairIntegrationTest extends SharedClusterSidecarIntegrationTestBase
+{
+ private static final String INSERT_STMT = "INSERT INTO %s (race_year, race_name, rank, cyclist_name) " +
+ "VALUES (2015, 'Tour of Japan - Stage 4 - Minami > Shinshu', %d, 'Benjamin PRADES')";
+
+ private static final String SELECT_STMT = "SELECT * FROM %s";
+
+ private static final String CREATE_STMT = "CREATE TABLE %s ( \n" +
+ " race_year int, \n" +
+ " race_name text, \n" +
+ " cyclist_name text, \n" +
+ " rank int, \n" +
+ " PRIMARY KEY ((race_year, race_name), rank) \n" +
+ ") WITH read_repair='NONE';";
+
+ private static final int NUM_ROWS = 300;
+
+ static final QualifiedName REPAIR_TEST_TABLE = TestUtils.uniqueTestTableFullName(TEST_KEYSPACE, TEST_TABLE_PREFIX);
+ static final QualifiedName REPAIR_REPLICAS_WITH_DATA_TABLE = TestUtils.uniqueTestTableFullName(TEST_KEYSPACE, TEST_TABLE_PREFIX);
+ static final QualifiedName REPAIR_REPLICAS_IR_WITH_DATA_TABLE = TestUtils.uniqueTestTableFullName(TEST_KEYSPACE, TEST_TABLE_PREFIX);
+ static final QualifiedName REPAIR_REPLICAS_TIMEOUT_TABLE = TestUtils.uniqueTestTableFullName(TEST_KEYSPACE, TEST_TABLE_PREFIX);
+
+ static final List<QualifiedName> ALL_TABLES = List.of(REPAIR_TEST_TABLE,
+ REPAIR_REPLICAS_WITH_DATA_TABLE,
+ REPAIR_REPLICAS_IR_WITH_DATA_TABLE,
+ REPAIR_REPLICAS_TIMEOUT_TABLE);
+
+ @Override
+ protected ClusterBuilderConfiguration testClusterConfiguration()
+ {
+ return super.testClusterConfiguration()
+ .nodesPerDc(3)
+ .requestFeature(Feature.NETWORK);
+ }
+
+ @Override
+ protected void initializeSchemaForTest()
+ {
+ createTestKeyspace(TEST_KEYSPACE, DC1_RF3);
+ for (QualifiedName table : ALL_TABLES)
+ {
+ createTestTable(table, CREATE_STMT);
+ }
+ populateData(REPAIR_TEST_TABLE, 10);
+
+ cluster.stream().forEach(node -> node.nodetoolResult("disableautocompaction",
+ REPAIR_REPLICAS_WITH_DATA_TABLE.keyspace(),
+ REPAIR_REPLICAS_WITH_DATA_TABLE.table())
+ .asserts().success());
+ populateDataSingleNode(REPAIR_REPLICAS_WITH_DATA_TABLE, 2, NUM_ROWS);
+ validateDataConsistency(REPAIR_REPLICAS_WITH_DATA_TABLE, List.of(2), NUM_ROWS);
+ validateDataConsistency(REPAIR_REPLICAS_WITH_DATA_TABLE, List.of(3, 1), 0);
+
+
+ cluster.stream().forEach(node -> node.nodetoolResult("disableautocompaction",
+ REPAIR_REPLICAS_IR_WITH_DATA_TABLE.keyspace(),
+ REPAIR_REPLICAS_IR_WITH_DATA_TABLE.table())
+ .asserts().success());
+ populateDataSingleNode(REPAIR_REPLICAS_IR_WITH_DATA_TABLE, 2, NUM_ROWS);
+ validateDataConsistency(REPAIR_REPLICAS_IR_WITH_DATA_TABLE, List.of(2), NUM_ROWS);
+ validateDataConsistency(REPAIR_REPLICAS_IR_WITH_DATA_TABLE, List.of(3, 1), 0);
+
+ cluster.stream().forEach(node -> node.nodetoolResult("disableautocompaction",
+ REPAIR_REPLICAS_TIMEOUT_TABLE.keyspace(),
+ REPAIR_REPLICAS_TIMEOUT_TABLE.table())
+ .asserts().success());
+ populateDataSingleNode(REPAIR_REPLICAS_TIMEOUT_TABLE, 2, NUM_ROWS);
+ validateDataConsistency(REPAIR_REPLICAS_TIMEOUT_TABLE, List.of(2), NUM_ROWS);
+ validateDataConsistency(REPAIR_REPLICAS_TIMEOUT_TABLE, List.of(3, 1), 0);
+ }
+
+ @Test
+ void repairTest()
+ {
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of(REPAIR_TEST_TABLE.table()))
+ .build();
+ String testRoute = "/api/v1/cassandra/keyspaces/" + TEST_KEYSPACE + "/repair";
+ OperationalJobResponse response = getBlocking(trustedClient().put(serverWrapper.serverPort, "localhost", testRoute)
+ .as(BodyCodec.json(OperationalJobResponse.class))
+ .sendJson(JsonObject.mapFrom(payload))
+ .expecting(HttpResponseExpectation.SC_ACCEPTED))
+ .body();
+
+ assertThat(response).isNotNull();
+ assertThat(response.jobId()).isNotNull();
+ pollStatusForState(response.jobId().toString());
+ }
+
+ @Test
+ void repairReplicasWithData()
+ {
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of(REPAIR_REPLICAS_WITH_DATA_TABLE.table()))
+ .build();
+ String testRoute = "/api/v1/cassandra/keyspaces/" + TEST_KEYSPACE + "/repair";
+ OperationalJobResponse response = getBlocking(trustedClient().put(serverWrapper.serverPort, "localhost", testRoute)
+ .as(BodyCodec.json(OperationalJobResponse.class))
+ .sendJson(JsonObject.mapFrom(payload))
+ .expecting(HttpResponseExpectation.SC_ACCEPTED))
+ .body();
+
+ assertThat(response).isNotNull();
+ assertThat(response.jobId()).isNotNull();
+ assertThat(response.status()).isEqualTo(RUNNING);
+
+ pollStatusForState(response.jobId().toString());
+ validateDataConsistency(REPAIR_REPLICAS_WITH_DATA_TABLE, List.of(2, 3, 1), NUM_ROWS);
+ }
+
+ @Test
+ void repairReplicasIRWithData()
+ {
+ RepairPayload payload = RepairPayload.builder()
+ .tables(List.of(REPAIR_REPLICAS_IR_WITH_DATA_TABLE.table()))
+ .repairType(RepairPayload.RepairType.INCREMENTAL)
+ .build();
+ String testRoute = "/api/v1/cassandra/keyspaces/" + TEST_KEYSPACE + "/repair";
+ OperationalJobResponse response = getBlocking(trustedClient().put(serverWrapper.serverPort, "localhost", testRoute)
+ .as(BodyCodec.json(OperationalJobResponse.class))
+ .sendJson(JsonObject.mapFrom(payload))
+ .expecting(HttpResponseExpectation.SC_ACCEPTED))
+ .body();
+
+ assertThat(response).isNotNull();
+ assertThat(response.jobId()).isNotNull();
+ assertThat(response.status()).isEqualTo(RUNNING);
+
+ pollStatusForState(response.jobId().toString());
+ validateDataConsistency(REPAIR_REPLICAS_IR_WITH_DATA_TABLE, List.of(2, 3, 1), NUM_ROWS);
+ }
+
+ @Test
+ void repairReplicasTimeout()
+ {
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of(REPAIR_REPLICAS_TIMEOUT_TABLE.table()))
+ .build();
+ String testRoute = "/api/v1/cassandra/keyspaces/" + TEST_KEYSPACE + "/repair";
+ OperationalJobResponse response = getBlocking(trustedClient().put(serverWrapper.serverPort, "localhost", testRoute)
+ .as(BodyCodec.json(OperationalJobResponse.class))
+ .sendJson(JsonObject.mapFrom(payload))
+ .expecting(HttpResponseExpectation.SC_ACCEPTED))
+ .body();
+
+ assertThat(response).isNotNull();
+ assertThat(response.jobId()).isNotNull();
+ assertThat(response.status()).isEqualTo(RUNNING);
+
+ pollStatusForState(response.jobId().toString());
+ validateDataConsistency(REPAIR_REPLICAS_TIMEOUT_TABLE, List.of(2, 3, 1), NUM_ROWS);
+ }
+
+ private void pollStatusForState(String uuid)
+ {
+ String status = "/api/v1/cassandra/operational-jobs/" + uuid;
+ loopAssert(30, 500, () -> {
+ HttpResponse<Buffer> resp = getBlocking(trustedClient().get(serverWrapper.serverPort, "localhost", status)
+ .send());
+ logger.info("Success Status Response code: {}", resp.statusCode());
+ logger.info("Status Response: {}", resp.bodyAsString());
+ if (resp.statusCode() == HttpResponseStatus.OK.code())
+ {
+ OperationalJobResponse jobStatusResp = resp.bodyAsJson(OperationalJobResponse.class);
+ assertThat(jobStatusResp.jobId()).isEqualTo(UUID.fromString(uuid));
+ assertThat(jobStatusResp.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ assertThat(jobStatusResp.operation()).isEqualTo("repair");
+ }
+ else
+ {
+ assertThat(resp.statusCode()).isEqualTo(HttpResponseStatus.ACCEPTED.code());
+ OperationalJobResponse jobStatusResp = resp.bodyAsJson(OperationalJobResponse.class);
+ assertThat(jobStatusResp.jobId()).isEqualTo(UUID.fromString(uuid));
+ fail("Repair is still in progress");
+ }
+ });
+ }
+
+ private void validateDataConsistency(QualifiedName tableName, List<Integer> nodes, long expectedNumRows)
+ {
+ nodes.forEach(nodeNumber -> {
+ IInstance node = cluster.get(nodeNumber);
+ SimpleQueryResult rows = node.executeInternalWithResult(String.format(SELECT_STMT, tableName));
+ long rowCount = StreamSupport.stream(rows.spliterator(), false).count();
+ assertThat(expectedNumRows).isEqualTo(rowCount);
+ });
+ }
+
+ private void populateDataSingleNode(QualifiedName tableName, int nodeNumber, int numRows)
+ {
+ IInstance node = cluster.get(nodeNumber);
+ IntStream.rangeClosed(1, numRows)
+ .forEach(i -> {
+ node.executeInternal(String.format(INSERT_STMT, tableName, i));
+ node.flush(TEST_KEYSPACE);
+ });
+ }
+
+ private void populateData(QualifiedName tableName, int numRows)
+ {
+ IntStream.rangeClosed(1, numRows)
+ .mapToObj(i -> String.format(INSERT_STMT, tableName, i))
+ .forEach(cluster::schemaChangeIgnoringStoppedInstances);
+ }
+}
diff --git a/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/RepairMultiDCIntegrationTest.java b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/RepairMultiDCIntegrationTest.java
new file mode 100644
index 00000000..02eb49f9
--- /dev/null
+++ b/integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/RepairMultiDCIntegrationTest.java
@@ -0,0 +1,278 @@
+/*
+ * 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.routes;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.IntStream;
+import java.util.stream.StreamSupport;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpResponseExpectation;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.web.client.HttpResponse;
+import io.vertx.ext.web.codec.BodyCodec;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.SimpleQueryResult;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.testing.QualifiedName;
+import org.apache.cassandra.sidecar.testing.SharedClusterSidecarIntegrationTestBase;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
+import org.apache.cassandra.testing.TestUtils;
+
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.RUNNING;
+import static org.apache.cassandra.testing.TestUtils.DC1_RF2_DC2_RF2;
+import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE;
+import static org.apache.cassandra.testing.TestUtils.TEST_TABLE_PREFIX;
+import static org.apache.cassandra.testing.utils.AssertionUtils.getBlocking;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+
+/**
+ * Integration tests for repair operations in multi-datacenter environments.
+ * This test class validates both cross-DC repair (default behavior when no datacenter is specified)
+ * and single-DC repair (when a specific datacenter is specified) scenarios.
+ */
+@Tag("heavy")
+class RepairMultiDCIntegrationTest extends SharedClusterSidecarIntegrationTestBase
+{
+ private static final String INSERT_STMT = "INSERT INTO %s (race_year, race_name, rank, cyclist_name) " +
+ "VALUES (2015, 'Tour of Japan - Stage 4 - Minami > Shinshu', %d, 'Benjamin PRADES')";
+
+ private static final String SELECT_STMT = "SELECT * FROM %s";
+
+ private static final String CREATE_STMT = "CREATE TABLE %s ( \n" +
+ " race_year int, \n" +
+ " race_name text, \n" +
+ " cyclist_name text, \n" +
+ " rank int, \n" +
+ " PRIMARY KEY ((race_year, race_name), rank) \n" +
+ ") WITH read_repair='NONE'";
+
+ private static final int NUM_ROWS = 100;
+
+ static final QualifiedName CROSS_DC_REPAIR_TABLE = TestUtils.uniqueTestTableFullName(TEST_KEYSPACE, TEST_TABLE_PREFIX);
+ static final QualifiedName SINGLE_DC_REPAIR_TABLE = TestUtils.uniqueTestTableFullName(TEST_KEYSPACE, TEST_TABLE_PREFIX);
+ static final QualifiedName CROSS_DC_IR_TABLE = TestUtils.uniqueTestTableFullName(TEST_KEYSPACE, TEST_TABLE_PREFIX);
+
+ static final List<QualifiedName> ALL_TABLES = List.of(CROSS_DC_REPAIR_TABLE,
+ SINGLE_DC_REPAIR_TABLE,
+ CROSS_DC_IR_TABLE);
+
+ @Override
+ protected ClusterBuilderConfiguration testClusterConfiguration()
+ {
+ return super.testClusterConfiguration()
+ .nodesPerDc(2) // 2 nodes per datacenter
+ .dcCount(2) // 2 datacenters (datacenter1 and datacenter2)
+ .requestFeature(Feature.NETWORK);
+ }
+
+ @Override
+ protected void initializeSchemaForTest()
+ {
+ // Create keyspace with replication across both datacenters
+ createTestKeyspace(TEST_KEYSPACE, DC1_RF2_DC2_RF2);
+
+ for (QualifiedName table : ALL_TABLES)
+ {
+ createTestTable(table, CREATE_STMT);
+
+ // Disable auto-compaction to ensure data inconsistency persists until repair
+ cluster.stream().forEach(node -> node.nodetoolResult("disableautocompaction", table.keyspace(), table.table())
+ .asserts().success());
+
+ // Create data inconsistency by writing to only one node in datacenter1
+ populateDataSingleNode(table, 1, NUM_ROWS); // Node 1 is in datacenter1
+
+ // Verify initial data distribution - only datacenter1 node 1 should have data
+ validateDataConsistency(table, List.of(1), NUM_ROWS); // datacenter1 node 1 has data
+ validateDataConsistency(table, List.of(2, 3, 4), 0); // Other nodes have no data
+ }
+ }
+
+ /**
+ * Test cross-datacenter repair (default behavior when no datacenter is specified).
+ */
+ @Test
+ void testCrossDCRepair()
+ {
+ // Perform cross-DC repair (no datacenter specified = repair across all DCs)
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of(CROSS_DC_REPAIR_TABLE.table()))
+ .build(); // No datacenter specified - should repair across all DCs
+
+ String testRoute = "/api/v1/cassandra/keyspaces/" + TEST_KEYSPACE + "/repair";
+ OperationalJobResponse response = getBlocking(trustedClient().put(serverWrapper.serverPort, "localhost", testRoute)
+ .as(BodyCodec.json(OperationalJobResponse.class))
+ .sendJson(JsonObject.mapFrom(payload))
+ .expecting(HttpResponseExpectation.SC_ACCEPTED))
+ .body();
+
+ assertThat(response).isNotNull();
+ assertThat(response.jobId()).isNotNull();
+ assertThat(response.status()).isEqualTo(RUNNING);
+
+ // Wait for repair to complete
+ pollStatusForState(response.jobId().toString());
+
+ // Verify data is now consistent across ALL nodes in BOTH datacenters
+ validateDataConsistency(CROSS_DC_REPAIR_TABLE, List.of(1, 2, 3, 4), NUM_ROWS);
+ }
+
+ /**
+ * Test single-datacenter repair (when a specific datacenter is specified).
+ * This validates that repair can be limited to a specific datacenter.
+ */
+ @Test
+ void testSingleDCRepair()
+ {
+ // Debug: Print datacenter information for each node
+ for (int i = 1; i <= 4; i++)
+ {
+ IInstance node = cluster.get(i);
+ String datacenter = node.config().localDatacenter();
+ logger.info("Node {} is in datacenter: {}", i, datacenter);
+ }
+
+ // Perform single-DC repair (specify datacenter1 only)
+ RepairPayload payload = RepairPayload.builder()
+ .datacenter("datacenter1") // Repair only datacenter1
+ .isPrimaryRange(true)
+ .tables(List.of(SINGLE_DC_REPAIR_TABLE.table()))
+ .build();
+
+ String testRoute = "/api/v1/cassandra/keyspaces/" + TEST_KEYSPACE + "/repair";
+ OperationalJobResponse response = getBlocking(trustedClient().put(serverWrapper.serverPort, "localhost", testRoute)
+ .as(BodyCodec.json(OperationalJobResponse.class))
+ .sendJson(JsonObject.mapFrom(payload))
+ .expecting(HttpResponseExpectation.SC_ACCEPTED))
+ .body();
+
+ assertThat(response).isNotNull();
+ assertThat(response.jobId()).isNotNull();
+ assertThat(response.status()).isEqualTo(RUNNING);
+
+ // Wait for repair to complete
+ pollStatusForState(response.jobId().toString());
+
+ // Debug: Check data on all nodes after repair
+ for (int i = 1; i <= 4; i++)
+ {
+ IInstance node = cluster.get(i);
+ SimpleQueryResult rows = node.executeInternalWithResult(String.format(SELECT_STMT, SINGLE_DC_REPAIR_TABLE));
+ long rowCount = StreamSupport.stream(rows.spliterator(), false).count();
+ String datacenter = node.config().localDatacenter();
+ logger.info("After repair - Node {} (DC: {}) has {} rows", i, datacenter, rowCount);
+ }
+
+ // Based on actual node distribution:
+ // datacenter1: nodes 1, 3
+ // datacenter2: nodes 2, 4
+ validateDataConsistency(SINGLE_DC_REPAIR_TABLE, List.of(1, 3), NUM_ROWS); // datacenter1 nodes should be consistent
+ validateDataConsistency(SINGLE_DC_REPAIR_TABLE, List.of(2, 4), 0); // datacenter2 nodes should still have no data
+ }
+
+ /**
+ * Test incremental repair across datacenters.
+ */
+ @Test
+ void testCrossDCIncrementalRepair()
+ {
+ // Perform incremental cross-DC repair
+ RepairPayload payload = RepairPayload.builder()
+ .repairType(RepairPayload.RepairType.INCREMENTAL)
+ .tables(List.of(CROSS_DC_IR_TABLE.table()))
+ .build(); // No datacenter = cross-DC repair
+
+ String testRoute = "/api/v1/cassandra/keyspaces/" + TEST_KEYSPACE + "/repair";
+ OperationalJobResponse response = getBlocking(trustedClient().put(serverWrapper.serverPort, "localhost", testRoute)
+ .as(BodyCodec.json(OperationalJobResponse.class))
+ .sendJson(JsonObject.mapFrom(payload))
+ .expecting(HttpResponseExpectation.SC_ACCEPTED))
+ .body();
+
+ assertThat(response).isNotNull();
+ assertThat(response.jobId()).isNotNull();
+ assertThat(response.status()).isEqualTo(RUNNING);
+
+ // Wait for repair to complete
+ pollStatusForState(response.jobId().toString());
+
+ // Verify data consistency across all nodes
+ validateDataConsistency(CROSS_DC_IR_TABLE, List.of(1, 2, 3, 4), NUM_ROWS);
+ }
+
+ private void pollStatusForState(String uuid)
+ {
+ String status = "/api/v1/cassandra/operational-jobs/" + uuid;
+ loopAssert(30, 500, () -> {
+ HttpResponse<Buffer> resp = getBlocking(trustedClient().get(serverWrapper.serverPort, "localhost", status)
+ .send());
+ logger.info("Success Status Response code: {}", resp.statusCode());
+ logger.info("Status Response: {}", resp.bodyAsString());
+ if (resp.statusCode() == HttpResponseStatus.OK.code())
+ {
+ OperationalJobResponse jobStatusResp = resp.bodyAsJson(OperationalJobResponse.class);
+ assertThat(jobStatusResp.jobId()).isEqualTo(UUID.fromString(uuid));
+ assertThat(jobStatusResp.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ assertThat(jobStatusResp.reason()).isNull();
+ assertThat(jobStatusResp.operation()).isEqualTo("repair");
+ }
+ else
+ {
+ assertThat(resp.statusCode()).isEqualTo(HttpResponseStatus.ACCEPTED.code());
+ OperationalJobResponse jobStatusResp = resp.bodyAsJson(OperationalJobResponse.class);
+ assertThat(jobStatusResp.jobId()).isEqualTo(UUID.fromString(uuid));
+ fail("Repair is still in progress");
+ }
+ });
+ }
+
+ private void validateDataConsistency(QualifiedName tableName, List<Integer> nodes, long expectedNumRows)
+ {
+ nodes.forEach(nodeNumber -> {
+ IInstance node = cluster.get(nodeNumber);
+ SimpleQueryResult rows = node.executeInternalWithResult(String.format(SELECT_STMT, tableName));
+ long rowCount = StreamSupport.stream(rows.spliterator(), false).count();
+ assertThat(rowCount).as("Node %d should have %d rows", nodeNumber, expectedNumRows)
+ .isEqualTo(expectedNumRows);
+ });
+ }
+
+ private void populateDataSingleNode(QualifiedName tableName, int nodeNumber, int numRows)
+ {
+ IInstance node = cluster.get(nodeNumber);
+ IntStream.rangeClosed(1, numRows)
+ .forEach(i -> {
+ node.executeInternal(String.format(INSERT_STMT, tableName, i));
+ node.flush(TEST_KEYSPACE);
+ });
+ }
+}
diff --git a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/StorageOperations.java b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/StorageOperations.java
index 163a30c9..a5a72a2b 100644
--- a/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/StorageOperations.java
+++ b/server-common/src/main/java/org/apache/cassandra/sidecar/common/server/StorageOperations.java
@@ -135,6 +135,24 @@ public interface StorageOperations
*/
String clusterName();
+ /**
+ * Triggers a repair operation for the given keyspace and options
+ *
+ * @param keyspace keyspace for the repair operation
+ * @param options repair options
+ * @return an integer value representing the status of the repair operation; Only returns 0 for replication factor 1
+ * which can be used as a reference to check for the status of the repair session via {@link #getParentRepairStatus(int)}.
+ */
+ int repairAsync(String keyspace, Map<String, String> options);
+
+ /**
+ * Get the status of a given parent repair session.
+ *
+ * @param cmd the integer value representing a reference to a repair session
+ * @return status of parent repair
+ */
+ List<String> getParentRepairStatus(int cmd);
+
/**
* Triggers stop native transport of the Cassandra node
*/
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java b/server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java
index 9e9943eb..e5d33a62 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java
@@ -66,6 +66,7 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
public static final Permission MOVE_NODE = new DomainAwarePermission("NODE:MOVE", OPERATION_SCOPE);
+ public static final Permission REPAIR = new StandardPermission("REPAIR", OPERATION_SCOPE);
public static final Permission DRAIN_NODE = new DomainAwarePermission("NODE:DRAIN", OPERATION_SCOPE);
// Permissions related to Schema Reporting
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/RepairJobsConfiguration.java b/server/src/main/java/org/apache/cassandra/sidecar/config/RepairJobsConfiguration.java
new file mode 100644
index 00000000..b9829365
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/config/RepairJobsConfiguration.java
@@ -0,0 +1,37 @@
+/*
+ * 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.config;
+
+import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+
+/**
+ * Configuration for Repair jobs
+ */
+public interface RepairJobsConfiguration
+{
+ /**
+ * @return the max retry attempts for the repair job status to be valid
+ */
+ int repairStatusMaxAttempts();
+
+ /**
+ * @return the polling interval (in milliseconds) that checks for the status of the repair job
+ */
+ MillisecondBoundConfiguration repairPollInterval();
+}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/ServiceConfiguration.java b/server/src/main/java/org/apache/cassandra/sidecar/config/ServiceConfiguration.java
index 99e306e8..9f1cf6f5 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/config/ServiceConfiguration.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/config/ServiceConfiguration.java
@@ -120,6 +120,11 @@ public interface ServiceConfiguration
*/
ThrottleConfiguration throttleConfiguration();
+ /**
+ * @return the configuration for repair
+ */
+ RepairJobsConfiguration repairConfiguration();
+
/**
* @return the configuration for SSTable component uploads on this service
*/
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/RepairJobsConfigurationImpl.java b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/RepairJobsConfigurationImpl.java
new file mode 100644
index 00000000..36179dd9
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/RepairJobsConfigurationImpl.java
@@ -0,0 +1,73 @@
+/*
+ * 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.config.yaml;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.config.RepairJobsConfiguration;
+
+/**
+ * Configuration for Repair jobs
+ */
+public class RepairJobsConfigurationImpl implements RepairJobsConfiguration
+{
+ public static final int DEFAULT_VALID_REPAIR_STATUS_ATTEMPTS = 5;
+ public static final MillisecondBoundConfiguration DEFAULT_REPAIR_STATUS_POLLING_INTERVAL = MillisecondBoundConfiguration.parse("2s");
+
+ @JsonProperty(value = "status_attempts")
+ protected final int validRepairStatusAttempts;
+
+ @JsonProperty(value = "status_polling_interval")
+ protected final MillisecondBoundConfiguration repairStatusPollInterval;
+
+ /**
+ * Default constructor that sets default values
+ */
+ public RepairJobsConfigurationImpl()
+ {
+ this(DEFAULT_VALID_REPAIR_STATUS_ATTEMPTS, DEFAULT_REPAIR_STATUS_POLLING_INTERVAL);
+ }
+
+ /**
+ * Constructor with parameters for JSON deserialization
+ *
+ * @param validRepairStatusAttempts the max retry attempts for the repair job status to be valid
+ * @param repairStatusPollInterval the polling interval for repair job status
+ */
+ @JsonCreator
+ public RepairJobsConfigurationImpl(@JsonProperty(value = "status_attempts") int validRepairStatusAttempts,
+ @JsonProperty(value = "status_polling_interval") MillisecondBoundConfiguration repairStatusPollInterval)
+ {
+ this.validRepairStatusAttempts = validRepairStatusAttempts;
+ this.repairStatusPollInterval = repairStatusPollInterval;
+ }
+
+ @Override
+ public int repairStatusMaxAttempts()
+ {
+ return validRepairStatusAttempts;
+ }
+
+ @Override
+ public MillisecondBoundConfiguration repairPollInterval()
+ {
+ return repairStatusPollInterval;
+ }
+}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/ServiceConfigurationImpl.java b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/ServiceConfigurationImpl.java
index a924d100..a0b9c221 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/ServiceConfigurationImpl.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/ServiceConfigurationImpl.java
@@ -35,6 +35,7 @@ import org.apache.cassandra.sidecar.common.server.utils.MinuteBoundConfiguration
import org.apache.cassandra.sidecar.config.CdcConfiguration;
import org.apache.cassandra.sidecar.config.CoordinationConfiguration;
import org.apache.cassandra.sidecar.config.JmxConfiguration;
+import org.apache.cassandra.sidecar.config.RepairJobsConfiguration;
import org.apache.cassandra.sidecar.config.SSTableImportConfiguration;
import org.apache.cassandra.sidecar.config.SSTableSnapshotConfiguration;
import org.apache.cassandra.sidecar.config.SSTableUploadConfiguration;
@@ -68,11 +69,13 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
public static final MinuteBoundConfiguration DEFAULT_ALLOWABLE_TIME_SKEW = MinuteBoundConfiguration.parse("1h");
private static final String SERVER_VERTICLE_INSTANCES_PROPERTY = "server_verticle_instances";
private static final String OPERATIONAL_JOB_TRACKER_SIZE_PROPERTY = "operations_job_tracker_size";
- private static final String OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME_PROPERTY = "operations_job_sync_response_timeout";
+ public static final String OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME_PROPERTY = "operations_job_execution_max_wait_time";
private static final int DEFAULT_SERVER_VERTICLE_INSTANCES = 1;
private static final int DEFAULT_OPERATIONAL_JOB_TRACKER_SIZE = 64;
private static final MillisecondBoundConfiguration DEFAULT_OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME =
MillisecondBoundConfiguration.parse("5s");
+ private static final MillisecondBoundConfiguration MAX_OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME =
+ MillisecondBoundConfiguration.parse("1m");
public static final String THROTTLE_PROPERTY = "throttle";
public static final String SSTABLE_UPLOAD_PROPERTY = "sstable_upload";
public static final String SSTABLE_IMPORT_PROPERTY = "sstable_import";
@@ -80,6 +83,7 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
public static final String WORKER_POOLS_PROPERTY = "worker_pools";
private static final String JMX_PROPERTY = "jmx";
private static final String TRAFFIC_SHAPING_PROPERTY = "traffic_shaping";
+ private static final String REPAIR_PROPERTY = "repair";
private static final String SCHEMA = "schema";
private static final String CDC = "cdc";
private static final String COORDINATION = "coordination";
@@ -119,8 +123,7 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
@JsonProperty(value = OPERATIONAL_JOB_TRACKER_SIZE_PROPERTY, defaultValue = DEFAULT_OPERATIONAL_JOB_TRACKER_SIZE + "")
protected final int operationalJobTrackerSize;
- @JsonProperty(value = OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME_PROPERTY)
- protected final MillisecondBoundConfiguration operationalJobExecutionMaxWaitTime;
+ protected MillisecondBoundConfiguration operationalJobExecutionMaxWaitTime;
@JsonProperty(value = THROTTLE_PROPERTY)
protected final ThrottleConfiguration throttleConfiguration;
@@ -143,6 +146,9 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
@JsonProperty(value = TRAFFIC_SHAPING_PROPERTY)
protected final TrafficShapingConfiguration trafficShapingConfiguration;
+ @JsonProperty(value = REPAIR_PROPERTY)
+ protected final RepairJobsConfiguration repairJobsConfiguration;
+
@JsonProperty(value = SCHEMA)
protected final SchemaKeyspaceConfiguration schemaKeyspaceConfiguration;
@@ -180,6 +186,7 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
serverVerticleInstances = builder.serverVerticleInstances;
operationalJobTrackerSize = builder.operationalJobTrackerSize;
operationalJobExecutionMaxWaitTime = builder.operationalJobExecutionMaxWaitTime;
+ repairJobsConfiguration = builder.repairJobsConfiguration;
throttleConfiguration = builder.throttleConfiguration;
sstableUploadConfiguration = builder.sstableUploadConfiguration;
sstableImportConfiguration = builder.sstableImportConfiguration;
@@ -352,12 +359,24 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
* {@inheritDoc}
*/
@Override
- @JsonProperty(value = OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME_PROPERTY)
public MillisecondBoundConfiguration operationalJobExecutionMaxWaitTime()
{
return operationalJobExecutionMaxWaitTime;
}
+ @JsonProperty(value = OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME_PROPERTY)
+ public void setOperationalJobExecutionMaxWaitTime(MillisecondBoundConfiguration operationalJobExecutionMaxWaitTime)
+ {
+ if (operationalJobExecutionMaxWaitTime.compareTo(MAX_OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME) > 0)
+ {
+ throw new ConfigurationException(String.format("Invalid %s value (%s). The maximum allowed value is %s.",
+ OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME_PROPERTY,
+ operationalJobExecutionMaxWaitTime,
+ MAX_OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME));
+ }
+ this.operationalJobExecutionMaxWaitTime = operationalJobExecutionMaxWaitTime;
+ }
+
/**
* {@inheritDoc}
*/
@@ -428,6 +447,16 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
return trafficShapingConfiguration;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @JsonProperty(value = REPAIR_PROPERTY)
+ public RepairJobsConfiguration repairConfiguration()
+ {
+ return repairJobsConfiguration;
+ }
+
/**
* {@inheritDoc}
*/
@@ -483,6 +512,7 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
protected int serverVerticleInstances = DEFAULT_SERVER_VERTICLE_INSTANCES;
protected int operationalJobTrackerSize = DEFAULT_OPERATIONAL_JOB_TRACKER_SIZE;
protected MillisecondBoundConfiguration operationalJobExecutionMaxWaitTime = DEFAULT_OPERATIONAL_JOB_EXECUTION_MAX_WAIT_TIME;
+ protected RepairJobsConfiguration repairJobsConfiguration = new RepairJobsConfigurationImpl();
protected ThrottleConfiguration throttleConfiguration = new ThrottleConfigurationImpl();
protected SSTableUploadConfiguration sstableUploadConfiguration = new SSTableUploadConfigurationImpl();
protected SSTableImportConfiguration sstableImportConfiguration = new SSTableImportConfigurationImpl();
@@ -708,6 +738,18 @@ public class ServiceConfigurationImpl implements ServiceConfiguration
return update(b -> b.schemaKeyspaceConfiguration = schemaKeyspaceConfiguration);
}
+ /**
+ * Sets the {@code repairConfiguration} and returns a reference to this Builder enabling method
+ * chaining.
+ *
+ * @param repairJobsConfiguration the {@code repairConfiguration} to set
+ * @return a reference to this Builder
+ */
+ public Builder repairConfiguration(RepairJobsConfiguration repairJobsConfiguration)
+ {
+ return update(b -> b.repairJobsConfiguration = repairJobsConfiguration);
+ }
+
/**
* Set the {@code cdcConfiguration} and returns a reference to this Builder enabling
* method chaining.
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/handlers/RepairHandler.java b/server/src/main/java/org/apache/cassandra/sidecar/handlers/RepairHandler.java
new file mode 100644
index 00000000..17b68491
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/handlers/RepairHandler.java
@@ -0,0 +1,179 @@
+/*
+ * 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.handlers;
+
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.Set;
+
+import com.datastax.driver.core.utils.UUIDs;
+import com.google.inject.Inject;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.json.DecodeException;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import io.vertx.ext.web.handler.HttpException;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.data.Name;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.config.ServiceConfiguration;
+import org.apache.cassandra.sidecar.handlers.data.RepairRequestParam;
+import org.apache.cassandra.sidecar.job.OperationalJobManager;
+import org.apache.cassandra.sidecar.job.RepairJob;
+import org.apache.cassandra.sidecar.tasks.PeriodicTaskExecutor;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.apache.cassandra.sidecar.utils.OperationalJobUtils;
+import org.jetbrains.annotations.NotNull;
+
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for triggering repair
+ */
+public class RepairHandler extends AbstractHandler<RepairRequestParam> implements AccessProtected
+{
+ private final ServiceConfiguration config;
+ private final OperationalJobManager jobManager;
+ private final PeriodicTaskExecutor periodicTaskExecutor;
+
+ /**
+ * Constructs a handler with the provided {@code metadataFetcher}
+ *
+ * @param metadataFetcher the metadata fetcher
+ * @param executorPools executor pools for blocking executions
+ * @param serviceConfiguration configuration object holding config details of Sidecar
+ * @param validator a validator instance to validate Cassandra-specific input
+ * @param jobManager manager for long-running operational jobs
+ * @param periodicTaskExecutor executor for periodic tasks
+ */
+ @Inject
+ protected RepairHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ ServiceConfiguration serviceConfiguration,
+ CassandraInputValidator validator,
+ OperationalJobManager jobManager,
+ PeriodicTaskExecutor periodicTaskExecutor)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ this.config = serviceConfiguration;
+ this.periodicTaskExecutor = periodicTaskExecutor;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected RepairRequestParam extractParamsOrThrow(RoutingContext context)
+ {
+ Name keyspace = keyspace(context, true);
+ if (keyspace == null)
+ {
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST, "'keyspace' is required but not supplied");
+ }
+
+ RepairPayload payload;
+ try
+ {
+ payload = context.body().asPojo(RepairPayload.class);
+ }
+ catch (DecodeException decodeException)
+ {
+ logger.warn("Bad request to create repair job. Received invalid JSON payload.");
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST,
+ "Invalid request payload",
+ decodeException);
+ }
+ if (payload == null)
+ {
+ logger.warn("Bad request to create repair job. Received null payload.");
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST, "Unexpected null payload for request");
+ }
+
+ // Validate token range if provided
+ validateTokenRange(payload);
+
+ return RepairRequestParam.from(keyspace, payload);
+ }
+
+ @Override
+ protected void handleInternal(RoutingContext context,
+ HttpServerRequest httpRequest,
+ @NotNull String host,
+ SocketAddress remoteAddress,
+ RepairRequestParam repairRequestParam)
+ {
+ StorageOperations operations = metadataFetcher.delegate(host).storageOperations();
+ RepairJob job = new RepairJob(periodicTaskExecutor, config.repairConfiguration(), UUIDs.timeBased(), operations, repairRequestParam);
+
+ jobManager.trySubmitJob(job,
+ (completedJob, exception) ->
+ OperationalJobUtils.sendStatusBasedResponse(context, completedJob, exception),
+ executorPools.service(),
+ config.operationalJobExecutionMaxWaitTime());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Set<Authorization> requiredAuthorizations()
+ {
+ return Collections.singleton(BasicPermissions.REPAIR.toAuthorization());
+ }
+
+ /**
+ * Validates the token range in the repair payload if both start and end tokens are provided.
+ *
+ * @param payload the repair payload to validate
+ * @throws HttpException if the token range is invalid
+ */
+ private void validateTokenRange(RepairPayload payload)
+ {
+ if (payload.startToken() != null && payload.endToken() != null)
+ {
+ try
+ {
+ String startTokenStr = payload.startToken();
+ String endTokenStr = payload.endToken();
+
+ // Validate tokens using BigInteger for proper numeric comparison
+ BigInteger startToken = new BigInteger(startTokenStr);
+ BigInteger endToken = new BigInteger(endTokenStr);
+
+ if (startToken.compareTo(endToken) >= 0)
+ {
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST,
+ "Start token must be less than end token. Got start: " +
+ startTokenStr + ", end: " + endTokenStr);
+ }
+ }
+ catch (NumberFormatException e)
+ {
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST,
+ "Invalid token format. Tokens must be numeric values.", e);
+ }
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/handlers/data/RepairRequestParam.java b/server/src/main/java/org/apache/cassandra/sidecar/handlers/data/RepairRequestParam.java
new file mode 100644
index 00000000..e5c1fbe7
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/handlers/data/RepairRequestParam.java
@@ -0,0 +1,59 @@
+/*
+ * 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.handlers.data;
+
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
+import org.apache.cassandra.sidecar.common.server.data.Name;
+
+/**
+ * Holder class for the {@link org.apache.cassandra.sidecar.handlers.RepairHandler}
+ * request parameters
+ */
+public class RepairRequestParam
+{
+ private final Name keyspace;
+ private final RepairPayload requestPayload;
+
+ private RepairRequestParam(Name keyspace, RepairPayload requestPayload)
+ {
+ this.keyspace = keyspace;
+ this.requestPayload = requestPayload;
+ }
+
+ public static RepairRequestParam from(Name keyspace, RepairPayload requestPayload)
+ {
+ return new RepairRequestParam(keyspace, requestPayload);
+ }
+
+ /**
+ * @return the keyspace in Cassandra
+ */
+ public Name keyspace()
+ {
+ return keyspace;
+ }
+
+ /**
+ * @return the Repair request payload
+ */
+ public RepairPayload requestPayload()
+ {
+ return requestPayload;
+ }
+}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/handlers/validations/ValidateTableExistenceHandler.java b/server/src/main/java/org/apache/cassandra/sidecar/handlers/validations/ValidateTableExistenceHandler.java
index 4c089481..24a3ec9d 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/handlers/validations/ValidateTableExistenceHandler.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/handlers/validations/ValidateTableExistenceHandler.java
@@ -91,6 +91,7 @@ public class ValidateTableExistenceHandler extends AbstractHandler<QualifiedTabl
String table = input.maybeQuotedTableName();
if (table == null)
{
+ // Table is not required, so skip table validation
context.next();
return;
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDecommissionJob.java b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDecommissionJob.java
index 32a05fa9..778ea450 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDecommissionJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDecommissionJob.java
@@ -18,13 +18,16 @@
package org.apache.cassandra.sidecar.job;
+import java.util.List;
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.vertx.core.Future;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.jetbrains.annotations.NotNull;
/**
* Implementation of {@link OperationalJob} to perform node decommission operation.
@@ -43,9 +46,20 @@ public class NodeDecommissionJob extends OperationalJob
this.isForce = isForce;
}
+ /**
+ * Node Decommission job determines conflict based on the current operation mode of the cluster.
+ * If the cluster is in the middle of a decommission or leaving operation, the job will be rejected.
+ * @param sameOperationJobs jobs tracked by the tracker are not relevant for this conflict check
+ * @return true if the job has a conflict, false otherwise
+ */
@Override
- public boolean isRunningOnCassandra()
+ public boolean hasConflict(@NotNull List<OperationalJob> sameOperationJobs)
{
+ if (!sameOperationJobs.isEmpty())
+ {
+ return true;
+ }
+
String operationMode = storageOperations.operationMode();
return "LEAVING".equals(operationMode) || "DECOMMISSIONED".equals(operationMode);
}
@@ -76,16 +90,11 @@ public class NodeDecommissionJob extends OperationalJob
* {@inheritDoc}
*/
@Override
- protected void executeInternal()
+ protected Future<Void> executeInternal()
{
- if (isRunningOnCassandra())
- {
- LOGGER.info("Not executing job as an ongoing or completed decommission operation was found jobId={}", this.jobId());
- return;
- }
-
LOGGER.info("Executing decommission operation. jobId={}", this.jobId());
storageOperations.decommission(isForce);
+ return Future.succeededFuture();
}
/**
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDrainJob.java b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDrainJob.java
index d6485764..16fe02e2 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDrainJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeDrainJob.java
@@ -19,15 +19,18 @@
package org.apache.cassandra.sidecar.job;
import java.io.IOException;
+import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.vertx.core.Future;
import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
import org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException;
+import org.jetbrains.annotations.NotNull;
/**
* Implementation of {@link OperationalJob} to perform node drain operation.
@@ -84,7 +87,7 @@ public class NodeDrainJob extends OperationalJob
* {@inheritDoc}
*/
@Override
- public boolean isRunningOnCassandra()
+ public boolean hasConflict(@NotNull List<OperationalJob> sameOperationJobs)
{
String operationMode = storageOperations.operationMode();
NodeDrainStateEnum nodeDrainStateEnum = NodeDrainStateEnum.fromOperationMode(operationMode);
@@ -111,14 +114,8 @@ public class NodeDrainJob extends OperationalJob
* {@inheritDoc}
*/
@Override
- protected void executeInternal()
+ protected Future<Void> executeInternal()
{
- if (isRunningOnCassandra())
- {
- LOGGER.info("Not executing job as an ongoing drain operation was found jobId={}", this.jobId());
- return;
- }
-
LOGGER.info("Executing drain operation. jobId={}", this.jobId());
try
{
@@ -129,6 +126,8 @@ public class NodeDrainJob extends OperationalJob
LOGGER.error("Job execution failed. jobId={} reason='{}'", this.jobId(), ex.getMessage(), ex);
throw OperationalJobException.wraps(ex);
}
+
+ return Future.succeededFuture();
}
/**
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeMoveJob.java b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeMoveJob.java
index f03d6338..d0690095 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/NodeMoveJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/NodeMoveJob.java
@@ -18,12 +18,17 @@
package org.apache.cassandra.sidecar.job;
+import java.io.IOException;
+import java.util.List;
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.vertx.core.Future;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException;
+import org.jetbrains.annotations.NotNull;
/**
* Implementation of {@link OperationalJob} to perform node move operation.
@@ -44,7 +49,7 @@ public class NodeMoveJob extends OperationalJob
}
@Override
- public boolean isRunningOnCassandra()
+ public boolean hasConflict(@NotNull List<OperationalJob> sameOperationJobs)
{
String operationMode = storageOperations.operationMode();
return OPERATION_MODE_MOVING.equals(operationMode);
@@ -52,12 +57,24 @@ public class NodeMoveJob extends OperationalJob
/**
* {@inheritDoc}
+ *
+ * @return
*/
@Override
- protected void executeInternal() throws Exception
+ protected Future<Void> executeInternal() throws Exception
{
- LOGGER.info("Executing move operation. jobId={} newToken={}", this.jobId(), newToken);
- storageOperations.move(newToken);
+ try
+ {
+ LOGGER.info("Executing move operation. jobId={} newToken={}", this.jobId(), newToken);
+ storageOperations.move(newToken);
+ }
+ catch (IOException ex)
+ {
+ LOGGER.error("Job execution failed. jobId={} reason='{}'", this.jobId(), ex.getMessage(), ex);
+ throw OperationalJobException.wraps(ex);
+ }
+
+ return Future.succeededFuture();
}
/**
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java
index 04fa5d0c..660f7db0 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.sidecar.job;
+import java.util.List;
import java.util.UUID;
import org.slf4j.Logger;
@@ -32,6 +33,7 @@ import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
import org.apache.cassandra.sidecar.common.utils.Preconditions;
import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
import org.apache.cassandra.sidecar.tasks.Task;
+import org.jetbrains.annotations.NotNull;
/**
* An abstract class representing operational jobs that run on Cassandra
@@ -100,11 +102,13 @@ public abstract class OperationalJob implements Task<Void>
}
/**
- * The concrete-job-specific implementation to determine if the job is running on the Cassandra node.
- * @return true if the job is running on the Cassandra node. For example, node decommission is tracked by the
- * operationMode exposed from Cassandra.
+ * Validates if the job has conflict either leveraging either other same operation jobs on the same node and/or a
+ * custom job-specific implementation for conflict determination.
+ *
+ * @param sameOperationJobs list of jobs being tracked by the tracker that have the same operation
+ * @return true if the job is has a conflict.
*/
- public abstract boolean isRunningOnCassandra();
+ public abstract boolean hasConflict(@NotNull List<OperationalJob> sameOperationJobs);
/**
* Determines the status of the job. OperationalJob subclasses could choose to override the method.
@@ -187,9 +191,9 @@ public abstract class OperationalJob implements Task<Void>
}
/**
- * OperationalJob body. The implementation is executed in a blocking manner.
+ * OperationalJob body. The implementation returns a Future representing the job execution.
*/
- protected abstract void executeInternal() throws Exception;
+ protected abstract Future<Void> executeInternal() throws Exception;
/**
* Execute the job behavior as specified in the internal execution {@link #executeInternal()},
@@ -203,18 +207,24 @@ public abstract class OperationalJob implements Task<Void>
promise.future().onComplete(executionPromise);
try
{
- // Blocking call to perform concrete job-specific execution, returning the status
- executeInternal();
- promise.tryComplete();
- if (LOGGER.isDebugEnabled())
- {
- LOGGER.debug("Complete job execution. jobId={} status={}", jobId, status());
- }
+ Future<Void> internalFuture = executeInternal();
+ internalFuture.onComplete(ar -> {
+ if (ar.succeeded())
+ {
+ promise.tryComplete();
+ LOGGER.info("Complete job execution. jobId={} status={}", jobId, status());
+ }
+ else
+ {
+ promise.tryFail(ar.cause());
+ LOGGER.error("Job execution failed. jobId={} reason={}", jobId, ar.cause().getMessage());
+ }
+ });
}
catch (Throwable e)
{
OperationalJobException oje = OperationalJobException.wraps(e);
- LOGGER.error("Job execution failed. jobId={} reason='{}'", jobId, oje.getMessage(), oje);
+ LOGGER.error("Job execution failed. jobId={} reason={}", jobId, oje.getMessage(), oje);
promise.tryFail(oje);
}
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java
index 6d36f353..0e70e1e6 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java
@@ -27,7 +27,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.inject.Inject;
-
import com.google.inject.Singleton;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
@@ -126,9 +125,8 @@ public class OperationalJobManager
*/
private void checkConflict(OperationalJob job) throws OperationalJobConflictException
{
- // If there are no tracked running jobs for same operation, then we confirm downstream
- // Downstream check is done in most cases - by design
- if (!jobTracker.inflightJobsByOperation(job.name()).isEmpty() || job.isRunningOnCassandra())
+ List<OperationalJob> sameOperationJobs = jobTracker.inflightJobsByOperation(job.name());
+ if (job.hasConflict(sameOperationJobs))
{
throw new OperationalJobConflictException("The same operational job is already running on Cassandra. operationName='" + job.name() + '\'');
}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/job/RepairJob.java b/server/src/main/java/org/apache/cassandra/sidecar/job/RepairJob.java
new file mode 100644
index 00000000..a9fe4ccb
--- /dev/null
+++ b/server/src/main/java/org/apache/cassandra/sidecar/job/RepairJob.java
@@ -0,0 +1,437 @@
+/*
+ * 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.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import org.apache.cassandra.sidecar.adapters.base.RepairOptions;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
+import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.config.RepairJobsConfiguration;
+import org.apache.cassandra.sidecar.handlers.data.RepairRequestParam;
+import org.apache.cassandra.sidecar.tasks.PeriodicTask;
+import org.apache.cassandra.sidecar.tasks.PeriodicTaskExecutor;
+import org.apache.cassandra.sidecar.tasks.ScheduleDecision;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Implementation of {@link OperationalJob} to perform repair operation.
+ * <p>
+ * Cassandra repair operations are asynchronous and can run for extended periods. When a repair is
+ * detected to be IN_PROGRESS, this job sets its status to RUNNING and completes its promise.
+ * <p>
+ * The job's status can be checked at any time using the {@link #status()} method, which directly
+ * queries the current repair status from Cassandra, regardless of whether the promise has been completed.
+ * <p>
+ * This design ensures timely responses to clients while allowing accurate status reporting for
+ * long-running repair operations.
+ */
+public class RepairJob extends OperationalJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(RepairJob.class);
+ private static final String OPERATION = "repair";
+ private static final String PREVIEW_KIND_REPAIRED = "REPAIRED";
+
+ private final RepairRequestParam repairParams;
+ private final RepairJobsConfiguration config;
+ private final PeriodicTaskExecutor periodicTaskExecutor;
+ protected StorageOperations storageOperations;
+ private volatile OperationalJobStatus currentStatus;
+ private volatile int commandId = -1; // Store the command ID for status checks
+
+ /**
+ * Enum representing the status of a parent repair session
+ */
+ public enum ParentRepairStatus
+ {
+ IN_PROGRESS, COMPLETED, FAILED
+ }
+
+ /**
+ * Container for status parsing results
+ */
+ private static class StatusResult
+ {
+ final OperationalJobStatus operationalStatus;
+ final List<String> messages;
+
+ StatusResult(OperationalJobStatus operationalStatus, List<String> messages)
+ {
+ this.operationalStatus = operationalStatus;
+ this.messages = messages;
+ }
+ }
+
+ /**
+ * Constructor for creation of RepairJob
+ *
+ * @param periodicTaskExecutor PeriodicTaskExecutor for status checking
+ * @param config Repair job configuration
+ * @param jobId UUID representing the Job to be created
+ * @param storageOps Reference to the storage operations interface
+ * @param repairParams Repair request parameters
+ */
+ public RepairJob(PeriodicTaskExecutor periodicTaskExecutor,
+ RepairJobsConfiguration config,
+ UUID jobId,
+ StorageOperations storageOps,
+ RepairRequestParam repairParams)
+ {
+ super(jobId);
+ this.periodicTaskExecutor = periodicTaskExecutor;
+ this.config = config;
+ this.storageOperations = storageOps;
+ this.repairParams = repairParams;
+ this.currentStatus = OperationalJobStatus.CREATED;
+ }
+
+ /**
+ * {@inheritDoc}
+ * RepairJob allows parallel executions since multiple repairs can run concurrently
+ * on different tables or with different parameters.
+ */
+ @Override
+ public boolean hasConflict(@NotNull List<OperationalJob> sameOperationJobs)
+ {
+ // Let Cassandra handle conflict detection - no pre-emptive conflict checking
+ // Conflicts will be handled through exception handling during repair execution
+ return false;
+ }
+
+ @Override
+ protected Future<Void> executeInternal()
+ {
+ try
+ {
+ Map<String, String> options = generateRepairOptions(repairParams.requestPayload());
+ String keyspace = repairParams.keyspace().maybeQuotedName();
+
+ LOGGER.info("Executing repair operation for keyspace='{}' jobId={}", keyspace, this.jobId());
+
+ try
+ {
+ commandId = storageOperations.repairAsync(keyspace, options);
+ }
+ catch (Exception e)
+ {
+ LOGGER.error("Failed to initiate repair for keyspace='{}' jobId={}", keyspace, this.jobId(), e);
+ currentStatus = OperationalJobStatus.FAILED;
+ return Future.failedFuture(e);
+ }
+
+ if (commandId <= 0)
+ {
+ // When there are no relevant token ranges for the keyspace or RF is 1, the repair is inapplicable
+ LOGGER.info("Replication factor is 1. No repair is needed for keyspace='{}' jobId={}", keyspace, this.jobId());
+ currentStatus = OperationalJobStatus.SUCCEEDED;
+ return Future.succeededFuture();
+ }
+
+ Promise<Void> jobSubmissionPromise = Promise.promise();
+ int maxAttempts = config.repairStatusMaxAttempts();
+ AtomicInteger attemptCounter = new AtomicInteger(0);
+
+ // Use PeriodicTaskExecutor for proper scheduling with delay between executions
+ RepairStatusCheckTask statusCheckTask = new RepairStatusCheckTask(keyspace, jobSubmissionPromise, attemptCounter, maxAttempts);
+ periodicTaskExecutor.schedule(statusCheckTask);
+
+ Future<Void> jobSubmissionFuture = jobSubmissionPromise.future();
+ jobSubmissionFuture.onComplete(ignore -> periodicTaskExecutor.unschedule(statusCheckTask));
+ return jobSubmissionFuture;
+ }
+ catch (Exception e)
+ {
+ // Catch any exceptions in the overall method
+ LOGGER.error("Failed to execute repair job jobId={}", this.jobId(), e);
+ currentStatus = OperationalJobStatus.FAILED;
+ return Future.failedFuture(e);
+ }
+ }
+
+ @Override
+ public OperationalJobStatus status()
+ {
+ refreshStatusFromCassandra();
+ return currentStatus != null ? currentStatus : super.status();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public String name()
+ {
+ return OPERATION;
+ }
+
+ /**
+ * Refreshes the repair status from Cassandra and updates the internal currentStatus.
+ *
+ * @return StatusResult containing the operational status and messages, or null if no status is available yet
+ */
+ private StatusResult refreshStatusFromCassandra()
+ {
+ // If we don't have a valid command ID, return current status
+ if (commandId <= 0)
+ {
+ return null;
+ }
+
+ List<String> status;
+ try
+ {
+ status = storageOperations.getParentRepairStatus(commandId);
+ }
+ catch (Exception e)
+ {
+ LOGGER.warn("Failed to get repair status for cmd: {}", commandId, e);
+ return null;
+ }
+
+ // If no status available yet, return null
+ if (status == null || status.isEmpty())
+ {
+ return null;
+ }
+
+ try
+ {
+ ParentRepairStatus parentRepairStatus = ParentRepairStatus.valueOf(status.get(0));
+
+ switch (parentRepairStatus)
+ {
+ case COMPLETED:
+ currentStatus = OperationalJobStatus.SUCCEEDED;
+ break;
+ case FAILED:
+ currentStatus = OperationalJobStatus.FAILED;
+ break;
+ case IN_PROGRESS:
+ currentStatus = OperationalJobStatus.RUNNING;
+ break;
+ default:
+ LOGGER.warn("Encountered unexpected repair status: {}", parentRepairStatus);
+ return null;
+ }
+
+
+ // Extract messages (everything after the first element)
+ List<String> messages = status.subList(1, status.size());
+
+ return new StatusResult(currentStatus, messages);
+ }
+ catch (IllegalArgumentException e)
+ {
+ LOGGER.warn("Invalid parent repair status={}", status.get(0), e);
+ return null;
+ }
+ }
+
+ /**
+ * Handles the job submission promise based on the repair status result.
+ *
+ * @param jobSubmissionPromise the promise to complete or fail
+ * @param statusResult the status result containing operational status and messages
+ */
+ private void handleJobSubmissionPromise(Promise<Void> jobSubmissionPromise, StatusResult statusResult)
+ {
+ LOGGER.info("Parent repair session has {} status. Messages: {}",
+ statusResult.operationalStatus, String.join("\n", statusResult.messages));
+
+ switch (statusResult.operationalStatus)
+ {
+ case SUCCEEDED:
+ case RUNNING:
+ jobSubmissionPromise.tryComplete();
+ break;
+ case FAILED:
+ String reason = !statusResult.messages.isEmpty() ? statusResult.messages.get(0) :
+ "Repair failed with no error message";
+ jobSubmissionPromise.tryFail(new IOException(reason));
+ break;
+ default:
+ String message = String.format("Encountered unexpected repair status: %s Messages: %s",
+ statusResult.operationalStatus, String.join("\n", statusResult.messages));
+ LOGGER.error(message);
+ jobSubmissionPromise.tryFail(new IOException(message));
+ break;
+ }
+ }
+
+ private Map<String, String> generateRepairOptions(RepairPayload repairPayload)
+ {
+ Map<String, String> options = new HashMap<>();
+
+ List<String> tables = repairPayload.tables();
+ if (tables != null && !tables.isEmpty())
+ {
+ options.put(RepairOptions.COLUMN_FAMILIES.optionName(), String.join(",", tables));
+ }
+
+ Boolean isPrimaryRange = repairPayload.isPrimaryRange();
+ if (isPrimaryRange != null)
+ {
+ options.put(RepairOptions.PRIMARY_RANGE.optionName(), String.valueOf(isPrimaryRange));
+ }
+
+ String dc = repairPayload.datacenter();
+ if (dc != null)
+ {
+ options.put(RepairOptions.DATACENTERS.optionName(), dc);
+ }
+
+ List<String> hosts = repairPayload.hosts();
+ if (hosts != null && !hosts.isEmpty())
+ {
+ options.put(RepairOptions.HOSTS.optionName(), String.join(",", hosts));
+ }
+
+ if (repairPayload.startToken() != null && repairPayload.endToken() != null)
+ {
+ // Use original string values for range construction
+ options.put(RepairOptions.RANGES.optionName(), repairPayload.startToken() + ":" + repairPayload.endToken());
+ }
+
+ if (repairPayload.repairType() == RepairPayload.RepairType.INCREMENTAL)
+ {
+ options.put(RepairOptions.INCREMENTAL.optionName(), Boolean.TRUE.toString());
+ }
+
+ if (Boolean.TRUE.equals(repairPayload.force()))
+ {
+ options.put(RepairOptions.FORCE_REPAIR.optionName(), Boolean.TRUE.toString());
+ }
+
+ if (Boolean.TRUE.equals(repairPayload.shouldValidate()))
+ {
+ options.put(RepairOptions.PREVIEW.optionName(), PREVIEW_KIND_REPAIRED);
+ }
+ return options;
+ }
+
+ /**
+ * PeriodicTask implementation for checking repair status with proper scheduling.
+ * This task ensures delay between executions and prevents overlapping status checks.
+ */
+ private class RepairStatusCheckTask implements PeriodicTask
+ {
+ private final String keyspace;
+ private final Promise<Void> jobSubmissionPromise;
+ private final AtomicInteger attemptCounter;
+ private final int maxAttempts;
+ private volatile boolean completed = false;
+
+ RepairStatusCheckTask(String keyspace, Promise<Void> jobSubmissionPromise, AtomicInteger attemptCounter, int maxAttempts)
+ {
+ this.keyspace = keyspace;
+ this.jobSubmissionPromise = jobSubmissionPromise;
+ this.attemptCounter = attemptCounter;
+ this.maxAttempts = maxAttempts;
+ }
+
+ @Override
+ public void execute(Promise<Void> promise)
+ {
+ try
+ {
+ int currentAttempt = attemptCounter.incrementAndGet();
+ if (currentAttempt > maxAttempts)
+ {
+ LOGGER.warn("Failed to obtain repair status after {} attempts for keyspace='{}' jobId={}",
+ maxAttempts, keyspace, jobId());
+ // Set status to RUNNING and complete the promise to ensure the job is properly handled
+ currentStatus = OperationalJobStatus.RUNNING;
+ jobSubmissionPromise.tryComplete();
+ completed = true;
+ promise.complete();
+ return;
+ }
+
+ StatusResult statusResult = refreshStatusFromCassandra();
+
+ // If no status available yet, continue polling
+ if (statusResult == null)
+ {
+ LOGGER.debug("No parent repair session status found for cmd: {} - repair may be initializing (attempt {}/{})" +
+ " for keyspace='{}' jobId={}", commandId, currentAttempt, maxAttempts, keyspace, jobId());
+ promise.complete(); // Continue to next execution
+ return;
+ }
+
+ // We have a definitive status, handle the promise and stop the task
+ handleJobSubmissionPromise(jobSubmissionPromise, statusResult);
+ completed = true;
+ promise.complete();
+ }
+ catch (Exception e)
+ {
+ LOGGER.error("Unexpected error in repair status check for keyspace='{}' jobId={}", keyspace, jobId(), e);
+ currentStatus = OperationalJobStatus.FAILED;
+ jobSubmissionPromise.tryFail(e);
+ completed = true;
+ promise.fail(e);
+ }
+ }
+
+ @Override
+ public DurationSpec delay()
+ {
+ return config.repairPollInterval();
+ }
+
+ @Override
+ public DurationSpec initialDelay()
+ {
+ // Start immediately for the first check
+ return MillisecondBoundConfiguration.ZERO;
+ }
+
+ @Override
+ public ScheduleDecision scheduleDecision()
+ {
+ // Stop scheduling if we've completed
+ return completed ? ScheduleDecision.SKIP : ScheduleDecision.EXECUTE;
+ }
+
+ @Override
+ public String identifier()
+ {
+ return "repair-status-check-" + jobId();
+ }
+
+ @Override
+ public String name()
+ {
+ return "RepairStatusCheck for job=" + jobId();
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java b/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java
index b1588b11..cc315449 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java
@@ -52,6 +52,7 @@ import org.apache.cassandra.sidecar.handlers.NodeDecommissionHandler;
import org.apache.cassandra.sidecar.handlers.NodeDrainHandler;
import org.apache.cassandra.sidecar.handlers.NodeMoveHandler;
import org.apache.cassandra.sidecar.handlers.OperationalJobHandler;
+import org.apache.cassandra.sidecar.handlers.RepairHandler;
import org.apache.cassandra.sidecar.handlers.RingHandler;
import org.apache.cassandra.sidecar.handlers.SchemaHandler;
import org.apache.cassandra.sidecar.handlers.StreamStatsHandler;
@@ -231,14 +232,37 @@ public class CassandraOperationsModule extends AbstractModule
.build();
}
+ @GET
+ @Path(ApiEndpointsV1.REPAIR_ROUTE)
+ @Operation(summary = "Trigger a repair operation",
+ description = "Returns the status of the repair operation on the Cassandra node")
+ @APIResponse(description = "Repair operation status retrieved successfully",
+ responseCode = "200",
+ content = @Content(mediaType = "application/json",
+ schema = @Schema(implementation = OperationalJobResponse.class)))
+ @ProvidesIntoMap
+ @KeyClassMapKey(VertxRouteMapKeys.CassandraRepairRouteKey.class)
+ VertxRoute cassandraRepairRoute(RouteBuilder.Factory factory,
+ ValidateTableExistenceHandler validateKeyspaceExistence,
+ RepairHandler repairhandler)
+ {
+ return factory.builderForRoute()
+ .setBodyHandler(true)
+ // We only require validation of keyspace here. It is guaranteed that the URI will provide a
+ // keyspace to validate
+ .handler(validateKeyspaceExistence)
+ .handler(repairhandler)
+ .build();
+ }
+
@GET
@Path(ApiEndpointsV1.STREAM_STATS_ROUTE)
@Operation(summary = "Get stream statistics",
- description = "Returns streaming statistics for the Cassandra node")
+ description = "Returns streaming statistics for the Cassandra node")
@APIResponse(description = "Stream statistics retrieved successfully",
- responseCode = "200",
- content = @Content(mediaType = "application/json",
- schema = @Schema(implementation = StreamStatsResponse.class)))
+ responseCode = "200",
+ content = @Content(mediaType = "application/json",
+ schema = @Schema(implementation = StreamStatsResponse.class)))
@ProvidesIntoMap
@KeyClassMapKey(VertxRouteMapKeys.CassandraStreamStatsRouteKey.class)
VertxRoute cassandraStreamStatsRoute(RouteBuilder.Factory factory,
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/modules/multibindings/VertxRouteMapKeys.java b/server/src/main/java/org/apache/cassandra/sidecar/modules/multibindings/VertxRouteMapKeys.java
index 2e0ce9be..df5a3728 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/modules/multibindings/VertxRouteMapKeys.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/modules/multibindings/VertxRouteMapKeys.java
@@ -109,6 +109,11 @@ public interface VertxRouteMapKeys
HttpMethod HTTP_METHOD = HttpMethod.GET;
String ROUTE_URI = ApiEndpointsV1.OPERATIONAL_JOB_ROUTE;
}
+ interface CassandraRepairRouteKey extends RouteClassKey
+ {
+ HttpMethod HTTP_METHOD = HttpMethod.PUT;
+ String ROUTE_URI = ApiEndpointsV1.REPAIR_ROUTE;
+ }
interface CassandraRingRouteKey extends RouteClassKey
{
HttpMethod HTTP_METHOD = HttpMethod.GET;
diff --git a/server/src/main/java/org/apache/cassandra/sidecar/utils/OperationalJobUtils.java b/server/src/main/java/org/apache/cassandra/sidecar/utils/OperationalJobUtils.java
index e640b94e..0edd5fd1 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/utils/OperationalJobUtils.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/utils/OperationalJobUtils.java
@@ -69,7 +69,11 @@ public class OperationalJobUtils
context.response().setStatusCode(HttpResponseStatus.ACCEPTED.code());
}
- String reason = status == FAILED ? job.asyncResult().cause().getMessage() : null;
+ String reason = null;
+ if (status == FAILED && job.asyncResult() != null && job.asyncResult().cause() != null)
+ {
+ reason = job.asyncResult().cause().getMessage();
+ }
context.json(new OperationalJobResponse(job.jobId(), status, job.name(), reason));
}
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/config/SidecarConfigurationTest.java b/server/src/test/java/org/apache/cassandra/sidecar/config/SidecarConfigurationTest.java
index 11dd5d87..b11f582c 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/config/SidecarConfigurationTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/config/SidecarConfigurationTest.java
@@ -35,6 +35,7 @@ import org.apache.cassandra.sidecar.common.server.dns.DnsResolver;
import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
import org.apache.cassandra.sidecar.config.yaml.MetricsFilteringConfigurationImpl;
import org.apache.cassandra.sidecar.config.yaml.SidecarConfigurationImpl;
+import org.apache.cassandra.sidecar.exceptions.ConfigurationException;
import org.assertj.core.api.Condition;
import static org.apache.cassandra.sidecar.common.ResourceUtils.writeResourceToPath;
@@ -560,6 +561,33 @@ class SidecarConfigurationTest
.containsEntry("localhost1", "localhost4");
}
+ @Test
+ void testInvalidOperationalJobExecutionMaxWaitTime()
+ {
+ String yaml = "sidecar:\n" +
+ " operations_job_execution_max_wait_time: 2m";
+ assertThatExceptionOfType(JsonMappingException.class)
+ .isThrownBy(() -> SidecarConfigurationImpl.fromYamlString(yaml))
+ .withRootCauseInstanceOf(ConfigurationException.class)
+ .withMessageContaining("Invalid operations_job_execution_max_wait_time value (2m). The maximum allowed value is 1m.");
+ }
+
+ @Test
+ void testRepairConfiguration() throws IOException
+ {
+ String yaml = "sidecar:\n" +
+ " repair:\n" +
+ " status_attempts: 2\n" +
+ " status_polling_interval: 500ms";
+ SidecarConfigurationImpl sidecarConfiguration = SidecarConfigurationImpl.fromYamlString(yaml);
+ ServiceConfiguration serviceConfiguration = sidecarConfiguration.serviceConfiguration();
+ assertThat(serviceConfiguration).isNotNull();
+ RepairJobsConfiguration repairConfig = serviceConfiguration.repairConfiguration();
+ assertThat(repairConfig).isNotNull();
+ assertThat(repairConfig.repairStatusMaxAttempts()).isEqualTo(2);
+ assertThat(repairConfig.repairPollInterval()).isEqualTo(MillisecondBoundConfiguration.parse("500ms"));
+ }
+
void validateSingleInstanceSidecarConfiguration(SidecarConfiguration config)
{
assertThat(config.cassandraInstances()).isNotNull().hasSize(1);
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/handlers/CommonTest.java b/server/src/test/java/org/apache/cassandra/sidecar/handlers/CommonTest.java
index 9d0a5c33..3be4d4f6 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/handlers/CommonTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/handlers/CommonTest.java
@@ -18,7 +18,7 @@
package org.apache.cassandra.sidecar.handlers;
-import java.util.Collections;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -40,6 +40,7 @@ import org.apache.cassandra.sidecar.TestModule;
import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
import org.apache.cassandra.sidecar.modules.SidecarModules;
import org.apache.cassandra.sidecar.server.Server;
@@ -51,7 +52,7 @@ import static org.mockito.Mockito.when;
*/
public class CommonTest
{
- static final Logger LOGGER = LoggerFactory.getLogger(CommonTest.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(CommonTest.class);
Vertx vertx;
Server server;
CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
@@ -61,7 +62,7 @@ public class CommonTest
{
Injector injector;
Module testOverride = Modules.override(new TestModule())
- .with(new CommonTestModule());
+ .with(new CommonTestModule(delegate));
injector = Guice.createInjector(Modules.override(SidecarModules.all())
.with(testOverride));
vertx = injector.getInstance(Vertx.class);
@@ -84,25 +85,56 @@ public class CommonTest
LOGGER.error("Close event timed out.");
}
- class CommonTestModule extends AbstractModule
+ /**
+ * Common test module to provide default mocks for testing
+ */
+ public static class CommonTestModule extends AbstractModule
{
+ protected static final int DEFAULT_INSTANCE_ID = 100;
+ protected static final String DEFAULT_HOST = "127.0.0.1";
+ protected static final int DEFAULT_PORT = 9042;
+
+ protected CassandraAdapterDelegate delegate;
+ protected StorageOperations storageOperations;
+
+ /**
+ * Test module with custom delegate
+ *
+ * @param delegate the CassandraAdapterDelegate mock to use
+ */
+ public CommonTestModule(CassandraAdapterDelegate delegate)
+ {
+ this.delegate = delegate;
+ this.storageOperations = delegate.storageOperations();
+ }
+
+ /**
+ * Test module with custom StorageOperations
+ *
+ * @param storageOperations the StorageOperations mock to use
+ */
+ public CommonTestModule(StorageOperations storageOperations)
+ {
+ this.delegate = mock(CassandraAdapterDelegate.class);
+ this.storageOperations = storageOperations;
+ when(delegate.storageOperations()).thenReturn(storageOperations);
+ }
+
@Provides
@Singleton
public InstancesMetadata instanceConfig()
{
- int instanceId = 100;
- String host = "127.0.0.1";
InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
- when(instanceMetadata.host()).thenReturn(host);
- when(instanceMetadata.port()).thenReturn(9042);
- when(instanceMetadata.id()).thenReturn(instanceId);
+ when(instanceMetadata.host()).thenReturn(DEFAULT_HOST);
+ when(instanceMetadata.port()).thenReturn(DEFAULT_PORT);
+ when(instanceMetadata.id()).thenReturn(DEFAULT_INSTANCE_ID);
when(instanceMetadata.stagingDir()).thenReturn("");
when(instanceMetadata.delegate()).thenReturn(delegate);
InstancesMetadata mockInstancesMetadata = mock(InstancesMetadata.class);
- when(mockInstancesMetadata.instances()).thenReturn(Collections.singletonList(instanceMetadata));
- when(mockInstancesMetadata.instanceFromId(instanceId)).thenReturn(instanceMetadata);
- when(mockInstancesMetadata.instanceFromHost(host)).thenReturn(instanceMetadata);
+ when(mockInstancesMetadata.instances()).thenReturn(List.of(instanceMetadata));
+ when(mockInstancesMetadata.instanceFromId(DEFAULT_INSTANCE_ID)).thenReturn(instanceMetadata);
+ when(mockInstancesMetadata.instanceFromHost(DEFAULT_HOST)).thenReturn(instanceMetadata);
return mockInstancesMetadata;
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandlerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandlerTest.java
index bc6f8cad..5a36bfde 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandlerTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/handlers/ListOperationalJobsHandlerTest.java
@@ -40,6 +40,7 @@ import com.google.inject.Module;
import com.google.inject.Provides;
import com.google.inject.Singleton;
import com.google.inject.util.Modules;
+import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.client.predicate.ResponsePredicate;
@@ -150,14 +151,15 @@ class ListOperationalJobsHandlerTest
}
@Override
- public boolean isRunningOnCassandra()
+ public boolean hasConflict(List<OperationalJob> jobs)
{
return false;
}
@Override
- protected void executeInternal() throws OperationalJobException
+ protected Future<Void> executeInternal() throws OperationalJobException
{
+ return Future.succeededFuture();
}
}
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/handlers/NodeDecommissionHandlerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/handlers/NodeDecommissionHandlerTest.java
index 70d530bf..ba4fa48e 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/handlers/NodeDecommissionHandlerTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/handlers/NodeDecommissionHandlerTest.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.sidecar.handlers;
-import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -29,12 +28,9 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.inject.AbstractModule;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.Module;
-import com.google.inject.Provides;
-import com.google.inject.Singleton;
import com.google.inject.util.Modules;
import io.vertx.core.Vertx;
import io.vertx.ext.web.client.WebClient;
@@ -42,9 +38,6 @@ import io.vertx.ext.web.client.predicate.ResponsePredicate;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
import org.apache.cassandra.sidecar.TestModule;
-import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
-import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
-import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
import org.apache.cassandra.sidecar.common.server.StorageOperations;
import org.apache.cassandra.sidecar.modules.SidecarModules;
@@ -78,7 +71,7 @@ public class NodeDecommissionHandlerTest
{
Injector injector;
Module testOverride = Modules.override(new TestModule())
- .with(new NodeDecommissionHandlerTest.NodeDecommissionTestModule());
+ .with(new CommonTest.CommonTestModule(mockStorageOperations));
injector = Guice.createInjector(Modules.override(SidecarModules.all())
.with(testOverride));
vertx = injector.getInstance(Vertx.class);
@@ -172,36 +165,4 @@ public class NodeDecommissionHandlerTest
context.completeNow();
}));
}
-
- /**
- * Test guice module for Node Decommission handler tests
- */
- class NodeDecommissionTestModule extends AbstractModule
- {
- @Provides
- @Singleton
- public InstancesMetadata instanceMetadata()
- {
- final int instanceId = 100;
- final String host = "127.0.0.1";
- final InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
- when(instanceMetadata.host()).thenReturn(host);
- when(instanceMetadata.port()).thenReturn(9042);
- when(instanceMetadata.id()).thenReturn(instanceId);
- when(instanceMetadata.stagingDir()).thenReturn("");
-
- CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
-
- when(delegate.storageOperations()).thenReturn(mockStorageOperations);
- when(instanceMetadata.delegate()).thenReturn(delegate);
-
- InstancesMetadata mockInstancesMetadata = mock(InstancesMetadata.class);
- when(mockInstancesMetadata.instances()).thenReturn(Collections.singletonList(instanceMetadata));
- when(mockInstancesMetadata.instanceFromId(instanceId)).thenReturn(instanceMetadata);
- when(mockInstancesMetadata.instanceFromHost(host)).thenReturn(instanceMetadata);
-
- return mockInstancesMetadata;
- }
- }
-
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/handlers/RepairHandlerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/handlers/RepairHandlerTest.java
new file mode 100644
index 00000000..6acc3545
--- /dev/null
+++ b/server/src/test/java/org/apache/cassandra/sidecar/handlers/RepairHandlerTest.java
@@ -0,0 +1,463 @@
+/*
+ * 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.handlers;
+
+import java.util.List;
+import java.util.Map;
+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 org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.TableMetadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.util.Modules;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.modules.SidecarModules;
+import org.apache.cassandra.sidecar.server.Server;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.mockito.AdditionalAnswers;
+import org.mockito.ArgumentCaptor;
+
+import static io.netty.handler.codec.http.HttpResponseStatus.ACCEPTED;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+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.SUCCEEDED;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the {@link RepairHandler}
+ */
+@ExtendWith(VertxExtension.class)
+class RepairHandlerTest
+{
+ static final Logger LOGGER = LoggerFactory.getLogger(RepairHandlerTest.class);
+ private static final String REPAIR_ROUTE = "/api/v1/cassandra/keyspaces/testkeyspace/repair";
+ Vertx vertx;
+ Server server;
+ StorageOperations mockStorageOperations = mock(StorageOperations.class);
+ InstanceMetadataFetcher mockMetadataFetcher = mock(InstanceMetadataFetcher.class);
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ // Set up the mock metadata chain
+ InstanceMetadata mockInstanceMetadata = mock(InstanceMetadata.class);
+ CassandraAdapterDelegate mockDelegate = mock(CassandraAdapterDelegate.class);
+ Metadata mockMetadata = mock(Metadata.class);
+ KeyspaceMetadata mockKeyspaceMetadata = mock(KeyspaceMetadata.class);
+ TableMetadata mockTableMetadata = mock(TableMetadata.class);
+
+ // Configure the mock chain
+ when(mockMetadataFetcher.instance(anyString())).thenReturn(mockInstanceMetadata);
+ when(mockMetadataFetcher.delegate(anyString())).thenReturn(mockDelegate); // Add this line to fix the NPE
+ when(mockInstanceMetadata.delegate()).thenReturn(mockDelegate);
+ when(mockDelegate.metadata()).thenReturn(mockMetadata);
+ when(mockDelegate.storageOperations()).thenReturn(mockStorageOperations);
+ when(mockMetadata.getKeyspace(anyString())).thenReturn(mockKeyspaceMetadata);
+ when(mockKeyspaceMetadata.getTable(anyString())).thenReturn(mockTableMetadata);
+
+ AbstractModule repairTestModule = new AbstractModule()
+ {
+ @Override
+ protected void configure()
+ {
+ // Bind the mocks needed for the test
+ bind(StorageOperations.class).toInstance(mockStorageOperations);
+ bind(InstanceMetadataFetcher.class).toInstance(mockMetadataFetcher);
+ }
+ };
+
+ // Create the injector with the proper module overrides
+ Injector injector = Guice.createInjector(
+ Modules.override(SidecarModules.all())
+ .with(Modules.override(new TestModule())
+ .with(new CommonTest.CommonTestModule(mockStorageOperations), repairTestModule))
+ );
+ vertx = injector.getInstance(Vertx.class);
+ server = injector.getInstance(Server.class);
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(s -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(5, TimeUnit.SECONDS);
+ }
+
+ @AfterEach
+ void after() throws InterruptedException
+ {
+ CountDownLatch closeLatch = new CountDownLatch(1);
+ server.close().onSuccess(res -> closeLatch.countDown());
+ if (closeLatch.await(60, TimeUnit.SECONDS))
+ LOGGER.info("Close event received before timeout.");
+ else
+ LOGGER.error("Close event timed out.");
+ }
+
+ @Test
+ void testRepairHandler(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .repairType(RepairPayload.RepairType.INCREMENTAL)
+ .tables(List.of("test_table"))
+ .build();
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(OK.code());
+ LOGGER.info("Repair Response: {}", response.bodyAsString());
+
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(SUCCEEDED);
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerIR(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ ArgumentCaptor<Map<String, String>> jobCapture = ArgumentCaptor.forClass(Map.class);
+ RepairPayload payload = RepairPayload.builder()
+ .repairType(RepairPayload.RepairType.INCREMENTAL)
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(OK.code());
+ LOGGER.info("Repair Response: {}", response.bodyAsString());
+
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(SUCCEEDED);
+ verify(mockStorageOperations).repairAsync(anyString(), jobCapture.capture());
+ assertThat(jobCapture.getValue()).containsKey("incremental");
+ assertThat(jobCapture.getValue().get("incremental")).isEqualTo("true");
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerWithRanges(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String expectedRanges = 0L + ":" + Integer.MAX_VALUE;
+ ArgumentCaptor<Map<String, String>> jobCapture = ArgumentCaptor.forClass(Map.class);
+ RepairPayload payload = RepairPayload.builder()
+ .startToken("0")
+ .endToken(Integer.toString(Integer.MAX_VALUE))
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(OK.code());
+ LOGGER.info("Repair Response: {}", response.bodyAsString());
+
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(SUCCEEDED);
+ verify(mockStorageOperations).repairAsync(anyString(), jobCapture.capture());
+ assertThat(jobCapture.getValue()).containsKey("ranges");
+ assertThat(jobCapture.getValue().get("ranges")).isEqualTo(expectedRanges);
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerWithHosts(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ ArgumentCaptor<Map<String, String>> jobCapture = ArgumentCaptor.forClass(Map.class);
+ RepairPayload payload = RepairPayload.builder()
+ .hosts(List.of("127.0.0.1"))
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(OK.code());
+ LOGGER.info("Repair Response: {}", response.bodyAsString());
+
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(SUCCEEDED);
+ verify(mockStorageOperations).repairAsync(anyString(), jobCapture.capture());
+ assertThat(jobCapture.getValue()).containsKey("hosts");
+ assertThat(jobCapture.getValue().get("hosts")).isEqualTo("127.0.0.1");
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerLongRunning(VertxTestContext context)
+ {
+ // Simulate repair invocation taking 6s
+ doAnswer(AdditionalAnswers.answersWithDelay(6000, invocation -> null))
+ .when(mockStorageOperations).repairAsync(anyString(), any());
+
+ WebClient client = WebClient.create(vertx);
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of("test_table"))
+ .build();
+
+ // Since it was the OperationalJobExecutionTimeout that triggered the response, the job is still
+ // in CREATED state, as we returned before polling for the job status.
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(ACCEPTED.code());
+ LOGGER.info("Repair Response: {}", response.bodyAsString());
+
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(CREATED);
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerBadRequest(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .send(context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(BAD_REQUEST.code());
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerFailed(VertxTestContext context)
+ {
+ doThrow(new RuntimeException("Simulated failure")).when(mockStorageOperations).repairAsync(anyString(), any());
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/api/v1/cassandra/keyspaces/testkeyspace/repair";
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of("test_table"))
+ .build();
+
+
+ client.put(server.actualPort(), "127.0.0.1", testRoute)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(OK.code());
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.jobId()).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(FAILED);
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerInvalidTokenRange_StartGreaterThanEnd(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ RepairPayload payload = RepairPayload.builder()
+ .startToken("100")
+ .endToken("50") // End token less than start token
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(BAD_REQUEST.code());
+ assertThat(response.bodyAsString()).contains("Start token must be less than end token");
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerInvalidTokenRange_StartEqualToEnd(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ RepairPayload payload = RepairPayload.builder()
+ .startToken("100")
+ .endToken("100") // End token equal to start token
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(BAD_REQUEST.code());
+ assertThat(response.bodyAsString()).contains("Start token must be less than end token");
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerInvalidTokenFormat_NonNumericStartToken(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ RepairPayload payload = RepairPayload.builder()
+ .startToken("invalid_token")
+ .endToken("100")
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(BAD_REQUEST.code());
+ assertThat(response.bodyAsString()).contains("Invalid token format");
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerInvalidTokenFormat_NonNumericEndToken(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ RepairPayload payload = RepairPayload.builder()
+ .startToken("50")
+ .endToken("invalid_token")
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(BAD_REQUEST.code());
+ assertThat(response.bodyAsString()).contains("Invalid token format");
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerValidLargeTokenRange(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ ArgumentCaptor<Map<String, String>> jobCapture = ArgumentCaptor.forClass(Map.class);
+ // Test with very large token values (BigInteger range)
+ String startToken = "123456789012345678901234567890";
+ String endToken = "987654321098765432109876543210";
+ RepairPayload payload = RepairPayload.builder()
+ .startToken(startToken)
+ .endToken(endToken)
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(OK.code());
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(SUCCEEDED);
+ verify(mockStorageOperations).repairAsync(anyString(), jobCapture.capture());
+ assertThat(jobCapture.getValue()).containsKey("ranges");
+ assertThat(jobCapture.getValue().get("ranges")).isEqualTo(startToken + ":" + endToken);
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerValidNegativeTokenRange(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ ArgumentCaptor<Map<String, String>> jobCapture = ArgumentCaptor.forClass(Map.class);
+ // Test with negative token values
+ String startToken = "-1000";
+ String endToken = "1000";
+ RepairPayload payload = RepairPayload.builder()
+ .startToken(startToken)
+ .endToken(endToken)
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(OK.code());
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(SUCCEEDED);
+ verify(mockStorageOperations).repairAsync(anyString(), jobCapture.capture());
+ assertThat(jobCapture.getValue()).containsKey("ranges");
+ assertThat(jobCapture.getValue().get("ranges")).isEqualTo(startToken + ":" + endToken);
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ void testRepairHandlerOnlyStartTokenProvided(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ ArgumentCaptor<Map<String, String>> jobCapture = ArgumentCaptor.forClass(Map.class);
+ // Test with only start token (should not add ranges option)
+ RepairPayload payload = RepairPayload.builder()
+ .startToken("100")
+ // No end token
+ .tables(List.of("test_table"))
+ .build();
+
+ client.put(server.actualPort(), "127.0.0.1", REPAIR_ROUTE)
+ .putHeader("Content-Type", "application/json")
+ .sendJson(payload, context.succeeding(response -> {
+ assertThat(response.statusCode()).isEqualTo(OK.code());
+ OperationalJobResponse repairResponse = response.bodyAsJson(OperationalJobResponse.class);
+ assertThat(repairResponse).isNotNull();
+ assertThat(repairResponse.status()).isEqualTo(SUCCEEDED);
+ verify(mockStorageOperations).repairAsync(anyString(), jobCapture.capture());
+ // Should not contain ranges since only start token was provided
+ assertThat(jobCapture.getValue()).doesNotContainKey("ranges");
+ context.completeNow();
+ }));
+ }
+}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/job/NodeDrainJobTest.java b/server/src/test/java/org/apache/cassandra/sidecar/job/NodeDrainJobTest.java
index 72df6fb1..4fb0ca9d 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/job/NodeDrainJobTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/job/NodeDrainJobTest.java
@@ -19,6 +19,7 @@
package org.apache.cassandra.sidecar.job;
import java.io.IOException;
+import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
@@ -34,7 +35,6 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -73,39 +73,35 @@ class NodeDrainJobTest
{
when(mockStorageOperations.operationMode()).thenReturn(OPERATION_MODE_DRAINING);
- assertThat(nodeDrainJob.isRunningOnCassandra()).isTrue();
+ assertThat(nodeDrainJob.hasConflict(Collections.emptyList())).isTrue();
}
@Test
void testIsRunningOnCassandra_WhenDrained()
{
when(mockStorageOperations.operationMode()).thenReturn(OPERATION_MODE_DRAINED);
-
- assertThat(nodeDrainJob.isRunningOnCassandra()).isFalse();
+ assertThat(nodeDrainJob.hasConflict(Collections.emptyList())).isFalse();
}
@Test
void testIsRunningOnCassandra_WhenNormal()
{
when(mockStorageOperations.operationMode()).thenReturn(OPERATION_MODE_NORMAL);
-
- assertThat(nodeDrainJob.isRunningOnCassandra()).isFalse();
+ assertThat(nodeDrainJob.hasConflict(Collections.emptyList())).isFalse();
}
@Test
void testIsRunningOnCassandra_WhenUnknownState()
{
when(mockStorageOperations.operationMode()).thenReturn(OPERATION_MODE_UNKNOWN);
-
- assertThat(nodeDrainJob.isRunningOnCassandra()).isFalse();
+ assertThat(nodeDrainJob.hasConflict(Collections.emptyList())).isFalse();
}
@Test
void testIsRunningOnCassandra_WhenNull()
{
when(mockStorageOperations.operationMode()).thenReturn(null);
-
- assertThat(nodeDrainJob.isRunningOnCassandra()).isFalse();
+ assertThat(nodeDrainJob.hasConflict(Collections.emptyList())).isFalse();
}
@Test
@@ -158,16 +154,6 @@ class NodeDrainJobTest
verify(mockStorageOperations).drain();
}
- @Test
- void testExecuteInternal_WhenAlreadyDraining() throws IOException, ExecutionException, InterruptedException
- {
- when(mockStorageOperations.operationMode()).thenReturn(OPERATION_MODE_DRAINING);
-
- nodeDrainJob.executeInternal();
-
- verify(mockStorageOperations, never()).drain();
- }
-
@Test
void testExecuteInternal_WhenDrainThrowsIOException() throws IOException, ExecutionException, InterruptedException
{
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/job/NodeMoveJobTest.java b/server/src/test/java/org/apache/cassandra/sidecar/job/NodeMoveJobTest.java
index ecd4271b..871ec7b0 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/job/NodeMoveJobTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/job/NodeMoveJobTest.java
@@ -19,6 +19,7 @@
package org.apache.cassandra.sidecar.job;
import java.io.IOException;
+import java.util.Collections;
import java.util.UUID;
import org.junit.jupiter.api.BeforeEach;
@@ -76,7 +77,7 @@ class NodeMoveJobTest
{
when(mockStorageOperations.operationMode()).thenReturn(OPERATION_MODE_MOVING);
NodeMoveJob job = new NodeMoveJob(jobId, newToken, mockStorageOperations);
- assertThat(job.isRunningOnCassandra()).isTrue();
+ assertThat(job.hasConflict(Collections.emptyList())).isTrue();
}
@Test
@@ -84,7 +85,7 @@ class NodeMoveJobTest
{
when(mockStorageOperations.operationMode()).thenReturn(OPERATION_MODE_NORMAL);
NodeMoveJob job = new NodeMoveJob(jobId, newToken, mockStorageOperations);
- assertThat(job.isRunningOnCassandra()).isFalse();
+ assertThat(job.hasConflict(Collections.emptyList())).isFalse();
}
@Test
@@ -92,7 +93,7 @@ class NodeMoveJobTest
{
when(mockStorageOperations.operationMode()).thenReturn(OPERATION_MODE_JOINING);
NodeMoveJob job = new NodeMoveJob(jobId, newToken, mockStorageOperations);
- assertThat(job.isRunningOnCassandra()).isFalse();
+ assertThat(job.hasConflict(Collections.emptyList())).isFalse();
}
@Test
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java
index d3574564..1b062f8a 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobManagerTest.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.sidecar.job;
+import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -28,6 +29,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import com.datastax.driver.core.utils.UUIDs;
+import io.vertx.core.Future;
import io.vertx.core.Vertx;
import org.apache.cassandra.sidecar.TestResourceReaper;
import org.apache.cassandra.sidecar.common.server.exceptions.OperationalJobException;
@@ -123,7 +125,7 @@ class OperationalJobManagerTest
// Job should be running initially
assertThat(testJob.asyncResult().isComplete()).isFalse();
assertThat(tracker.get(jobId)).isNotNull();
-
+
// Wait for completion
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(testJob.asyncResult().isComplete()).isTrue();
@@ -141,13 +143,13 @@ class OperationalJobManagerTest
OperationalJob failingJob = new OperationalJob(jobId)
{
@Override
- public boolean isRunningOnCassandra()
+ public boolean hasConflict(List<OperationalJob> jobs)
{
return false;
}
@Override
- protected void executeInternal() throws OperationalJobException
+ protected Future<Void> executeInternal() throws OperationalJobException
{
throw new OperationalJobException(msg);
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobTest.java b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobTest.java
index 9dab537b..fedd3c39 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/job/OperationalJobTest.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.sidecar.job;
+import java.util.List;
import java.util.UUID;
import com.google.common.util.concurrent.Uninterruptibles;
@@ -60,12 +61,13 @@ class OperationalJobTest
return new OperationalJob(jobId)
{
@Override
- protected void executeInternal() throws OperationalJobException
+ protected Future<Void> executeInternal() throws OperationalJobException
{
+ return Future.succeededFuture();
}
@Override
- public boolean isRunningOnCassandra()
+ public boolean hasConflict(List<OperationalJob> jobs)
{
return jobStatus == OperationalJobStatus.RUNNING;
}
@@ -94,13 +96,13 @@ class OperationalJobTest
return new OperationalJob(jobId)
{
@Override
- public boolean isRunningOnCassandra()
+ public boolean hasConflict(List<OperationalJob> jobs)
{
return false;
}
@Override
- protected void executeInternal() throws OperationalJobException
+ protected Future<Void> executeInternal() throws OperationalJobException
{
if (jobDuration != null)
{
@@ -111,6 +113,7 @@ class OperationalJobTest
{
throw jobFailure;
}
+ return Future.succeededFuture();
}
@Override
@@ -146,13 +149,13 @@ class OperationalJobTest
OperationalJob failingJob = new OperationalJob(UUIDs.timeBased())
{
@Override
- public boolean isRunningOnCassandra()
+ public boolean hasConflict(List<OperationalJob> jobs)
{
return false;
}
@Override
- protected void executeInternal() throws OperationalJobException
+ protected Future<Void> executeInternal() throws OperationalJobException
{
throw new OperationalJobException(msg);
}
diff --git a/server/src/test/java/org/apache/cassandra/sidecar/job/RepairJobTest.java b/server/src/test/java/org/apache/cassandra/sidecar/job/RepairJobTest.java
new file mode 100644
index 00000000..12d05e2d
--- /dev/null
+++ b/server/src/test/java/org/apache/cassandra/sidecar/job/RepairJobTest.java
@@ -0,0 +1,349 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+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.Promise;
+import io.vertx.core.Vertx;
+import org.apache.cassandra.sidecar.TestResourceReaper;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.request.data.RepairPayload;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.data.Name;
+import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.config.RepairJobsConfiguration;
+import org.apache.cassandra.sidecar.config.yaml.RepairJobsConfigurationImpl;
+import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
+import org.apache.cassandra.sidecar.coordination.ClusterLease;
+import org.apache.cassandra.sidecar.exceptions.OperationalJobConflictException;
+import org.apache.cassandra.sidecar.handlers.data.RepairRequestParam;
+import org.apache.cassandra.sidecar.tasks.PeriodicTaskExecutor;
+import org.mockito.Mockito;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests to validate job to run repairs
+ */
+class RepairJobTest
+{
+ private static final int MAX_ATTEMPTS = 5;
+ protected Vertx vertx;
+ protected ExecutorPools executorPool;
+ protected PeriodicTaskExecutor periodicTaskExecutor;
+ protected ClusterLease clusterLease;
+
+ @BeforeEach
+ void setup()
+ {
+ vertx = Vertx.vertx();
+ executorPool = new ExecutorPools(vertx, new ServiceConfigurationImpl());
+ clusterLease = new ClusterLease(ClusterLease.Ownership.CLAIMED);
+ periodicTaskExecutor = new PeriodicTaskExecutor(executorPool, clusterLease);
+ }
+
+ @AfterEach
+ void cleanup()
+ {
+ if (periodicTaskExecutor != null)
+ {
+ periodicTaskExecutor.close();
+ }
+ TestResourceReaper.create().with(executorPool).with(vertx).close();
+ }
+
+ @Test
+ void testRepairJob() throws Exception
+ {
+ StorageOperations storageOperations = mock(StorageOperations.class);
+ when(storageOperations.getParentRepairStatus(anyInt())).thenReturn(List.of(RepairJob.ParentRepairStatus.COMPLETED.name()));
+ when(storageOperations.repairAsync(any(), any())).thenReturn(1);
+
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of("testtable"))
+ .build();
+ RepairRequestParam repairParams = RepairRequestParam.from(new Name("testkeyspace"), payload);
+
+ RepairJobsConfiguration config = new RepairJobsConfigurationImpl(MAX_ATTEMPTS, MillisecondBoundConfiguration.parse("100ms")); // Use a shorter poll interval for testing
+ RepairJob testJob = new RepairJob(periodicTaskExecutor, config, UUIDs.timeBased(), storageOperations, repairParams);
+
+ Promise<Void> promise = Promise.promise();
+ testJob.execute(promise);
+
+ // Wait for the job to complete (with timeout)
+ if (!promise.future().isComplete())
+ {
+ promise.future().toCompletionStage().toCompletableFuture().get(1, TimeUnit.SECONDS);
+ }
+
+ assertThat(testJob.asyncResult().isComplete()).isTrue();
+ assertThat(testJob.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ }
+
+ @Test
+ void testRunningRepairJob()
+ {
+ StorageOperations storageOperations = mock(StorageOperations.class);
+ when(storageOperations.getParentRepairStatus(anyInt()))
+ .thenReturn(Collections.singletonList(RepairJob.ParentRepairStatus.IN_PROGRESS.name()));
+ when(storageOperations.repairAsync(any(), any())).thenReturn(1);
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of("testtable"))
+ .build();
+ RepairRequestParam repairParams = RepairRequestParam.from(new Name("testkeyspace"), payload);
+
+ RepairJobsConfiguration config = new RepairJobsConfigurationImpl(MAX_ATTEMPTS, MillisecondBoundConfiguration.parse("100ms")); // Shorter poll interval for testing
+ RepairJob testJob = new RepairJob(periodicTaskExecutor, config, UUIDs.timeBased(), storageOperations, repairParams);
+
+ Promise<Void> promise = Promise.promise();
+ testJob.execute(promise);
+
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ assertThat(testJob.asyncResult().isComplete()).isTrue();
+ assertThat(testJob.status()).isEqualTo(OperationalJobStatus.RUNNING);
+ Mockito.verify(storageOperations, atLeast(2)).getParentRepairStatus(anyInt());
+ }
+
+ @Test
+ void testLongRunningRepairJobTimeout() throws Exception
+ {
+ StorageOperations storageOperations = mock(StorageOperations.class);
+ // Return null for MAX_ATTEMPTS times to simulate timeout
+ for (int i = 0; i < MAX_ATTEMPTS; i++)
+ {
+ when(storageOperations.getParentRepairStatus(anyInt()))
+ .thenReturn(null);
+ }
+ when(storageOperations.repairAsync(any(), any())).thenReturn(1);
+
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of("testtable"))
+ .build();
+ RepairRequestParam repairParams = RepairRequestParam.from(new Name("testkeyspace"), payload);
+
+ RepairJobsConfiguration config = new RepairJobsConfigurationImpl(MAX_ATTEMPTS, MillisecondBoundConfiguration.parse("100ms")); // Shorter poll interval for testing
+ RepairJob testJob = new RepairJob(periodicTaskExecutor, config, UUIDs.timeBased(), storageOperations, repairParams);
+
+ Promise<Void> promise = Promise.promise();
+ testJob.execute(promise);
+
+ promise.future().toCompletionStage().toCompletableFuture().get(2, TimeUnit.SECONDS);
+ assertThat(testJob.asyncResult().isComplete()).isTrue();
+ assertThat(testJob.status()).isEqualTo(OperationalJobStatus.RUNNING);
+ }
+
+ @Test
+ void testTimersOnCompletion() throws Exception
+ {
+ StorageOperations storageOperations = mock(StorageOperations.class);
+ when(storageOperations.getParentRepairStatus(anyInt()))
+ .thenReturn(List.of(RepairJob.ParentRepairStatus.COMPLETED.name(), "Repair completed successfully"));
+ when(storageOperations.repairAsync(any(), any())).thenReturn(1);
+
+ RepairJobsConfiguration config = new RepairJobsConfigurationImpl(MAX_ATTEMPTS, MillisecondBoundConfiguration.parse("100ms")); // Short poll interval for quick test
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of("testtable"))
+ .build();
+ RepairRequestParam repairParams = RepairRequestParam.from(new Name("testkeyspace"), payload);
+
+ RepairJob testJob = new RepairJob(periodicTaskExecutor, config, UUIDs.timeBased(), storageOperations, repairParams);
+
+ Promise<Void> promise = Promise.promise();
+ testJob.execute(promise);
+
+ if (!promise.future().isComplete())
+ {
+ promise.future().toCompletionStage().toCompletableFuture().get(5, java.util.concurrent.TimeUnit.SECONDS);
+ }
+
+ assertThat(testJob.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ }
+
+ @Test
+ void testTimersOnFailure() throws Exception
+ {
+ StorageOperations storageOperations = mock(StorageOperations.class);
+ when(storageOperations.getParentRepairStatus(anyInt()))
+ .thenReturn(List.of(RepairJob.ParentRepairStatus.FAILED.name(), "Repair failed with error"));
+ when(storageOperations.repairAsync(any(), any())).thenReturn(1);
+
+ RepairJobsConfiguration config = new RepairJobsConfigurationImpl(MAX_ATTEMPTS, MillisecondBoundConfiguration.parse("100ms")); // Short poll interval for quick test
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of("testtable"))
+ .build();
+ RepairRequestParam repairParams = RepairRequestParam.from(new Name("testkeyspace"), payload);
+ RepairJob testJob = new RepairJob(periodicTaskExecutor, config, UUIDs.timeBased(), storageOperations, repairParams);
+
+ Promise<Void> promise = Promise.promise();
+ testJob.execute(promise);
+
+ try
+ {
+ promise.future().toCompletionStage().toCompletableFuture().get(5, java.util.concurrent.TimeUnit.SECONDS);
+ }
+ catch (java.util.concurrent.ExecutionException e)
+ {
+ // Expected since the job should fail
+ }
+
+ assertThat(testJob.status()).isEqualTo(OperationalJobStatus.FAILED);
+ }
+
+ @Test
+ void testMultipleRepairJobsRunningInParallel() throws Exception
+ {
+ // Create a job tracker and manager
+ OperationalJobTracker tracker = new OperationalJobTracker(10);
+ OperationalJobManager manager = new OperationalJobManager(tracker, executorPool);
+
+ // Mock the storage operations
+ StorageOperations storageOperations = mock(StorageOperations.class);
+ when(storageOperations.getParentRepairStatus(anyInt())).thenReturn(List.of(RepairJob.ParentRepairStatus.COMPLETED.name()));
+ when(storageOperations.repairAsync(any(), any())).thenReturn(1);
+
+ // Create configuration for repair jobs with shorter poll interval for testing
+ RepairJobsConfiguration config = new RepairJobsConfigurationImpl(MAX_ATTEMPTS, MillisecondBoundConfiguration.parse("100ms"));
+
+ // Create multiple repair jobs with different parameters
+ RepairJob job1 = createRepairJob(config, storageOperations, "keyspace1", "table1");
+ RepairJob job2 = createRepairJob(config, storageOperations, "keyspace1", "table2");
+ RepairJob job3 = createRepairJob(config, storageOperations, "keyspace2", "table1");
+
+ CountDownLatch latch = new CountDownLatch(1);
+ BiConsumer<OperationalJob, OperationalJobConflictException> onComplete = (job, exception) -> {
+ assertThat(exception).isNull();
+ latch.countDown();
+ };
+
+ // Submit all jobs to the manager
+ Promise<Void> promise1 = Promise.promise();
+ Promise<Void> promise2 = Promise.promise();
+ Promise<Void> promise3 = Promise.promise();
+
+ job1.execute(promise1);
+ job2.execute(promise2);
+ job3.execute(promise3);
+
+ manager.trySubmitJob(job1, onComplete, executorPool.service(), SecondBoundConfiguration.parse("5s"));
+ manager.trySubmitJob(job2, onComplete, executorPool.service(), SecondBoundConfiguration.parse("5s"));
+ manager.trySubmitJob(job3, onComplete, executorPool.service(), SecondBoundConfiguration.parse("5s"));
+
+ // Wait for all jobs to complete (with timeout)
+ if (!promise1.future().isComplete())
+ {
+ promise1.future().toCompletionStage().toCompletableFuture().get(2, TimeUnit.SECONDS);
+ }
+ if (!promise2.future().isComplete())
+ {
+ promise2.future().toCompletionStage().toCompletableFuture().get(2, TimeUnit.SECONDS);
+ }
+ if (!promise3.future().isComplete())
+ {
+ promise3.future().toCompletionStage().toCompletableFuture().get(2, TimeUnit.SECONDS);
+ }
+
+ // Verify all jobs completed successfully
+ assertThat(job1.asyncResult().isComplete()).isTrue();
+ assertThat(job1.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ assertThat(tracker.get(job1.jobId())).isNotNull();
+
+ assertThat(job2.asyncResult().isComplete()).isTrue();
+ assertThat(job2.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ assertThat(tracker.get(job2.jobId())).isNotNull();
+
+ assertThat(job3.asyncResult().isComplete()).isTrue();
+ assertThat(job3.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ assertThat(tracker.get(job3.jobId())).isNotNull();
+
+ // Verify that all jobs were tracked
+ assertThat(tracker.jobsView().size()).isEqualTo(3);
+ }
+
+ @Test
+ void testEmptyRepairStatusHandling() throws Exception
+ {
+ StorageOperations storageOperations = mock(StorageOperations.class);
+
+ // Return empty status (repair not started), then return IN_PROGRESS, then COMPLETED
+ when(storageOperations.getParentRepairStatus(anyInt()))
+ .thenReturn(Collections.emptyList())
+ .thenReturn(Collections.emptyList())
+ .thenReturn(List.of(RepairJob.ParentRepairStatus.COMPLETED.name()));
+
+ when(storageOperations.repairAsync(any(), any())).thenReturn(1);
+
+ RepairJobsConfiguration config = new RepairJobsConfigurationImpl(MAX_ATTEMPTS, MillisecondBoundConfiguration.parse("100ms"));
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of("testtable"))
+ .build();
+ RepairRequestParam repairParams = RepairRequestParam.from(new Name("testkeyspace"), payload);
+
+ RepairJob testJob = new RepairJob(periodicTaskExecutor, config, UUIDs.timeBased(), storageOperations, repairParams);
+ Promise<Void> promise = Promise.promise();
+ testJob.execute(promise);
+
+ if (!promise.future().isComplete())
+ {
+ promise.future().toCompletionStage().toCompletableFuture().get(5, java.util.concurrent.TimeUnit.SECONDS);
+ }
+
+ assertThat(testJob.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
+ // Verify the repair status was queried at least 4 times
+ Mockito.verify(storageOperations, Mockito.atLeast(4)).getParentRepairStatus(anyInt());
+ }
+
+ /**
+ * Helper method to create a repair job with specific parameters
+ */
+ private RepairJob createRepairJob(RepairJobsConfiguration config, StorageOperations storageOperations,
+ String keyspace, String table)
+ {
+ RepairPayload payload = RepairPayload.builder()
+ .isPrimaryRange(true)
+ .tables(List.of(table))
+ .build();
+ RepairRequestParam repairParams = RepairRequestParam.from(new Name(keyspace), payload);
+ return new RepairJob(periodicTaskExecutor, config, UUIDs.timeBased(), storageOperations, repairParams);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org