CASSSIDECAR-465: Improve HTTP 429 handling across sidecar server and …

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * Improve HTTP 429 handling across sidecar server and client (CASSSIDECAR-465)
  * Wire CDC configs in configs table to SidecarCdcOptions/SidecarStatePersister (CASSSIDECAR-483)
  * Implement durable operational job tracker (CASSSIDECAR-374)
  * Remove filesystem path from Http response (CASSSIDECAR-477)

client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java

@@ -372,6 +372,9 @@ protected void schedule(long delayMillis, Runnable runnable)
         {
             singleThreadExecutorService.schedule(runnable, delayMillis, TimeUnit.MILLISECONDS);
         }
-        runnable.run();
+        else
+        {
+            runnable.run();
+        }
     }
 }

client/src/main/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicy.java

@@ -161,6 +161,20 @@ public void onResponse(CompletableFuture<HttpResponse> responseFuture,
             return;
         }
 
+        if (response.statusCode() == HttpResponseStatus.TOO_MANY_REQUESTS.code())
+        {
+            if (canRetryOnADifferentHost)
+            {
+                retryImmediately(responseFuture, request, response, retryAction, attempts);
+            }
+            else
+            {
+                retry(responseFuture, request, response, retryAction, attempts,
+                      maybeParseRetryAfterOrDefault(response, attempts), null);
+            }
+            return;
+        }
+
         // 4xx Client Errors - 5xx Server Errors
         if (HttpStatusClass.CLIENT_ERROR.contains(response.statusCode()) ||
             HttpStatusClass.SERVER_ERROR.contains(response.statusCode()))

client/src/test/java/org/apache/cassandra/sidecar/client/RequestExecutorTest.java

@@ -0,0 +1,91 @@
+/*
+ * 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.client;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Unit tests for {@link RequestExecutor#schedule(long, Runnable)}.
+ */
+class RequestExecutorTest
+{
+    private RequestExecutor executor;
+
+    @BeforeEach
+    void setup()
+    {
+        executor = new RequestExecutor(mock(HttpClient.class));
+    }
+
+    @AfterEach
+    void tearDown() throws Exception
+    {
+        executor.close();
+    }
+
+    @Test
+    void scheduleDoesNotRunImmediatelyWhenDelayIsPositive()
+    {
+        AtomicInteger invocationCount = new AtomicInteger(0);
+        executor.schedule(200, invocationCount::incrementAndGet);
+        // must not have run synchronously -- it was scheduled 200ms in the future
+        assertThat(invocationCount.get()).isEqualTo(0);
+    }
+
+    @Test
+    void scheduleRunsExactlyOnceAfterThePositiveDelayElapses() throws InterruptedException
+    {
+        AtomicInteger invocationCount = new AtomicInteger(0);
+        CountDownLatch ran = new CountDownLatch(1);
+        executor.schedule(50, () -> {
+            invocationCount.incrementAndGet();
+            ran.countDown();
+        });
+
+        assertThat(ran.await(1, TimeUnit.SECONDS)).isTrue();
+        // give a buggy duplicate invocation (immediate-fire) time to have already happened by now
+        Thread.sleep(200);
+        assertThat(invocationCount.get()).isEqualTo(1);
+    }
+
+    @Test
+    void scheduleRunsImmediatelyWhenDelayIsZero()
+    {
+        AtomicInteger invocationCount = new AtomicInteger(0);
+        executor.schedule(0, invocationCount::incrementAndGet);
+        assertThat(invocationCount.get()).isEqualTo(1);
+    }
+
+    @Test
+    void scheduleRunsImmediatelyWhenDelayIsNegative()
+    {
+        AtomicInteger invocationCount = new AtomicInteger(0);
+        executor.schedule(-1, invocationCount::incrementAndGet);
+        assertThat(invocationCount.get()).isEqualTo(1);
+    }
+}

client/src/test/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicyTest.java

@@ -49,6 +49,7 @@
 import static io.netty.handler.codec.http.HttpResponseStatus.NOT_IMPLEMENTED;
 import static io.netty.handler.codec.http.HttpResponseStatus.OK;
 import static io.netty.handler.codec.http.HttpResponseStatus.SERVICE_UNAVAILABLE;
+import static io.netty.handler.codec.http.HttpResponseStatus.TOO_MANY_REQUESTS;
 import static org.apache.cassandra.sidecar.common.http.SidecarHttpResponseStatus.CHECKSUM_MISMATCH;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
@@ -167,6 +168,23 @@ void testRetriesWithChecksumMismatchStatusCode(boolean canRetryOnADifferentHost)
         testWithRetries(mockRequest, mockResponse, null, 5, 300, canRetryOnADifferentHost);
     }
 
+    @ParameterizedTest(name = "{index} => canRetryOnADifferentHost={0}")
+    @ValueSource(booleans = { true, false })
+    void testRetriesWithTooManyRequestsStatusCode(boolean canRetryOnADifferentHost)
+    {
+        when(mockResponse.statusCode()).thenReturn(TOO_MANY_REQUESTS.code());
+        testWithRetries(mockRequest, mockResponse, null, 5, 300, canRetryOnADifferentHost);
+    }
+
+    @Test
+    void testRetriesWithTooManyRequestsStatusCodeWithRetryAfterHeader()
+    {
+        when(mockResponse.statusCode()).thenReturn(TOO_MANY_REQUESTS.code());
+        headersMap.put("Retry-After", Collections.singletonList("5")); // 5 seconds -> 5,000 millis
+        testWithRetries(mockRequest, mockResponse, null, 5, 1000, 5000, false);
+        testWithRetries(mockRequest, mockResponse, null, 5, 1000, 0, true);
+    }
+
     @Test
     void testRetriesWithServiceUnavailableStatusCodeWithRetryAfterHeader()
     {
@@ -217,7 +235,8 @@ private static Stream<Arguments> clientStatusCodeArguments()
     {
         return IntStream.range(400, 500)
                         .filter(statusCode -> statusCode != NOT_FOUND.code()
-                                              && statusCode != CHECKSUM_MISMATCH.code())
+                                              && statusCode != CHECKSUM_MISMATCH.code()
+                                              && statusCode != TOO_MANY_REQUESTS.code())
                         .boxed()
                         .map(Arguments::of);
     }

conf/sidecar.yaml

@@ -108,6 +108,7 @@ sidecar:
   sstable_upload:
     concurrent_upload_limit: 80
     min_free_space_percent: 10
+    retry_after_seconds: 1                        # suggested wait, in seconds, sent to clients via the Retry-After header when the concurrent upload limit is exceeded
     # file_permissions: "rw-r--r--" # when not specified, the default file permissions are owner read & write, group & others read
   # The maximum allowable time skew between the server and the client.
   # Resolution is in minutes. The minimum configurable value is 1 minute.

docs/src/user.adoc

@@ -235,6 +235,7 @@ The `sidecar` section of the `sidecar.yaml` file is used to configure the Cassan
 * `sstable_upload`: This subsection manages the configuration for SSTable component uploads by Cassandra Sidecar.
 ** `concurrent_upload_limit`: This defines the maximum number of SSTable components that can be uploaded concurrently. By default, `80`.
 ** `min_free_space_percent`: This defines the minimum percentage of available disk required for SSTable component uploads to proceed. By default, `10`.
+** `retry_after_seconds`: This defines the number of seconds sent in the `Retry-After` header when a request is rejected for exceeding `concurrent_upload_limit`. By default, `1`.
 * `allowable_time_skew`: This defines the maximum allowable time skew between Cassandra Sidecar and clients of Cassandra Sidecar. The minimum resolution is defined in minutes. By default, this is set to 1 hour.
 * `sstable_import`: This subsection manages the configuration for Cassandra Sidecar's SSTable import functionality. The following properties are defined:
 ** `execute_interval`: The interval at which Cassandra Sidecar will execute SSTable import tasks.

server/src/main/java/org/apache/cassandra/sidecar/config/SSTableUploadConfiguration.java

@@ -45,4 +45,15 @@ public interface SSTableUploadConfiguration
      * @return the String representation of a set of posix file permissions used during an SSTable file upload
      */
     String filePermissions();
+
+    /**
+     * @return the number of seconds a client should wait before retrying, when this service is rejecting requests
+     * because the concurrent upload limit has been reached
+     */
+    default int retryAfterSeconds()
+    {
+        // default kept here (matching SSTableUploadConfigurationImpl.DEFAULT_RETRY_AFTER_SECONDS) so that adding
+        // this method doesn't break pre-existing implementers of this interface
+        return 1;
+    }
 }

server/src/main/java/org/apache/cassandra/sidecar/config/yaml/SSTableUploadConfigurationImpl.java

@@ -37,6 +37,9 @@ public class SSTableUploadConfigurationImpl implements SSTableUploadConfiguratio
     public static final String FILE_PERMISSIONS_PROPERTY = "file_permissions";
     public static final String DEFAULT_FILE_PERMISSIONS = "rw-r--r--";
 
+    public static final String RETRY_AFTER_SECONDS_PROPERTY = "retry_after_seconds";
+    public static final int DEFAULT_RETRY_AFTER_SECONDS = 1;
+
     @JsonProperty(value = CONCURRENT_UPLOAD_LIMIT_PROPERTY)
     protected final int concurrentUploadsLimit;
 
@@ -45,35 +48,56 @@ public class SSTableUploadConfigurationImpl implements SSTableUploadConfiguratio
 
     protected String filePermissions;
 
+    @JsonProperty(value = RETRY_AFTER_SECONDS_PROPERTY)
+    protected final int retryAfterSeconds;
+
     public SSTableUploadConfigurationImpl()
     {
-        this(DEFAULT_CONCURRENT_UPLOAD_LIMIT, DEFAULT_MIN_FREE_SPACE_PERCENT, DEFAULT_FILE_PERMISSIONS);
+        this(DEFAULT_CONCURRENT_UPLOAD_LIMIT, DEFAULT_MIN_FREE_SPACE_PERCENT, DEFAULT_FILE_PERMISSIONS,
+             DEFAULT_RETRY_AFTER_SECONDS);
     }
 
     public SSTableUploadConfigurationImpl(int concurrentUploadsLimit)
     {
-        this(concurrentUploadsLimit, DEFAULT_MIN_FREE_SPACE_PERCENT, DEFAULT_FILE_PERMISSIONS);
+        this(concurrentUploadsLimit, DEFAULT_MIN_FREE_SPACE_PERCENT, DEFAULT_FILE_PERMISSIONS,
+             DEFAULT_RETRY_AFTER_SECONDS);
     }
 
     public SSTableUploadConfigurationImpl(float minimumSpacePercentageRequired)
     {
-        this(DEFAULT_CONCURRENT_UPLOAD_LIMIT, minimumSpacePercentageRequired, DEFAULT_FILE_PERMISSIONS);
+        this(DEFAULT_CONCURRENT_UPLOAD_LIMIT, minimumSpacePercentageRequired, DEFAULT_FILE_PERMISSIONS,
+             DEFAULT_RETRY_AFTER_SECONDS);
     }
 
     public SSTableUploadConfigurationImpl(String filePermissions)
     {
-        this(DEFAULT_CONCURRENT_UPLOAD_LIMIT, DEFAULT_MIN_FREE_SPACE_PERCENT, filePermissions);
+        this(DEFAULT_CONCURRENT_UPLOAD_LIMIT, DEFAULT_MIN_FREE_SPACE_PERCENT, filePermissions,
+             DEFAULT_RETRY_AFTER_SECONDS);
     }
 
     public SSTableUploadConfigurationImpl(int concurrentUploadsLimit,
                                           float minimumSpacePercentageRequired,
-                                          String filePermissions)
+                                          String filePermissions,
+                                          int retryAfterSeconds)
     {
         this.concurrentUploadsLimit = concurrentUploadsLimit;
         this.minimumSpacePercentageRequired = minimumSpacePercentageRequired;
+        this.retryAfterSeconds = retryAfterSeconds;
         setFilePermissions(filePermissions);
     }
 
+    /**
+     * @deprecated use {@link #SSTableUploadConfigurationImpl(int, float, String, int)} instead; kept for
+     * backwards compatibility with callers built against the pre-{@code retryAfterSeconds} constructor
+     */
+    @Deprecated
+    public SSTableUploadConfigurationImpl(int concurrentUploadsLimit,
+                                          float minimumSpacePercentageRequired,
+                                          String filePermissions)
+    {
+        this(concurrentUploadsLimit, minimumSpacePercentageRequired, filePermissions, DEFAULT_RETRY_AFTER_SECONDS);
+    }
+
     /**
      * {@inheritDoc}
      */
@@ -126,4 +150,14 @@ public void setFilePermissions(String filePermissions)
             this.filePermissions = null;
         }
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonProperty(value = RETRY_AFTER_SECONDS_PROPERTY)
+    public int retryAfterSeconds()
+    {
+        return retryAfterSeconds;
+    }
 }

server/src/main/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandler.java

@@ -26,6 +26,7 @@
 import com.datastax.driver.core.Metadata;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
@@ -139,6 +140,7 @@ public void handleInternal(RoutingContext context,
         {
             String message = String.format("Concurrent upload limit (%d) exceeded", limiter.limit());
             instanceMetrics.uploadSSTable().throttled.metric.update(1);
+            context.response().putHeader(HttpHeaderNames.RETRY_AFTER, String.valueOf(configuration.retryAfterSeconds()));
             context.fail(wrapHttpException(HttpResponseStatus.TOO_MANY_REQUESTS, message));
             return;
         }

server/src/test/java/org/apache/cassandra/sidecar/config/yaml/SSTableUploadConfigurationImplTest.java

@@ -53,4 +53,22 @@ void testValidFilePermission(String value)
         SSTableUploadConfigurationImpl config = new SSTableUploadConfigurationImpl(value);
         assertThat(config.filePermissions()).isEqualTo(value);
     }
+
+    @Test
+    void testDefaultRetryAfterSeconds()
+    {
+        SSTableUploadConfigurationImpl config = new SSTableUploadConfigurationImpl();
+        assertThat(config.retryAfterSeconds()).isEqualTo(SSTableUploadConfigurationImpl.DEFAULT_RETRY_AFTER_SECONDS);
+    }
+
+    @Test
+    @SuppressWarnings("deprecation")
+    void testDeprecatedThreeArgConstructorDefaultsRetryAfterSeconds()
+    {
+        SSTableUploadConfigurationImpl config = new SSTableUploadConfigurationImpl(42, 15F, "rwxr--r--");
+        assertThat(config.concurrentUploadsLimit()).isEqualTo(42);
+        assertThat(config.minimumSpacePercentageRequired()).isEqualTo(15F);
+        assertThat(config.filePermissions()).isEqualTo("rwxr--r--");
+        assertThat(config.retryAfterSeconds()).isEqualTo(SSTableUploadConfigurationImpl.DEFAULT_RETRY_AFTER_SECONDS);
+    }
 }

server/src/test/java/org/apache/cassandra/sidecar/handlers/sstableuploads/BaseUploadsHandlerTest.java

@@ -110,6 +110,7 @@ void setup() throws InterruptedException, IOException
         mockSSTableUploadConfiguration = mock(SSTableUploadConfiguration.class);
         when(mockSSTableUploadConfiguration.concurrentUploadsLimit()).thenReturn(3);
         when(mockSSTableUploadConfiguration.minimumSpacePercentageRequired()).thenReturn(0F);
+        when(mockSSTableUploadConfiguration.retryAfterSeconds()).thenReturn(1);
         trafficShapingConfiguration = mock(TrafficShapingConfiguration.class);
         when(trafficShapingConfiguration.inboundGlobalBandwidthBytesPerSecond()).thenReturn(512 * 1024L);
         when(trafficShapingConfiguration.outboundGlobalBandwidthBytesPerSecond())

server/src/test/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandlerTest.java

@@ -244,9 +244,12 @@ void testConcurrentUploadLimitExceeded(VertxTestContext context) throws IOExcept
         when(mockSSTableUploadConfiguration.concurrentUploadsLimit()).thenReturn(0);
 
         UUID uploadId = UUID.randomUUID();
-        sendUploadRequestAndVerify(context, uploadId, "ks", "tbl", "without-md5-Dataa.db", null,
+        sendUploadRequestAndVerify(null, context, uploadId.toString(), "ks", "tbl", "without-md5-Dataa.db", null,
                                    Files.size(Paths.get(FILE_TO_BE_UPLOADED)),
-                                   HttpResponseStatus.TOO_MANY_REQUESTS.code(), false);
+                                   HttpResponseStatus.TOO_MANY_REQUESTS.code(), false,
+                                   response -> assertThat(response.getHeader(HttpHeaderNames.RETRY_AFTER.toString()))
+                                              .isEqualTo("1"),
+                                   FILE_TO_BE_UPLOADED);
     }
 
     @Test