This is an automated email from the ASF dual-hosted git repository.
sarankk 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 965af15e CASSSIDECAR-463: Add fast_forward_enabled flag to enable
eager pipelining of staging and importing in coordinated writes (#376)
965af15e is described below
commit 965af15e29d157bd1e1e7d3de706b4e3c62cf039
Author: Mansi Khara <[email protected]>
AuthorDate: Tue Aug 25 13:52:58 2026 -0400
CASSSIDECAR-463: Add fast_forward_enabled flag to enable eager pipelining
of staging and importing in coordinated writes (#376)
Patch by Mansi Khara; reviewed by Yifan Cai, Saranya Krishnakumar for
CASSSIDECAR-463
---
CHANGES.txt | 3 +-
NEWS.txt | 11 +++
.../sidecar/common/data/RestoreJobConstants.java | 14 ++++
.../data/CreateRestoreJobRequestPayload.java | 45 ++++++++++-
.../CreateRestoreJobRequestPayloadTest.java | 60 ++++++++++++--
.../apache/cassandra/sidecar/db/RestoreJob.java | 59 +++++++++++++-
.../sidecar/db/RestoreJobDatabaseAccessor.java | 2 +
.../sidecar/db/schema/RestoreJobsSchema.java | 8 +-
.../handlers/restore/UpdateRestoreJobHandler.java | 4 +-
.../restore/RestoreJobConsistencyChecker.java | 17 ++--
.../sidecar/restore/RestoreJobDiscoverer.java | 28 +++++--
.../sidecar/restore/RestoreProcessor.java | 5 +-
.../sidecar/restore/RestoreRangeTask.java | 10 +--
.../db/RestoreJobDatabaseAccessorIntTest.java | 29 +++++++
.../cassandra/sidecar/db/RestoreJobTest.java | 64 +++++++++++++++
.../cassandra/sidecar/db/SidecarSchemaTest.java | 29 +++----
.../restore/UpdateRestoreJobHandlerTest.java | 39 +++++++++
.../sidecar/restore/RestoreJobDiscovererTest.java | 92 +++++++++++++++++++---
.../sidecar/restore/RestoreProcessorTest.java | 42 ++++++++++
.../sidecar/restore/RestoreRangeTaskTest.java | 63 +++++++++++++++
20 files changed, 560 insertions(+), 64 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 3772e216..588af8c4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,6 +1,7 @@
0.5.0
-----
- * CachingSchemaStore should be reloaded when Schemastore kafka configs are
changed (CASSSIDECAR-493)
+ * Add fast_forward_enabled flag to enable eager pipelining of staging and
importing in coordinated writes (CASSSIDECAR-463)
+ * CachingSchemaStore should be reloaded when Schemastore kafka configs are
changed (CASSSIDECAR-493)
* Support retrying on HTTP 429 responses and fix duplicate immediate
execution in RequestExecutor retry scheduling (CASSSIDECAR-465)
* Add RFC 6902-inspired JSON Patch support to ConfigurationManager
(CASSSIDECAR-429)
* Implement job coordination for cluster-wide operations (CASSSIDECAR-377)
diff --git a/NEWS.txt b/NEWS.txt
index 788b4fd8..a304e695 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -3,6 +3,9 @@
Upgrading
---------
+ - In CASSSIDECAR-463, a new internal table (restore_job_v7) is created to
track the fastForwardEnabled
+ option of restore jobs. It replaces restore_job_v6. Spark jobs
performing S3 restores should be paused
+ during sidecar upgrade to ensure in-flight restore job state is not lost.
- In CASSSIDECAR-461, the list snapshot endpoint now validates the
snapshot name, aligned with
the validation introduced by CASSANDRA-21389. A new
cassandra_input_validation.allowed_chars_for_snapshot_name
setting in sidecar.yaml controls the accepted pattern; the default is
"[a-zA-Z0-9_.+-]{1,255}".
@@ -13,6 +16,14 @@ Upgrading
New features
------------
+ - Eager pipelining of the staging and the importing phases of coordinated
restore jobs
+ (CASSSIDECAR-463). Set fastForwardEnabled to true when creating a
restore job so that Sidecar stages
+ slices while the job is still in CREATED status and imports the staged
slices once the job enters
+ STAGED status, instead of waiting for the STAGE_READY and IMPORT_READY
signals. Only enable it when
+ the caller can accept that the imported data becomes immediately visible
and that importing cannot
+ be rolled back. Slices uploaded while the job is in CREATED status are
picked up by the periodic job
+ discovery, and, until all the declared slices are found, by every job
progress poll; set sliceCount
+ when creating the job and poll the progress to keep staging close to the
uploads.
- Snapshot name validation on the list snapshot endpoint
(CASSSIDECAR-461), with a configurable
allowed_chars_for_snapshot_name pattern under cassandra_input_validation
in sidecar.yaml.
diff --git
a/client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobConstants.java
b/client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobConstants.java
index 21991cf7..4609ffc4 100644
---
a/client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobConstants.java
+++
b/client-common/src/main/java/org/apache/cassandra/sidecar/common/data/RestoreJobConstants.java
@@ -48,6 +48,20 @@ public class RestoreJobConstants
* In this case, there is an external coordinator that manges the restore
job in each datacenter.
*/
public static final String JOB_RESTORE_TO_LOCAL_DATA_CENTER_ONLY =
"restoreToLocalDatacenterOnly";
+ /**
+ * A boolean field; when fastForwardEnabled is set to true, Sidecar
eagerly pipelines the staging and the importing
+ * phases of the restore job, instead of waiting for the phase signals
from the external controller. Concretely,
+ * slices are staged while the job is still in {@code CREATED} status, and
staged slices are imported once the job
+ * enters {@code STAGED} status. The {@code STAGE_READY} and {@code
IMPORT_READY} signals become a fence and a
+ * confirmation, respectively, rather than triggers.
+ * The option should only be enabled by callers that can accept that the
imported data becomes immediately visible
+ * and that importing cannot be rolled back.
+ * Slices that are uploaded while the job is in {@code CREATED} status are
picked up by the periodic restore job
+ * discovery of Sidecar, and, until all the declared slices are found, by
every poll of the job progress endpoint.
+ * A caller that wants the staging to keep up with its uploads should
therefore set {@code sliceCount} when
+ * creating the job and poll the job progress, rather than relying on the
discovery loop alone.
+ */
+ public static final String JOB_FAST_FORWARD_ENABLED = "fastForwardEnabled";
public static final String SLICE_ID = "sliceId";
public static final String BUCKET_ID = "bucketId";
public static final String SLICE_START_TOKEN = "startToken";
diff --git
a/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateRestoreJobRequestPayload.java
b/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateRestoreJobRequestPayload.java
index f6dbd49c..550a25d2 100644
---
a/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateRestoreJobRequestPayload.java
+++
b/client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateRestoreJobRequestPayload.java
@@ -24,6 +24,7 @@ import java.util.UUID;
import java.util.function.Consumer;
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 org.apache.cassandra.sidecar.common.DataObjectBuilder;
@@ -42,6 +43,7 @@ import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_A
import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_CONSISTENCY_LEVEL;
import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_CREDENTIAL_TYPE;
import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_EXPIRE_AT;
+import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_FAST_FORWARD_ENABLED;
import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_ID;
import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_IMPORT_OPTIONS;
import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_LOCAL_DATA_CENTER;
@@ -65,6 +67,8 @@ import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_S
* type is rejected at construction time to surface misconfigurations before
they reach the AWS SDK.
*/
@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+// Ignore the unknown fields, so that a newer client can talk to an older
server that does not know all the fields yet
+@JsonIgnoreProperties(ignoreUnknown = true)
public class CreateRestoreJobRequestPayload
{
private final UUID jobId;
@@ -75,6 +79,7 @@ public class CreateRestoreJobRequestPayload
private final long expireAtInMillis;
private final ConsistencyConfig consistencyConfig;
private final boolean localDatacenterOnly;
+ private final boolean fastForwardEnabled;
/**
* Builder to build a {@link CreateRestoreJobRequestPayload}.
@@ -109,6 +114,9 @@ public class CreateRestoreJobRequestPayload
* @param consistencyLevel consistency level a job should satisfy
* @param localDatacenter the local datacenter name; required if using
local consistency level and localDatacenterOnly is specified
* @param localDatacenterOnly whether the job should restore to the
specified local datacenter only
+ * @param fastForwardEnabled whether Sidecar should eagerly pipeline the
staging and the importing phases,
+ * i.e. stage while the job is in {@link
RestoreJobStatus#CREATED} status and import
+ * once the job is in {@link
RestoreJobStatus#STAGED} status
*/
@JsonCreator
public CreateRestoreJobRequestPayload(@JsonProperty(JOB_ID) UUID jobId,
@@ -119,7 +127,8 @@ public class CreateRestoreJobRequestPayload
@JsonProperty(JOB_EXPIRE_AT) long
expireAtInMillis,
@JsonProperty(JOB_CONSISTENCY_LEVEL)
String consistencyLevel,
@JsonProperty(JOB_LOCAL_DATA_CENTER)
String localDatacenter,
-
@JsonProperty(JOB_RESTORE_TO_LOCAL_DATA_CENTER_ONLY) boolean
localDatacenterOnly)
+
@JsonProperty(JOB_RESTORE_TO_LOCAL_DATA_CENTER_ONLY) boolean
localDatacenterOnly,
+
@JsonProperty(JOB_FAST_FORWARD_ENABLED) boolean fastForwardEnabled)
{
Preconditions.checkArgument(jobId == null || jobId.version() == 1,
"Only time based UUIDs allowed for jobId");
@@ -142,6 +151,11 @@ public class CreateRestoreJobRequestPayload
Preconditions.checkArgument(!localDatacenterOnly ||
StringUtils.isNotEmpty(localDatacenter),
"Must specify a localDatacenter when
restoreToLocalDatacenterOnly is true");
this.localDatacenterOnly = localDatacenterOnly;
+ // Fast forward only alters the phase transitions of Sidecar-managed
jobs. A job is Sidecar-managed if and only
+ // if it declares a consistency level. Reject the combination instead
of silently ignoring the flag.
+ Preconditions.checkArgument(!fastForwardEnabled ||
this.consistencyConfig.consistencyLevel != null,
+ "Must specify a " + JOB_CONSISTENCY_LEVEL
+ " when fastForwardEnabled is true");
+ this.fastForwardEnabled = fastForwardEnabled;
}
/**
@@ -242,6 +256,15 @@ public class CreateRestoreJobRequestPayload
return consistencyConfig;
}
+ /**
+ * @return whether Sidecar should eagerly pipeline the staging and the
importing phases of the job
+ */
+ @JsonProperty(JOB_FAST_FORWARD_ENABLED)
+ public boolean fastForwardEnabled()
+ {
+ return fastForwardEnabled;
+ }
+
/**
* @return the AWS region of the S3 bucket, derived from {@code
secrets.readCredentials().region()}
*/
@@ -261,6 +284,7 @@ public class CreateRestoreJobRequestPayload
JOB_CONSISTENCY_LEVEL + "='" + consistencyLevel() + "', " +
JOB_LOCAL_DATA_CENTER + "='" + localDatacenter() + "', " +
JOB_RESTORE_TO_LOCAL_DATA_CENTER_ONLY + "='" +
shouldRestoreToLocalDatacenterOnly() + "', " +
+ JOB_FAST_FORWARD_ENABLED + "='" + fastForwardEnabled + "', " +
JOB_IMPORT_OPTIONS + "='" + importOptions + "'}";
}
@@ -279,6 +303,7 @@ public class CreateRestoreJobRequestPayload
private ConsistencyLevel consistencyLevel = null;
private String localDc = null;
private boolean localDatacenterOnly = false;
+ private boolean fastForwardEnabled = false;
Builder(RestoreJobSecrets secrets, long expireAtInMillis)
{
@@ -324,6 +349,21 @@ public class CreateRestoreJobRequestPayload
return update(b -> b.localDatacenterOnly = localDatacenterOnly);
}
+ /**
+ * Enable eager pipelining of the staging and the importing phases. It
requires a consistency level, i.e. the
+ * restore job must be Sidecar-managed.
+ *
+ * <p>Only enable it when the caller can accept that the imported data
becomes immediately visible and that
+ * importing cannot be rolled back.
+ *
+ * @param fastForwardEnabled whether to enable fast forward
+ * @return builder
+ */
+ public Builder fastForwardEnabled(boolean fastForwardEnabled)
+ {
+ return update(b -> b.fastForwardEnabled = fastForwardEnabled);
+ }
+
@Override
public Builder self()
{
@@ -350,7 +390,8 @@ public class CreateRestoreJobRequestPayload
builder.expireAtInMillis,
nameOrNull(builder.consistencyLevel),
builder.localDc,
- builder.localDatacenterOnly);
+ builder.localDatacenterOnly,
+ builder.fastForwardEnabled);
}
private static String nameOrNull(ConsistencyLevel cl)
diff --git
a/client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateRestoreJobRequestPayloadTest.java
b/client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateRestoreJobRequestPayloadTest.java
index dc102c94..b0aedb62 100644
---
a/client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateRestoreJobRequestPayloadTest.java
+++
b/client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateRestoreJobRequestPayloadTest.java
@@ -28,7 +28,6 @@ import org.junit.jupiter.api.Test;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException;
import com.fasterxml.jackson.databind.exc.ValueInstantiationException;
import org.apache.cassandra.sidecar.common.data.ConsistencyLevel;
import org.apache.cassandra.sidecar.common.data.CredentialType;
@@ -39,6 +38,7 @@ import
org.apache.cassandra.sidecar.common.request.data.CreateRestoreJobRequestP
import org.apache.cassandra.sidecar.foundation.RestoreJobSecretsGen;
import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_CONSISTENCY_LEVEL;
+import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_FAST_FORWARD_ENABLED;
import static
org.apache.cassandra.sidecar.common.data.RestoreJobConstants.JOB_RESTORE_TO_LOCAL_DATA_CENTER_ONLY;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -94,17 +94,21 @@ class CreateRestoreJobRequestPayloadTest
}
@Test
- void testReadFromJsonFailsWithUnknownFields() throws
JsonProcessingException
+ void testReadFromJsonIgnoresUnknownFields() throws JsonProcessingException
{
String uuid = "e870e5dc-d25e-11ed-afa1-0242ac120002";
+ RestoreJobSecrets secrets =
RestoreJobSecretsGen.genRestoreJobSecrets();
+ // "status" is not a field of the create payload; it is ignored, so
that a newer client that sends fields
+ // unknown to this version can still be served. Note that the status
of a created job is always CREATED.
String json = "{\"jobId\":\"" + uuid + "\"," +
"\"jobAgent\":\"Spark Bulk Analytics\"," +
"\"status\":\"Completed\"," +
"\"expireAt\":" + (System.currentTimeMillis() + 1000) +
- ",\"secrets\":" +
MAPPER.writeValueAsString(RestoreJobSecretsGen.genRestoreJobSecrets()) + "}";
- assertThatThrownBy(() -> MAPPER.readValue(json,
CreateRestoreJobRequestPayload.class))
- .isInstanceOf(UnrecognizedPropertyException.class)
- .hasMessageContaining("Unrecognized field \"status\"");
+ ",\"secrets\":" + MAPPER.writeValueAsString(secrets) +
"}";
+ CreateRestoreJobRequestPayload payload = MAPPER.readValue(json,
CreateRestoreJobRequestPayload.class);
+ assertThat(payload.jobId()).hasToString(uuid);
+ assertThat(payload.jobAgent()).isEqualTo("Spark Bulk Analytics");
+ assertThat(payload.secrets()).isEqualTo(secrets);
}
@Test
@@ -398,4 +402,48 @@ class CreateRestoreJobRequestPayloadTest
.hasCauseInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Static credentials must have all key fields
present for writeCredentials");
}
+
+ @Test
+ void testFastForwardEnabledSerDeser() throws JsonProcessingException
+ {
+ RestoreJobSecrets secrets =
RestoreJobSecretsGen.genRestoreJobSecrets();
+ CreateRestoreJobRequestPayload req = CreateRestoreJobRequestPayload
+ .builder(secrets,
System.currentTimeMillis() + 10000)
+ .jobAgent("agent")
+
.consistencyLevel(ConsistencyLevel.QUORUM)
+ .fastForwardEnabled(true)
+ .build();
+ assertThat(req.fastForwardEnabled()).isTrue();
+
+ String json = MAPPER.writeValueAsString(req);
+ assertThat(json).contains("\"" + JOB_FAST_FORWARD_ENABLED + "\":true");
+ assertThat(MAPPER.readValue(json,
CreateRestoreJobRequestPayload.class).fastForwardEnabled()).isTrue();
+ }
+
+ @Test
+ void testFastForwardDisabledByDefault() throws JsonProcessingException
+ {
+ RestoreJobSecrets secrets =
RestoreJobSecretsGen.genRestoreJobSecrets();
+ CreateRestoreJobRequestPayload req = CreateRestoreJobRequestPayload
+ .builder(secrets,
System.currentTimeMillis() + 10000)
+
.consistencyLevel(ConsistencyLevel.QUORUM)
+ .build();
+ assertThat(req.fastForwardEnabled()).isFalse();
+ assertThat(MAPPER.writeValueAsString(req))
+ .describedAs("Default value fields should be excluded")
+ .doesNotContain(JOB_FAST_FORWARD_ENABLED);
+ }
+
+ @Test
+ void testFastForwardEnabledWithoutConsistencyLevelFails()
+ {
+ RestoreJobSecrets secrets =
RestoreJobSecretsGen.genRestoreJobSecrets();
+ // fast forward only applies to Sidecar-managed jobs, i.e. the jobs
that declare a consistency level
+ assertThatThrownBy(() -> CreateRestoreJobRequestPayload
+ .builder(secrets, System.currentTimeMillis()
+ 10000)
+ .fastForwardEnabled(true)
+ .build())
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Must specify a " + JOB_CONSISTENCY_LEVEL + " when
fastForwardEnabled is true");
+ }
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreJob.java
b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreJob.java
index f2df7917..52f0e68c 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreJob.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreJob.java
@@ -67,6 +67,9 @@ public class RestoreJob
public final @Nullable String localDatacenter;
// whether a restore job should restore to the local Cassandra nodes only;
default is false
public final boolean shouldRestoreToLocalDatacenterOnly;
+ // whether the staging and the importing phases of the job are eagerly
pipelined, i.e. slices are staged while the
+ // job is in CREATED status and staged slices are imported once the job is
in STAGED status; default is false
+ public final boolean fastForwardEnabled;
public final Manager restoreJobManager;
public final Long sliceCount;
@@ -100,6 +103,7 @@ public class RestoreJob
.consistencyLevel(consistencyConfig.consistencyLevel)
.localDatacenter(consistencyConfig.localDatacenter)
.shouldRestoreToLocalDatacenterOnly(row.getBool("local_datacenter_only"))
+ .fastForwardEnabled(row.getBool("fast_forward_enabled"))
.sliceCount(row.get("slice_count", Long.class));
return builder.build();
@@ -189,6 +193,7 @@ public class RestoreJob
this.localDatacenter = builder.localDatacenter;
this.restoreJobManager = builder.manager;
this.sliceCount = builder.sliceCount;
+ this.fastForwardEnabled = builder.fastForwardEnabled;
}
public Builder unbuild()
@@ -216,16 +221,54 @@ public class RestoreJob
return expireAt != null && referenceTimestampMillis >=
expireAt.getTime();
}
+ /**
+ * Determine whether the restore ranges of the job should be staged now.
+ *
+ * <p>Staging is normally triggered by the {@link
RestoreJobStatus#STAGE_READY} signal from the external
+ * controller. When fast forward is enabled, staging starts as soon as the
job is created; the individual slices
+ * are already available on the storage cloud as they are uploaded
progressively. {@code STAGE_READY} then becomes
+ * a fence, which guarantees that all slices of the job have been
uploaded, rather than the trigger of the phase.
+ *
+ * @return true if the ranges of this job can be staged in the current job
status
+ */
+ public boolean shouldStageNow()
+ {
+ return status == RestoreJobStatus.STAGE_READY
+ || (fastForwardEnabled && status == RestoreJobStatus.CREATED);
+ }
+
+ /**
+ * Determine whether the staged restore ranges of the job should be
imported now.
+ *
+ * <p>Importing is normally triggered by the {@link
RestoreJobStatus#IMPORT_READY} signal from the external
+ * controller. When fast forward is enabled, importing starts once the job
enters {@link RestoreJobStatus#STAGED},
+ * i.e. all clusters have staged their data and satisfied the consistency
requirement of the job.
+ * {@code IMPORT_READY} then becomes a confirmation rather than the
trigger of the phase.
+ *
+ * @return true if the staged ranges of this job can be imported in the
current job status
+ */
+ public boolean shouldImportNow()
+ {
+ return status == RestoreJobStatus.IMPORT_READY
+ || (fastForwardEnabled && status == RestoreJobStatus.STAGED);
+ }
+
/**
* Determine the expected range status based on the job status
* @return the expected next range status in order to succeed
*/
public RestoreRangeStatus expectedNextRangeStatus()
{
- Preconditions.checkArgument(status != RestoreJobStatus.CREATED,
+ // Ranges of a fast forward job are already staging while the job is
in CREATED status, so their progress can
+ // be examined. For all other jobs, no range exists yet in CREATED
status.
+ Preconditions.checkArgument(fastForwardEnabled || status !=
RestoreJobStatus.CREATED,
"Cannot check progress for restore job in
CREATED status. jobId: " + jobId);
- return status == RestoreJobStatus.STAGE_READY || status ==
RestoreJobStatus.STAGED
+ // The job is still staging its ranges when shouldStageNow() holds,
i.e. in CREATED status for a fast forward
+ // job and in STAGE_READY status for any job.
+ // The STAGED status maps to STAGED as well, but only in the normal
flow, where importing waits for the
+ // IMPORT_READY signal. With fast forward, importing has already
started in STAGED status, hence SUCCEEDED.
+ return shouldStageNow() || (status == RestoreJobStatus.STAGED &&
!fastForwardEnabled)
? RestoreRangeStatus.STAGED
: RestoreRangeStatus.SUCCEEDED;
}
@@ -245,13 +288,14 @@ public class RestoreJob
"createdAt='%s', jobId='%s', keyspaceName='%s', "
+
"tableName='%s', status='%s', secrets='%s',
importOptions='%s', " +
"expireAt='%s', bucketCount='%s',
consistencyLevel='%s', localDatacenter='%s', " +
- "shouldRestoreToLocalDatacenterOnly='%s'}",
+ "shouldRestoreToLocalDatacenterOnly='%s',
fastForwardEnabled='%s'}",
createdAt.toString(), jobId.toString(),
keyspaceName, tableName,
statusText, secrets, importOptions,
expireAt, bucketCount,
consistencyLevel, localDatacenter,
- shouldRestoreToLocalDatacenterOnly);
+ shouldRestoreToLocalDatacenterOnly,
+ fastForwardEnabled);
}
public static LocalDate toLocalDate(UUID jobId)
@@ -291,6 +335,7 @@ public class RestoreJob
private ConsistencyLevel consistencyLevel;
private String localDatacenter;
private boolean shouldRestoreToLocalDatacenterOnly = false;
+ private boolean fastForwardEnabled = false;
private Manager manager;
private Long sliceCount;
@@ -317,6 +362,7 @@ public class RestoreJob
this.localDatacenter = restoreJob.localDatacenter;
this.manager = restoreJob.restoreJobManager;
this.sliceCount = restoreJob.sliceCount;
+ this.fastForwardEnabled = restoreJob.fastForwardEnabled;
}
public Builder createdAt(LocalDate createdAt)
@@ -413,6 +459,11 @@ public class RestoreJob
return update(b -> b.shouldRestoreToLocalDatacenterOnly =
localDatacenterOnly);
}
+ public Builder fastForwardEnabled(boolean fastForwardEnabled)
+ {
+ return update(b -> b.fastForwardEnabled = fastForwardEnabled);
+ }
+
@Override
public Builder self()
{
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreJobDatabaseAccessor.java
b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreJobDatabaseAccessor.java
index 98454185..1d62a2d1 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreJobDatabaseAccessor.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/db/RestoreJobDatabaseAccessor.java
@@ -87,6 +87,7 @@ public class RestoreJobDatabaseAccessor extends
DatabaseAccessor<RestoreJobsSche
.consistencyLevel(payload.consistencyConfig().consistencyLevel)
.localDatacenter(payload.consistencyConfig().localDatacenter)
.shouldRestoreToLocalDatacenterOnly(payload.shouldRestoreToLocalDatacenterOnly())
+
.fastForwardEnabled(payload.fastForwardEnabled())
.build();
ByteBuffer secrets = serializeValue(job.secrets, "secrets");
ByteBuffer importOptions = serializeValue(job.importOptions, "sstable
import options");
@@ -103,6 +104,7 @@ public class RestoreJobDatabaseAccessor extends
DatabaseAccessor<RestoreJobsSche
job.localDatacenter,
job.shouldRestoreToLocalDatacenterOnly,
job.credentialType.name(),
+ job.fastForwardEnabled,
job.expireAt);
execute(statement);
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreJobsSchema.java
b/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreJobsSchema.java
index 42c50418..ea81eefb 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreJobsSchema.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/db/schema/RestoreJobsSchema.java
@@ -32,7 +32,7 @@ import org.jetbrains.annotations.NotNull;
*/
public class RestoreJobsSchema extends TableSchema implements
ExecuteOnClusterLeaseholderOnly
{
- private static final String RESTORE_JOB_TABLE_NAME = "restore_job_v6";
+ private static final String RESTORE_JOB_TABLE_NAME = "restore_job_v7";
private final SchemaKeyspaceConfiguration keyspaceConfig;
private final SecondBoundConfiguration tableTtl;
@@ -97,6 +97,7 @@ public class RestoreJobsSchema extends TableSchema implements
ExecuteOnClusterLe
" local_datacenter text," +
" local_datacenter_only boolean," +
" credential_type text," +
+ " fast_forward_enabled boolean," +
" PRIMARY KEY (created_at, job_id)" +
") WITH default_time_to_live = %s",
keyspaceConfig.keyspace(),
RESTORE_JOB_TABLE_NAME, tableTtl.toSeconds());
@@ -159,8 +160,9 @@ public class RestoreJobsSchema extends TableSchema
implements ExecuteOnClusterLe
" local_datacenter," +
" local_datacenter_only," +
" credential_type," +
+ " fast_forward_enabled," +
" expire_at" +
- ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
?)", config);
+ ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
?)", config);
}
static String updateBlobSecrets(SchemaKeyspaceConfiguration config)
@@ -223,6 +225,7 @@ public class RestoreJobsSchema extends TableSchema
implements ExecuteOnClusterLe
"local_datacenter, " +
"local_datacenter_only, " +
"credential_type, " +
+ "fast_forward_enabled, " +
"expire_at, " +
"slice_count " +
"FROM %s.%s " +
@@ -243,6 +246,7 @@ public class RestoreJobsSchema extends TableSchema
implements ExecuteOnClusterLe
"local_datacenter, " +
"local_datacenter_only, " +
"credential_type, " +
+ "fast_forward_enabled, " +
"expire_at, " +
"slice_count " +
"FROM %s.%s " +
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandler.java
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandler.java
index c217512c..aae9973b 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandler.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandler.java
@@ -168,7 +168,9 @@ public class UpdateRestoreJobHandler extends
AbstractHandler<UpdateRestoreJobReq
private void notifyPhaseSignalMaybe(RestoreJob updatedJob)
{
- if (updatedJob.status != RestoreJobStatus.IMPORT_READY &&
updatedJob.status != RestoreJobStatus.STAGE_READY)
+ // A phase signal is any status that lets Sidecar start staging or
importing. For a fast forward job, that
+ // includes STAGED, which starts the importing phase.
+ if (!updatedJob.shouldStageNow() && !updatedJob.shouldImportNow())
{
return;
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyChecker.java
b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyChecker.java
index ae73dd73..f260d90a 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyChecker.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobConsistencyChecker.java
@@ -40,7 +40,6 @@ import
org.apache.cassandra.sidecar.cluster.ConsistencyVerifiers;
import org.apache.cassandra.sidecar.cluster.locator.InstanceSetByDc;
import org.apache.cassandra.sidecar.common.data.ConsistencyVerificationResult;
import org.apache.cassandra.sidecar.common.data.RestoreJobProgressFetchPolicy;
-import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
import org.apache.cassandra.sidecar.common.server.cluster.locator.Token;
import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
@@ -75,7 +74,9 @@ public class RestoreJobConsistencyChecker
private final RestoreRangeDatabaseAccessor rangeDatabaseAccessor;
private final TaskExecutorPool taskExecutorPool;
private final RestoreMetrics restoreMetrics;
- private volatile boolean firstTimeSinceImportReady = true;
+ // Set to false after the first consistency check of the importing phase,
which starts at IMPORT_READY normally
+ // and at STAGED when the job has fast forward enabled
+ private volatile boolean firstCheckOfImportPhase = true;
@Inject
public RestoreJobConsistencyChecker(RingTopologyRefresher
ringTopologyRefresher,
@@ -140,11 +141,12 @@ public class RestoreJobConsistencyChecker
return true;
}
- if (restoreJob.status == RestoreJobStatus.IMPORT_READY &&
firstTimeSinceImportReady)
+ if (restoreJob.shouldImportNow() && firstCheckOfImportPhase)
{
- firstTimeSinceImportReady = false;
- LOGGER.info("First time checking consistency of the restore job
after import_ready. " +
- "Mark the progress as pending and force restore job
discover run. jobId={}", restoreJob.jobId);
+ firstCheckOfImportPhase = false;
+ LOGGER.info("First time checking consistency of the restore job
after the importing phase has started. " +
+ "Mark the progress as pending and force restore job
discover run. jobId={} status={}",
+ restoreJob.jobId, restoreJob.status);
return true;
}
return false;
@@ -313,7 +315,8 @@ public class RestoreJobConsistencyChecker
RestoreRangeStatus successCriteria,
RestoreRange
range)
{
- return concludeOneRange(populateReplicas(topology), verifier,
successCriteria, range.tokenRange().range, range.statusByReplica());
+ return concludeOneRange(populateReplicas(topology), verifier,
successCriteria,
+ range.tokenRange().range,
range.statusByReplica());
}
@VisibleForTesting
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
index 7bfd73e5..c49c2fc3 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscoverer.java
@@ -398,18 +398,27 @@ public class RestoreJobDiscoverer implements
PeriodicTask, RingTopologyChangeLis
private void processOneJob(RestoreJob job, RestoreJobManagerGroup
restoreJobManagers, RunContext context)
{
+ // read the status seen from the prior run before shouldLogJob records
the current one
+ boolean statusChanged = jobIdsByDay.getKnownStatus(job.jobId,
job.createdAt.getDaysSinceEpoch()) != job.status;
if (jobIdsByDay.shouldLogJob(job))
{
LOGGER.info("Found job. jobId={} job={}", job.jobId, job);
}
+ if (statusChanged)
+ {
+ // Unset the flag, so that the slices are re-discovered once in
the phase the job just entered.
+ // The flag is deliberately not unset on the subsequent runs of
the same status; a fast forward job finds
+ // and submits its slices while in STAGED status, and repeating
the discovery (which is expensive) on
+ // every run of the same phase would be wasteful.
+ jobIdsByDay.unsetSlicesDiscovered(job);
+ }
+
switch (job.status)
{
- case STAGED:
- // unset the flag, so that it can re-discover the slices when
the job status changes to a ready status
- jobIdsByDay.unsetSlicesDiscovered(job); // no break by design
case CREATED:
case STAGE_READY:
+ case STAGED:
case IMPORT_READY:
if (job.hasExpired(context.nowMillis))
{
@@ -465,8 +474,13 @@ public class RestoreJobDiscoverer implements PeriodicTask,
RingTopologyChangeLis
if (shouldFindSlicesAndSubmit(job))
{
findSlicesAndSubmit(job);
- // Mark the flag. It prevents finding slices (which is expensive)
until the flag is unset.
- jobIdsByDay.markSlicesDiscovered(job);
+ if (job.status != RestoreJobStatus.CREATED)
+ {
+ // Mark the flag. It prevents finding slices (which is
expensive) until the flag is unset.
+ // The flag is not marked while a fast forward job is still in
CREATED status, since its slices are
+ // uploaded progressively and each run should look for the
newly uploaded ones.
+ jobIdsByDay.markSlicesDiscovered(job);
+ }
}
}
@@ -507,7 +521,9 @@ public class RestoreJobDiscoverer implements PeriodicTask,
RingTopologyChangeLis
private boolean shouldFindSlicesAndSubmit(RestoreJob job)
{
- return (job.status == RestoreJobStatus.STAGE_READY || job.status ==
RestoreJobStatus.IMPORT_READY)
+ // Note that the discovered flag is never marked while a fast forward
job is in CREATED status, since its
+ // slices are uploaded progressively. The discovery therefore repeats
on every run of that phase.
+ return (job.shouldStageNow() || job.shouldImportNow())
&& !jobIdsByDay.isSliceDiscovered(job);
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreProcessor.java
b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreProcessor.java
index 0befaa2f..a837bdb8 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreProcessor.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreProcessor.java
@@ -36,7 +36,6 @@ import com.google.inject.Singleton;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import org.apache.cassandra.sidecar.cluster.locator.LocalTokenRangesProvider;
-import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
import
org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
import org.apache.cassandra.sidecar.concurrent.ConcurrencyLimiter;
@@ -168,9 +167,9 @@ public class RestoreProcessor implements PeriodicTask
break; // break in order to complete promise
}
- // only create task when the restore job status is import ready
for the staged ranges
+ // only create task when the restore job is ready to import the
staged ranges
// otherwise, putting the range back to the staged queue and exit
early.
- if (range.hasStaged() && range.job().status !=
RestoreJobStatus.IMPORT_READY)
+ if (range.hasStaged() && !range.job().shouldImportNow())
{
workQueue.offerStaged(range);
processMaxConcurrency.releasePermit();
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreRangeTask.java
b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreRangeTask.java
index 02f8cc52..ba3a1bea 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreRangeTask.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/restore/RestoreRangeTask.java
@@ -35,7 +35,6 @@ import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.ext.web.handler.HttpException;
import org.apache.cassandra.sidecar.cluster.locator.LocalTokenRangesProvider;
-import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
import org.apache.cassandra.sidecar.common.data.SSTableImportOptions;
import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
import org.apache.cassandra.sidecar.common.server.utils.ThrowableUtils;
@@ -145,7 +144,7 @@ public class RestoreRangeTask implements RestoreRangeHandler
RestoreJob job = range.job();
if (job.isManagedBySidecar())
{
- if (job.status == RestoreJobStatus.STAGE_READY)
+ if (job.shouldStageNow())
{
if (Files.exists(range.stagedObjectPath()))
{
@@ -169,7 +168,7 @@ public class RestoreRangeTask implements RestoreRangeHandler
return Future.succeededFuture();
});
}
- else if (job.status == RestoreJobStatus.IMPORT_READY)
+ else if (job.shouldImportNow())
{
if (range.hasStaged())
{
@@ -193,8 +192,9 @@ public class RestoreRangeTask implements RestoreRangeHandler
}
else
{
- String msg = "Unexpected restore job status. Expected only
STAGE_READY or IMPORT_READY when " +
- "processing active slices. Found status: " +
job.statusWithOptionalDescription();
+ String msg = "Unexpected restore job status. Expected the
job to be ready to stage or to import " +
+ "when processing active slices. Found status:
" + job.statusWithOptionalDescription() +
+ " and fast forward: " +
(job.fastForwardEnabled ? "enabled" : "disabled");
Exception unexpectedState = new IllegalStateException(msg);
return
Future.failedFuture(RestoreJobExceptions.ofFatal("Unexpected restore job
status",
range, unexpectedState));
diff --git
a/server/src/test/integration/org/apache/cassandra/sidecar/db/RestoreJobDatabaseAccessorIntTest.java
b/server/src/test/integration/org/apache/cassandra/sidecar/db/RestoreJobDatabaseAccessorIntTest.java
index 40dda516..1024f11d 100644
---
a/server/src/test/integration/org/apache/cassandra/sidecar/db/RestoreJobDatabaseAccessorIntTest.java
+++
b/server/src/test/integration/org/apache/cassandra/sidecar/db/RestoreJobDatabaseAccessorIntTest.java
@@ -120,6 +120,35 @@ class RestoreJobDatabaseAccessorIntTest extends
IntegrationTestBase
assertThat(found.secrets.readCredentials().sessionToken()).isNull();
}
+ @CassandraIntegrationTest
+ void testFastForwardEnabledRoundTrips()
+ {
+ waitForSchemaReady(30, TimeUnit.SECONDS);
+
+ RestoreJobDatabaseAccessor accessor =
injector.getInstance(RestoreJobDatabaseAccessor.class);
+ UUID jobId = UUIDs.timeBased();
+ CreateRestoreJobRequestPayload payload =
CreateRestoreJobRequestPayload.builder(secrets, expiresAtMillis)
+
.jobId(jobId)
+
.jobAgent("agent")
+
.consistencyLevel(ConsistencyLevel.QUORUM)
+
.fastForwardEnabled(true)
+
.build();
+ accessor.create(payload, qualifiedTableName);
+
+ RestoreJob found = accessor.find(jobId);
+ assertThat(found).isNotNull();
+ assertThat(found.fastForwardEnabled).isTrue();
+
+ // the flag is preserved when other fields of the job are updated
+ UpdateRestoreJobRequestPayload markStaged
+ = new UpdateRestoreJobRequestPayload(null, null,
RestoreJobStatus.STAGED, null, null);
+ accessor.update(markStaged, found);
+ assertThat(accessor.find(jobId).fastForwardEnabled).isTrue();
+
+ // and it is disabled by default
+
assertThat(accessor.find(createJob(accessor)).fastForwardEnabled).isFalse();
+ }
+
@CassandraIntegrationTest
void testStaticCredentialTypeRoundTrips()
{
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/db/RestoreJobTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/db/RestoreJobTest.java
index d94c5788..74c88581 100644
--- a/server/src/test/java/org/apache/cassandra/sidecar/db/RestoreJobTest.java
+++ b/server/src/test/java/org/apache/cassandra/sidecar/db/RestoreJobTest.java
@@ -20,6 +20,8 @@ package org.apache.cassandra.sidecar.db;
import java.util.Arrays;
import java.util.Date;
+import java.util.EnumMap;
+import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@@ -86,6 +88,12 @@ public class RestoreJobTest
return builder.build();
}
+ public static RestoreJob createFastForwardTestingJob(UUID jobId,
RestoreJobStatus status)
+ throws DataObjectMappingException
+ {
+ return createTestingJob(jobId,
status).unbuild().fastForwardEnabled(true).build();
+ }
+
public static RestoreJob createUpdatedJob(UUID jobId, String jobAgent,
RestoreJobStatus status,
RestoreJobSecrets secrets,
@@ -141,6 +149,62 @@ public class RestoreJobTest
}
}
+ @Test
+ void testExpectedNextRangeStatusOfFastForwardJob()
+ {
+ UUID jobId = UUIDs.timeBased();
+ // The expected range status of a fast forward job per job status. It
differs from the normal flow in the
+ // CREATED status, where the ranges are staged ahead of the
STAGE_READY signal, and in the STAGED status,
+ // where importing has already started instead of waiting for the
IMPORT_READY signal.
+ Map<RestoreJobStatus, RestoreRangeStatus> expectedByJobStatus = new
EnumMap<>(RestoreJobStatus.class);
+ expectedByJobStatus.put(RestoreJobStatus.CREATED,
RestoreRangeStatus.STAGED);
+ expectedByJobStatus.put(RestoreJobStatus.STAGE_READY,
RestoreRangeStatus.STAGED);
+ expectedByJobStatus.put(RestoreJobStatus.STAGED,
RestoreRangeStatus.SUCCEEDED);
+ expectedByJobStatus.put(RestoreJobStatus.IMPORT_READY,
RestoreRangeStatus.SUCCEEDED);
+
+ expectedByJobStatus.forEach((jobStatus, expectedRangeStatus) ->
+
assertThat(createFastForwardTestingJob(jobId,
jobStatus).expectedNextRangeStatus())
+ .describedAs("Expecting the ranges of a
fast forward job in " + jobStatus +
+ " status to enter " +
expectedRangeStatus)
+ .isEqualTo(expectedRangeStatus));
+ }
+
+ @Test
+ void testPhaseGatesOfRegularJob()
+ {
+ UUID jobId = UUIDs.timeBased();
+ for (RestoreJobStatus status : RestoreJobStatus.values())
+ {
+ RestoreJob job = createTestingJob(jobId, status);
+ assertThat(job.shouldStageNow()).isEqualTo(status ==
RestoreJobStatus.STAGE_READY);
+ assertThat(job.shouldImportNow()).isEqualTo(status ==
RestoreJobStatus.IMPORT_READY);
+ }
+ }
+
+ @Test
+ void testPhaseGatesOfFastForwardJob()
+ {
+ UUID jobId = UUIDs.timeBased();
+ for (RestoreJobStatus status : RestoreJobStatus.values())
+ {
+ RestoreJob job = createFastForwardTestingJob(jobId, status);
+ assertThat(job.shouldStageNow())
+ .describedAs("Expecting a fast forward job to stage in both
CREATED and STAGE_READY status. status=" + status)
+ .isEqualTo(status == RestoreJobStatus.CREATED || status ==
RestoreJobStatus.STAGE_READY);
+ assertThat(job.shouldImportNow())
+ .describedAs("Expecting a fast forward job to import in both
STAGED and IMPORT_READY status. status=" + status)
+ .isEqualTo(status == RestoreJobStatus.STAGED || status ==
RestoreJobStatus.IMPORT_READY);
+ }
+ }
+
+ @Test
+ void testFastForwardEnabledPreservedThroughUnbuild()
+ {
+ RestoreJob job = createFastForwardTestingJob(UUIDs.timeBased(),
RestoreJobStatus.CREATED);
+ assertThat(job.fastForwardEnabled).isTrue();
+ assertThat(job.unbuild().build().fastForwardEnabled).isTrue();
+ }
+
@Test
void testCreateLocalConsistencyLevelJobWithoutLocalDcFails()
{
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/db/SidecarSchemaTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/db/SidecarSchemaTest.java
index 1cece57b..aa728510 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/db/SidecarSchemaTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/db/SidecarSchemaTest.java
@@ -140,7 +140,7 @@ public class SidecarSchemaTest
assertThat(interceptedExecStmts.get(0)).as("Create keyspace should
be executed the first")
.contains("CREATE KEYSPACE
IF NOT EXISTS sidecar_internal");
assertThat(interceptedExecStmts).as("Create table should be
executed for job table")
- .anyMatch(stmt ->
stmt.contains("CREATE TABLE IF NOT EXISTS sidecar_internal.restore_job_v6"));
+ .anyMatch(stmt ->
stmt.contains("CREATE TABLE IF NOT EXISTS sidecar_internal.restore_job_v7"));
assertThat(interceptedExecStmts).as("Create table should be
executed for slice table")
.anyMatch(stmt ->
stmt.contains("CREATE TABLE IF NOT EXISTS sidecar_internal.restore_slice_v3"));
assertThat(interceptedExecStmts).as("Create table should be
executed for range table")
@@ -163,27 +163,28 @@ public class SidecarSchemaTest
.anyMatch(stmt ->
stmt.contains("CREATE TABLE IF NOT EXISTS
sidecar_internal.cluster_ops_node_state"));
List<String> expectedPrepStatements = Arrays.asList(
- "INSERT INTO sidecar_internal.restore_job_v6 ( created_at,
job_id, keyspace_name, table_name, " +
- "job_agent, status, blob_secrets, import_options,
consistency_level, local_datacenter, local_datacenter_only, credential_type,
expire_at) " +
- "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ "INSERT INTO sidecar_internal.restore_job_v7 ( created_at,
job_id, keyspace_name, table_name, " +
+ "job_agent, status, blob_secrets, import_options,
consistency_level, local_datacenter, " +
+ "local_datacenter_only, credential_type, fast_forward_enabled,
expire_at) " +
+ "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
- "INSERT INTO sidecar_internal.restore_job_v6 ( created_at,
job_id, blob_secrets) VALUES (?, ? ,?)",
+ "INSERT INTO sidecar_internal.restore_job_v7 ( created_at,
job_id, blob_secrets) VALUES (?, ? ,?)",
- "INSERT INTO sidecar_internal.restore_job_v6 ( created_at,
job_id, status) VALUES (?, ?, ?)",
+ "INSERT INTO sidecar_internal.restore_job_v7 ( created_at,
job_id, status) VALUES (?, ?, ?)",
- "INSERT INTO sidecar_internal.restore_job_v6 ( created_at,
job_id, job_agent) VALUES (?, ?, ?)",
+ "INSERT INTO sidecar_internal.restore_job_v7 ( created_at,
job_id, job_agent) VALUES (?, ?, ?)",
- "INSERT INTO sidecar_internal.restore_job_v6 ( created_at,
job_id, expire_at) VALUES (?, ?, ?)",
+ "INSERT INTO sidecar_internal.restore_job_v7 ( created_at,
job_id, expire_at) VALUES (?, ?, ?)",
- "INSERT INTO sidecar_internal.restore_job_v6 ( created_at,
job_id, slice_count) VALUES (?, ?, ?)",
+ "INSERT INTO sidecar_internal.restore_job_v7 ( created_at,
job_id, slice_count) VALUES (?, ?, ?)",
"SELECT created_at, job_id, keyspace_name, table_name, job_agent,
status, blob_secrets, import_options, " +
- "consistency_level, local_datacenter, local_datacenter_only,
credential_type, expire_at, slice_count " +
- "FROM sidecar_internal.restore_job_v6 WHERE created_at = ? AND
job_id = ?",
+ "consistency_level, local_datacenter, local_datacenter_only,
credential_type, fast_forward_enabled, expire_at, slice_count " +
+ "FROM sidecar_internal.restore_job_v7 WHERE created_at = ? AND
job_id = ?",
"SELECT created_at, job_id, keyspace_name, table_name, job_agent,
status, blob_secrets, import_options, " +
- "consistency_level, local_datacenter, local_datacenter_only,
credential_type, expire_at, slice_count " +
- "FROM sidecar_internal.restore_job_v6 WHERE created_at = ?",
+ "consistency_level, local_datacenter, local_datacenter_only,
credential_type, fast_forward_enabled, expire_at, slice_count " +
+ "FROM sidecar_internal.restore_job_v7 WHERE created_at = ?",
"INSERT INTO sidecar_internal.restore_slice_v3 ( job_id,
bucket_id, slice_id, bucket, key, " +
"checksum, start_token, end_token, compressed_size,
uncompressed_size) " +
@@ -257,7 +258,7 @@ public class SidecarSchemaTest
.containsExactlyInAnyOrderElementsOf(expectedPrepStatements);
assertThat(sidecarSchema.isInitialized()).as("Schema is
successfully initialized").isTrue();
-
assertTableSchema(sidecarSchema.tableSchema(RestoreJobsSchema.class),
"sidecar_internal.restore_job_v6");
+
assertTableSchema(sidecarSchema.tableSchema(RestoreJobsSchema.class),
"sidecar_internal.restore_job_v7");
assertTableSchema(sidecarSchema.tableSchema(RestoreRangesSchema.class),
"sidecar_internal.restore_range_v1");
assertTableSchema(sidecarSchema.tableSchema(RestoreSlicesSchema.class),
"sidecar_internal.restore_slice_v3");
assertTableSchema(sidecarSchema.tableSchema(SidecarLeaseSchema.class),
"sidecar_internal.sidecar_lease_v1");
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandlerTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandlerTest.java
index 6ace186b..55ea8232 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandlerTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/handlers/restore/UpdateRestoreJobHandlerTest.java
@@ -232,6 +232,40 @@ class UpdateRestoreJobHandlerTest extends
BaseRestoreJobTests
assertThat(Uninterruptibles.awaitUninterruptibly(latch, 5,
TimeUnit.SECONDS)).isTrue();
}
+ @Test
+ void
testUpdateToStagedNotifiesDiscovererForFastForwardJob(VertxTestContext context)
throws Throwable
+ {
+ String jobId = "8e5799a4-d277-11ed-8d85-6916bb9b8056";
+ CountDownLatch latch = new CountDownLatch(1);
+ testRestoreJobDiscoverer.processJobNowCallback = job ->
latch.countDown();
+ mockLookupRestoreJob(id -> createTestFastForwardJobWithStatus(jobId,
RestoreJobStatus.STAGE_READY));
+ mockUpdateRestoreJob(payload ->
createTestFastForwardJobWithStatus(jobId, RestoreJobStatus.STAGED));
+ JsonObject payload = new JsonObject();
+ payload.put("status", "STAGED");
+ sendUpdateRestoreJobRequestAndVerify("ks", "table", jobId,
+ payload, context,
HttpResponseStatus.OK.code());
+ assertThat(Uninterruptibles.awaitUninterruptibly(latch, 5,
TimeUnit.SECONDS))
+ .describedAs("STAGED starts the importing phase of a fast forward job,
hence it is a phase signal")
+ .isTrue();
+ }
+
+ @Test
+ void
testUpdateToStagedDoesNotNotifyDiscovererForRegularJob(VertxTestContext
context) throws Throwable
+ {
+ String jobId = "8e5799a4-d277-11ed-8d85-6916bb9b8056";
+ CountDownLatch latch = new CountDownLatch(1);
+ testRestoreJobDiscoverer.processJobNowCallback = job ->
latch.countDown();
+ mockLookupRestoreJob(id -> createTestJobWithStatus(jobId,
RestoreJobStatus.STAGE_READY));
+ mockUpdateRestoreJob(payload -> createTestJobWithStatus(jobId,
RestoreJobStatus.STAGED));
+ JsonObject payload = new JsonObject();
+ payload.put("status", "STAGED");
+ sendUpdateRestoreJobRequestAndVerify("ks", "table", jobId,
+ payload, context,
HttpResponseStatus.OK.code());
+ assertThat(Uninterruptibles.awaitUninterruptibly(latch, 1,
TimeUnit.SECONDS))
+ .describedAs("A regular job still waits for the IMPORT_READY signal,
so STAGED is not a phase signal")
+ .isFalse();
+ }
+
private RestoreJob createTestNewJob(String jobId)
{
return RestoreJob.builder()
@@ -256,6 +290,11 @@ class UpdateRestoreJobHandlerTest extends
BaseRestoreJobTests
.build();
}
+ private RestoreJob createTestFastForwardJobWithStatus(String jobId,
RestoreJobStatus status)
+ {
+ return createTestJobWithStatus(jobId,
status).unbuild().fastForwardEnabled(true).build();
+ }
+
private JsonObject getRequestPayload()
{
JsonObject payload = new JsonObject();
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
index a0e946d6..4404f05a 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreJobDiscovererTest.java
@@ -79,6 +79,7 @@ import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -293,11 +294,87 @@ class RestoreJobDiscovererTest
verify(mockRangeAccessor, never()).create(any());
}
+ @Test
+ void testDiscoverFastForwardJobInCreatedStatus() throws Exception
+ {
+ UUID jobId = UUIDs.timeBased();
+ RestoreJob job = createTestingJob(jobId, RestoreJobStatus.CREATED,
ConsistencyLevel.QUORUM)
+ .unbuild()
+ .fastForwardEnabled(true)
+ .build();
+ mockSidecarManagedJobDiscovery(job);
+
+ executeBlocking();
+ verify(mockSliceAccessor,
times(1)).selectByJobByBucketByTokenRange(any(), anyShort(), any());
+ ArgumentCaptor<RestoreRange> restoreRangeCaptor =
ArgumentCaptor.forClass(RestoreRange.class);
+ verify(mockRangeAccessor).create(restoreRangeCaptor.capture());
+ assertThat(restoreRangeCaptor.getValue().jobId()).isEqualTo(jobId);
+
+ executeBlocking();
+ // slices of a fast forward job are uploaded progressively, hence the
discovery repeats in CREATED status
+ verify(mockSliceAccessor,
times(2)).selectByJobByBucketByTokenRange(any(), anyShort(), any());
+ }
+
+ @Test
+ void testSkipDiscoveringSlicesOfRegularJobInCreatedStatus() throws
Exception
+ {
+ mockSidecarManagedJobDiscovery(createTestingJob(UUIDs.timeBased(),
RestoreJobStatus.CREATED, ConsistencyLevel.QUORUM));
+
+ executeBlocking();
+
+ verify(mockSliceAccessor,
never()).selectByJobByBucketByTokenRange(any(), anyShort(), any());
+ verify(mockRangeAccessor, never()).create(any());
+ }
+
+ @Test
+ void testDiscoveringSlicesOfStageReadyJobIsNotRepeated() throws Exception
+ {
+ mockSidecarManagedJobDiscovery(createTestingJob(UUIDs.timeBased(),
RestoreJobStatus.STAGE_READY, ConsistencyLevel.QUORUM));
+
+ executeBlocking();
+ executeBlocking();
+
+ // STAGE_READY guarantees that all slices are uploaded, so discovering
them once is sufficient
+ verify(mockSliceAccessor,
times(1)).selectByJobByBucketByTokenRange(any(), anyShort(), any());
+ }
+
+ @Test
+ void testDiscoveringSlicesOfFastForwardJobInStagedStatusIsNotRepeated()
throws Exception
+ {
+ UUID jobId = UUIDs.timeBased();
+ RestoreJob job = createTestingJob(jobId, RestoreJobStatus.STAGED,
ConsistencyLevel.QUORUM)
+ .unbuild()
+ .fastForwardEnabled(true)
+ .build();
+ mockSidecarManagedJobDiscovery(job);
+
+ executeBlocking();
+ // on entering STAGED, the ranges are discovered and submitted, so
that the staged ones can be imported
+ verify(mockSliceAccessor,
times(1)).selectByJobByBucketByTokenRange(any(), anyShort(), any());
+
+ executeBlocking();
+ // all slices are uploaded by the time the job is STAGED, so the
expensive discovery is not repeated
+ verify(mockSliceAccessor,
times(1)).selectByJobByBucketByTokenRange(any(), anyShort(), any());
+ }
+
private UUID discoverSidecarManagedJob(boolean isJobFailed) throws
Exception
{
- when(sidecarSchema.isInitialized()).thenReturn(true);
UUID jobId = UUIDs.timeBased();
RestoreJob sidecarManagedJob = createTestingJob(jobId,
RestoreJobStatus.STAGE_READY, ConsistencyLevel.QUORUM);
+ mockSidecarManagedJobDiscovery(sidecarManagedJob);
+ if (isJobFailed)
+ {
+ doThrow(RestoreJobExceptions.ofFatal("Job failed",
mock(RestoreRange.class), null))
+ .when(mockManagers).trySubmit(any(), any(), any());
+ }
+
+ executeBlocking();
+ return jobId;
+ }
+
+ private void mockSidecarManagedJobDiscovery(RestoreJob sidecarManagedJob)
throws Exception
+ {
+ when(sidecarSchema.isInitialized()).thenReturn(true);
assertThat(sidecarManagedJob.isManagedBySidecar()).isTrue();
when(mockJobAccessor.findAllRecent(anyLong(), anyInt()))
@@ -309,18 +386,7 @@ class RestoreJobDiscovererTest
when(instanceMetadataFetcher.instance(anyInt())).thenReturn(instance);
when(mockSliceAccessor.selectByJobByBucketByTokenRange(any(),
anyShort(), any()))
.thenReturn(Collections.singletonList(RestoreSliceTest.createTestingSlice(sidecarManagedJob,
"sliceId", 0, 10)));
- if (isJobFailed)
- {
- doThrow(RestoreJobExceptions.ofFatal("Job failed",
mock(RestoreRange.class), null))
- .when(mockManagers).trySubmit(any(), any(), any());
- }
- else
- {
- when(mockManagers.trySubmit(any(), any(),
any())).thenReturn(RestoreJobProgressTracker.Status.CREATED);
- }
-
- executeBlocking();
- return jobId;
+ when(mockManagers.trySubmit(any(), any(),
any())).thenReturn(RestoreJobProgressTracker.Status.CREATED);
}
@Test
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreProcessorTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreProcessorTest.java
index 2c1b65ea..6b347354 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreProcessorTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreProcessorTest.java
@@ -30,12 +30,14 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import com.codahale.metrics.SharedMetricRegistries;
+import com.datastax.driver.core.utils.UUIDs;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import org.apache.cassandra.sidecar.ExecutorPoolsHelper;
import org.apache.cassandra.sidecar.TestModule;
import org.apache.cassandra.sidecar.TestResourceReaper;
import org.apache.cassandra.sidecar.cluster.locator.LocalTokenRangesProvider;
+import org.apache.cassandra.sidecar.common.data.RestoreJobStatus;
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;
@@ -43,6 +45,8 @@ import
org.apache.cassandra.sidecar.config.SidecarConfiguration;
import org.apache.cassandra.sidecar.config.yaml.RestoreJobConfigurationImpl;
import org.apache.cassandra.sidecar.config.yaml.SidecarConfigurationImpl;
import org.apache.cassandra.sidecar.coordination.ClusterLease;
+import org.apache.cassandra.sidecar.db.RestoreJob;
+import org.apache.cassandra.sidecar.db.RestoreJobTest;
import org.apache.cassandra.sidecar.db.RestoreRange;
import org.apache.cassandra.sidecar.db.RestoreRangeDatabaseAccessor;
import org.apache.cassandra.sidecar.db.schema.SidecarSchema;
@@ -63,6 +67,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyDouble;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
@@ -300,6 +305,34 @@ class RestoreProcessorTest
});
}
+ @Test
+ void testStagedRangeWaitsForImportSignal()
+ {
+ CountDownLatch readyToFinish = new CountDownLatch(1);
+ RestoreRange range = stagedRange(readyToFinish,
RestoreJobTest.createTestingJob(UUIDs.timeBased(), RestoreJobStatus.STAGED));
+ processor.submit(range);
+ processor.execute(Promise.promise());
+ assertThat(processor.activeTasks())
+ .describedAs("A staged range of a regular job should wait for the
IMPORT_READY signal, " +
+ "even when the job has entered STAGED status")
+ .isZero();
+ readyToFinish.countDown();
+ }
+
+ @Test
+ void testStagedRangeOfFastForwardJobImportsOnceStaged()
+ {
+ CountDownLatch readyToFinish = new CountDownLatch(1);
+ RestoreRange range = stagedRange(readyToFinish,
RestoreJobTest.createFastForwardTestingJob(UUIDs.timeBased(),
RestoreJobStatus.STAGED));
+ processor.submit(range);
+ processor.execute(Promise.promise());
+ assertThat(processor.activeTasks())
+ .describedAs("A staged range of a fast forward job should import as
soon as the job enters STAGED status")
+ .isOne();
+ readyToFinish.countDown();
+ loopAssert(3, () -> assertThat(processor.activeTasks()).isZero());
+ }
+
private InstanceMetrics instanceMetrics()
{
return new InstanceMetricsImpl(registry(1));
@@ -310,6 +343,15 @@ class RestoreProcessorTest
return mockSlowRestoreRange(latch, System::nanoTime);
}
+ // a range that has completed its staging phase and belongs to the given
restore job
+ private RestoreRange stagedRange(CountDownLatch latch, RestoreJob job)
+ {
+ RestoreRange range = mockSlowRestoreRange(latch);
+ doReturn(true).when(range).hasStaged();
+ doReturn(job).when(range).job();
+ return range;
+ }
+
private RestoreRange mockSlowRestoreRange(CountDownLatch latch,
RestoreRange mockRange)
{
return mockSlowRestoreRange(latch, System::nanoTime, mockRange);
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreRangeTaskTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreRangeTaskTest.java
index 4a8a170c..4e8d4e85 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreRangeTaskTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/restore/RestoreRangeTaskTest.java
@@ -342,6 +342,69 @@ class RestoreRangeTaskTest
assertThat(mockRange.hasImported()).isTrue();
}
+ @Test
+ void testStagingOfFastForwardJobInCreatedStatus()
+ {
+ // A fast forward job stages its slices ahead of the STAGE_READY
signal, i.e. while it is still CREATED
+ RestoreJob job =
spy(RestoreJobTest.createFastForwardTestingJob(UUIDs.timeBased(),
RestoreJobStatus.CREATED));
+ doReturn(true).when(job).isManagedBySidecar();
+ doReturn(job).when(mockRange).job();
+ doReturn(Paths.get("nonexist")).when(mockRange).stagedObjectPath();
+ HeadObjectResponse headObjectResponse = mock(HeadObjectResponse.class);
+ when(headObjectResponse.contentLength()).thenReturn(1L);
+
when(mockStorageClient.objectExists(mockRange)).thenReturn(CompletableFuture.completedFuture(headObjectResponse));
+ when(mockStorageClient.downloadObjectIfAbsent(eq(mockRange),
any(TaskExecutorPool.class)))
+ .thenReturn(Future.succeededFuture(new File(".")));
+ RestoreRangeTask task = createTask(mockRange, job);
+
+ Promise<RestoreRange> promise = Promise.promise();
+ task.handle(promise);
+ getBlocking(promise.future()); // no error is thrown
+
+ verify(mockRange, times(1)).completeStagePhase();
+ verify(mockRange, times(0)).completeImportPhase(); // should not be
called in this phase
+ verify(mockRangeAccessor, times(1)).updateStatus(mockRange);
+ assertThat(mockRange.hasStaged()).isTrue();
+ assertThat(mockRange.hasImported()).isFalse();
+ }
+
+ @Test
+ void testImportPhaseOfFastForwardJobInStagedStatus()
+ {
+ // A fast forward job imports its staged slices ahead of the
IMPORT_READY signal, i.e. once it is STAGED
+ RestoreJob job =
spy(RestoreJobTest.createFastForwardTestingJob(UUIDs.timeBased(),
RestoreJobStatus.STAGED));
+ doReturn(true).when(job).isManagedBySidecar();
+ doReturn(job).when(mockRange).job();
+ doReturn(true).when(mockRange).hasStaged();
+ RestoreRangeTask task = createTask(mockRange, job);
+
+ Promise<RestoreRange> promise = Promise.promise();
+ task.handle(promise);
+ getBlocking(promise.future()); // no error is thrown
+
+ verify(mockRange, times(0)).completeStagePhase(); // should not be
called in the phase
+ verify(mockRange, times(1)).completeImportPhase();
+ verify(mockRangeAccessor, times(1)).updateStatus(mockRange);
+ assertThat(mockRange.hasImported()).isTrue();
+ }
+
+ @Test
+ void testFailsWhenJobIsNotReadyToStageOrImport()
+ {
+ // Without fast forward, no phase can start while the job is in
CREATED status
+ RestoreJob job =
spy(RestoreJobTest.createTestingJob(UUIDs.timeBased(),
RestoreJobStatus.CREATED));
+ doReturn(true).when(job).isManagedBySidecar();
+ doReturn(job).when(mockRange).job();
+ RestoreRangeTask task = createTask(mockRange, job);
+
+ Promise<RestoreRange> promise = Promise.promise();
+ task.handle(promise);
+ assertThatThrownBy(() -> getBlocking(promise.future()))
+ .hasMessageContaining("Unexpected restore job status");
+ verify(mockRange, times(0)).completeStagePhase();
+ verify(mockRange, times(0)).completeImportPhase();
+ }
+
@Test
void testImportSSTables(@TempDir Path testFolder)
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
