From c50d0dedeedc2a60dacead21748efe715a477617 Mon Sep 17 00:00:00 2001 From: Arjun Ashok Date: Mon, 3 Aug 2026 10:41:51 -0700 Subject: [PATCH] CASSSIDECAR-465: Improve HTTP 429 handling across sidecar server and client Patch by Arjun Ashok for CASSSIDECAR-465 --- CHANGES.txt | 1 + .../sidecar/client/RequestExecutor.java | 5 +- .../client/retry/BasicRetryPolicy.java | 14 +++ .../sidecar/client/RequestExecutorTest.java | 91 +++++++++++++++++++ .../client/retry/BasicRetryPolicyTest.java | 21 ++++- conf/sidecar.yaml | 1 + docs/src/user.adoc | 1 + .../config/SSTableUploadConfiguration.java | 11 +++ .../yaml/SSTableUploadConfigurationImpl.java | 44 ++++++++- .../sstableuploads/SSTableUploadHandler.java | 2 + .../SSTableUploadConfigurationImplTest.java | 18 ++++ .../BaseUploadsHandlerTest.java | 1 + .../SSTableUploadHandlerTest.java | 7 +- 13 files changed, 208 insertions(+), 9 deletions(-) create mode 100644 client/src/test/java/org/apache/cassandra/sidecar/client/RequestExecutorTest.java diff --git a/CHANGES.txt b/CHANGES.txt index d9caa34bb..57bcc1f03 100644 --- a/CHANGES.txt +++ b/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) diff --git a/client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java b/client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java index 353b652a6..aadd76522 100644 --- a/client/src/main/java/org/apache/cassandra/sidecar/client/RequestExecutor.java +++ b/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(); + } } } diff --git a/client/src/main/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicy.java b/client/src/main/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicy.java index fc6edeae8..dbf6a5778 100644 --- a/client/src/main/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicy.java +++ b/client/src/main/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicy.java @@ -161,6 +161,20 @@ public void onResponse(CompletableFuture 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())) diff --git a/client/src/test/java/org/apache/cassandra/sidecar/client/RequestExecutorTest.java b/client/src/test/java/org/apache/cassandra/sidecar/client/RequestExecutorTest.java new file mode 100644 index 000000000..1a62ed691 --- /dev/null +++ b/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); + } +} diff --git a/client/src/test/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicyTest.java b/client/src/test/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicyTest.java index f057b140b..a85eeac74 100644 --- a/client/src/test/java/org/apache/cassandra/sidecar/client/retry/BasicRetryPolicyTest.java +++ b/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 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); } diff --git a/conf/sidecar.yaml b/conf/sidecar.yaml index 37cbf7394..fea8ad3e1 100644 --- a/conf/sidecar.yaml +++ b/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. diff --git a/docs/src/user.adoc b/docs/src/user.adoc index ec6f65965..e417251ea 100644 --- a/docs/src/user.adoc +++ b/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. diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/SSTableUploadConfiguration.java b/server/src/main/java/org/apache/cassandra/sidecar/config/SSTableUploadConfiguration.java index f36620c02..8607316e2 100644 --- a/server/src/main/java/org/apache/cassandra/sidecar/config/SSTableUploadConfiguration.java +++ b/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; + } } diff --git a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/SSTableUploadConfigurationImpl.java b/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/SSTableUploadConfigurationImpl.java index 2f69352c7..704f335d8 100644 --- a/server/src/main/java/org/apache/cassandra/sidecar/config/yaml/SSTableUploadConfigurationImpl.java +++ b/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; + } } diff --git a/server/src/main/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandler.java b/server/src/main/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandler.java index 345eea72b..3a9a55551 100644 --- a/server/src/main/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandler.java +++ b/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; } diff --git a/server/src/test/java/org/apache/cassandra/sidecar/config/yaml/SSTableUploadConfigurationImplTest.java b/server/src/test/java/org/apache/cassandra/sidecar/config/yaml/SSTableUploadConfigurationImplTest.java index 9cca38fc6..4fc7cea0e 100644 --- a/server/src/test/java/org/apache/cassandra/sidecar/config/yaml/SSTableUploadConfigurationImplTest.java +++ b/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); + } } diff --git a/server/src/test/java/org/apache/cassandra/sidecar/handlers/sstableuploads/BaseUploadsHandlerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/handlers/sstableuploads/BaseUploadsHandlerTest.java index 64b4ebc0d..f59ae48fd 100644 --- a/server/src/test/java/org/apache/cassandra/sidecar/handlers/sstableuploads/BaseUploadsHandlerTest.java +++ b/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()) diff --git a/server/src/test/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandlerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandlerTest.java index 090bbe325..cd191c102 100644 --- a/server/src/test/java/org/apache/cassandra/sidecar/handlers/sstableuploads/SSTableUploadHandlerTest.java +++ b/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