CASSSIDECAR-463: Add fast_forward_enabled flag to enable eager pipeli…

CHANGES.txt

@@ -13,6 +13,7 @@
 
 0.4.0
 -----
+ * Add fast_forward_enabled flag to enable eager pipelining of staging and importing in coordinated writes (CASSSIDECAR-463)
  * Fix restore job failing with "Keyspace does not exist" for quoted mixed-case keyspace names (CASSSIDECAR-475)
  * Split RestoreJobDiscoverer into a fast status-check loop and a slow slice-discovery loop (CASSSIDECAR-462)
  * Ability to load Cassandra connection secrets from filesystem (CASSSIDECAR-407)

NEWS.txt

@@ -21,6 +21,9 @@ New features
 
 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-415, a new internal table (restore_job_v6) is created to support IAM credential
       tracking for restore jobs. Spark jobs performing S3 restores should be paused during sidecar
       upgrade to ensure in-flight restore job state is not lost.
@@ -31,6 +34,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.
     - IAM instance profile credential support for S3 restore jobs (CASSSIDECAR-415). Sidecar can now
       resolve AWS credentials automatically from the platform (IMDS/ECS/EKS) without requiring key
       material to be passed by the caller. Set credentialType to IAM when creating a restore job to

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";

client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/CreateRestoreJobRequestPayload.java

@@ -42,6 +42,7 @@
 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;
@@ -75,6 +76,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 +111,9 @@ public static Builder builder(RestoreJobSecrets secrets, long expireAtInMillis)
      * @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 +124,8 @@ public CreateRestoreJobRequestPayload(@JsonProperty(JOB_ID) UUID jobId,
                                           @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 +148,11 @@ public CreateRestoreJobRequestPayload(@JsonProperty(JOB_ID) UUID jobId,
         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 +253,15 @@ public ConsistencyConfig consistencyConfig()
         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 +281,7 @@ public String toString()
                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 +300,7 @@ public static class Builder implements DataObjectBuilder<Builder, CreateRestoreJ
         private ConsistencyLevel consistencyLevel = null;
         private String localDc = null;
         private boolean localDatacenterOnly = false;
+        private boolean fastForwardEnabled = false;
 
         Builder(RestoreJobSecrets secrets, long expireAtInMillis)
         {
@@ -324,6 +346,21 @@ public Builder restoreToLocalDatacenterOnly(boolean localDatacenterOnly)
             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 +387,8 @@ private CreateRestoreJobRequestPayload(Builder builder)
              builder.expireAtInMillis,
              nameOrNull(builder.consistencyLevel),
              builder.localDc,
-             builder.localDatacenterOnly);
+             builder.localDatacenterOnly,
+             builder.fastForwardEnabled);
     }
 
     private static String nameOrNull(ConsistencyLevel cl)

client-common/src/test/java/org/apache/cassandra/sidecar/common/request/CreateRestoreJobRequestPayloadTest.java

@@ -39,6 +39,7 @@
 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;
@@ -398,4 +399,48 @@ void testPartialWriteCredentialsAreRejected() throws JsonProcessingException
         .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");
+    }
 }