From 4232fbe6215ac3fe2b01328e3b8263402d23ce6d Mon Sep 17 00:00:00 2001 From: Robin Han Date: Mon, 12 Oct 2026 22:05:43 +0800 Subject: [PATCH 1/2] feat(s3stream): support per-bucket CRT S3 clients --- devkit/README.md | 16 ++ .../stream/s3/operator/AwsObjectStorage.java | 197 ++++++++++++++++-- .../automq/stream/s3/operator/BucketURI.java | 3 + .../s3/operator/AwsObjectStorageTest.java | 167 +++++++++++++++ 4 files changed, 370 insertions(+), 13 deletions(-) diff --git a/devkit/README.md b/devkit/README.md index cdabc731a9..46eb55b0d5 100644 --- a/devkit/README.md +++ b/devkit/README.md @@ -426,6 +426,22 @@ num.partitions=4 log.retention.hours=24 ``` +**Select the S3 SDK per bucket** — add camel-case options to each data, ops, or WAL bucket URI: + +```properties +s3.data.buckets=0@s3://automq-data?region=us-east-1&endpoint=http://minio:9000&pathStyle=true&sdkImpl=crt&connectionTimeoutMs=2000&numRetries=2&maxConcurrency=-1&crtTargetThroughputInGbps=10&crtMinimumThroughputInBps=33554432&crtMinimumThroughputTimeoutSeconds=2&crtMaxNativeMemoryLimitInBytes=-1 +``` + +Omitting `sdkImpl` selects Netty. `connectionTimeoutMs`, `numRetries` (excluding the initial attempt), +and `maxConcurrency` apply to both clients. `maxConcurrency=-1` inherits AutoMQ's existing client concurrency +without changing its inflight quotas. CRT-only options are ignored by Netty; omitted options use the values +in the example. `crtMaxNativeMemoryLimitInBytes=-1` leaves native memory budgeting to CRT. +`crtMinimumThroughputTimeoutSeconds` must be at least two. The throughput target is per client and is not +a bandwidth limiter. Existing `apiCallTimeoutMs` and `apiCallAttemptTimeoutMs` remain supported by Netty. +CRT automatic multipart uploads are disabled with a fixed `Long.MAX_VALUE` threshold; AutoMQ's explicit +multipart uploads remain enabled, preserving its request-level precomputed checksums. +Configure each `s3.ops.buckets` and `s3.wal.path` URI separately; settings do not propagate between buckets. + **Add a new feature** — create `config/features/my-feature.properties`: ```properties my.setting=value diff --git a/s3stream/src/main/java/com/automq/stream/s3/operator/AwsObjectStorage.java b/s3stream/src/main/java/com/automq/stream/s3/operator/AwsObjectStorage.java index dc9db45ccd..b8fb24fbac 100644 --- a/s3stream/src/main/java/com/automq/stream/s3/operator/AwsObjectStorage.java +++ b/s3stream/src/main/java/com/automq/stream/s3/operator/AwsObjectStorage.java @@ -66,13 +66,13 @@ import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration; import software.amazon.awssdk.core.exception.ApiCallAttemptTimeoutException; import software.amazon.awssdk.core.exception.SdkClientException; -import software.amazon.awssdk.core.retry.RetryMode; import software.amazon.awssdk.http.HttpStatusCode; import software.amazon.awssdk.http.async.SdkAsyncHttpClient; import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.s3.S3AsyncClient; import software.amazon.awssdk.services.s3.S3AsyncClientBuilder; +import software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder; import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm; import software.amazon.awssdk.services.s3.model.ChecksumMode; import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest; @@ -143,10 +143,8 @@ public AwsObjectStorage(BucketURI bucketURI, Map tagging, } this.checksumAlgorithm = checksumAlgorithm; - long apiCallTimeoutMs = Long.parseLong(bucketURI.extensionString(BucketURI.API_CALL_TIMEOUT_KEY, "30000")); - long apiCallAttemptTimeoutMs = Long.parseLong(bucketURI.extensionString(BucketURI.API_CALL_ATTEMPT_TIMEOUT_KEY, "10000")); - - Supplier clientSupplier = () -> newS3Client(bucketURI.endpoint(), bucketURI.region(), bucketURI.extensionBool(PATH_STYLE_KEY, false), credentialsProviders, getMaxObjectStorageConcurrency(), apiCallTimeoutMs, apiCallAttemptTimeoutMs); + Configuration configuration = Configuration.fromBucketURI(bucketURI); + Supplier clientSupplier = () -> newS3Client(bucketURI, credentialsProviders, configuration); this.writeS3Client = clientSupplier.get(); this.readS3Client = readWriteIsolate ? clientSupplier.get() : writeS3Client; } @@ -495,8 +493,27 @@ private String range(long start, long end) { return "bytes=" + start + "-" + (end - 1); } - protected S3AsyncClient newS3Client(String endpoint, String region, boolean forcePathStyle, - List credentialsProviders, int maxConcurrency, long apiCallTimeoutMs, long apiCallAttemptTimeoutMs) { + protected S3AsyncClient newS3Client(BucketURI uri, List credentialsProviders, + Configuration configuration) { + String endpoint = uri.endpoint(); + String region = uri.region(); + boolean forcePathStyle = uri.extensionBool(PATH_STYLE_KEY, false); + int maxConcurrency = getMaxObjectStorageConcurrency(); + if ("crt".equals(configuration.sdkImplementation())) { + S3CrtAsyncClientBuilder crtBuilder = S3AsyncClient.crtBuilder() + .region(Region.of(region)) + .credentialsProvider(newCredentialsProviderChain(credentialsProviders)) + .forcePathStyle(forcePathStyle) + .checksumValidationEnabled(false); + configuration.configureCrtBuilder(crtBuilder, maxConcurrency); + if (StringUtils.isNotBlank(endpoint)) { + crtBuilder.endpointOverride(URI.create(endpoint)); + } + logger.info("Using CRT S3 client with minimum throughput {} B/s and health timeout {} seconds; " + + "SDK API call/attempt timeouts are not supported by this client", + configuration.crtMinimumThroughputBps(), configuration.crtMinimumThroughputTimeout().getSeconds()); + return crtBuilder.build(); + } S3AsyncClientBuilder builder = S3AsyncClient.builder().region(Region.of(region)); if (StringUtils.isNotBlank(endpoint)) { builder.endpointOverride(URI.create(endpoint)); @@ -505,20 +522,22 @@ protected S3AsyncClient newS3Client(String endpoint, String region, boolean forc logger.warn("OpenSSL is not available, using JDK SSL provider, which may have performance issue.", OpenSsl.unavailabilityCause()); } SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder() - .maxConcurrency(maxConcurrency) + .maxConcurrency(configuration.maxConcurrency() == -1 ? maxConcurrency : configuration.maxConcurrency()) + .connectionTimeout(configuration.connectionTimeout()) .build(); builder.httpClient(httpClient); builder.serviceConfiguration(c -> c.pathStyleAccessEnabled(forcePathStyle).checksumValidationEnabled(false)); builder.credentialsProvider(newCredentialsProviderChain(credentialsProviders)); - builder.overrideConfiguration(clientOverrideConfiguration(apiCallTimeoutMs, apiCallAttemptTimeoutMs)); + builder.overrideConfiguration(clientOverrideConfiguration(configuration)); return builder.build(); } - protected ClientOverrideConfiguration clientOverrideConfiguration(long apiCallTimeoutMs, long apiCallAttemptTimeoutMs) { + protected ClientOverrideConfiguration clientOverrideConfiguration(Configuration configuration) { return ClientOverrideConfiguration.builder() - .apiCallTimeout(Duration.ofMillis(apiCallTimeoutMs)) - .apiCallAttemptTimeout(Duration.ofMillis(apiCallAttemptTimeoutMs)) - .retryStrategy(RetryMode.STANDARD) + .apiCallTimeout(configuration.apiCallTimeout()) + .apiCallAttemptTimeout(configuration.apiCallAttemptTimeout()) + .retryStrategy(software.amazon.awssdk.awscore.retry.AwsRetryStrategy.standardRetryStrategy() + .toBuilder().maxAttempts(configuration.numRetries() + 1).build()) .build(); } @@ -745,6 +764,158 @@ public boolean readinessCheck() { } + /** + * Immutable per-bucket client options. SDK-specific constraints are checked by the SDK when constructing clients. + */ + public record Configuration(String sdkImplementation, long crtMinimumThroughputBps, Duration crtMinimumThroughputTimeout, + double crtTargetThroughputGbps, int maxConcurrency, long crtMaxNativeMemoryBytes, + Duration connectionTimeout, int numRetries, + Duration apiCallTimeout, Duration apiCallAttemptTimeout) { + /** + * Reads camel-case client options from a bucket URI without altering the URI or other buckets. + */ + public static Configuration fromBucketURI(BucketURI uri) { + String implementation = uri.extensionString("sdkImpl", "netty"); + if (!"netty".equals(implementation) && !"crt".equals(implementation)) { + throw new IllegalArgumentException("Unsupported sdkImpl: " + implementation); + } + Builder builder = builder().sdkImplementation(implementation) + .maxConcurrency(Integer.parseInt(uri.extensionString(BucketURI.MAX_CONCURRENCY_KEY, "-1"))) + .connectionTimeout(Duration.ofMillis(Long.parseLong(uri.extensionString(BucketURI.CONNECTION_TIMEOUT_KEY, "2000")))) + .numRetries(Integer.parseInt(uri.extensionString(BucketURI.NUM_RETRIES_KEY, "2"))) + .apiCallTimeout(Duration.ofMillis(Long.parseLong(uri.extensionString(BucketURI.API_CALL_TIMEOUT_KEY, "30000")))) + .apiCallAttemptTimeout(Duration.ofMillis(Long.parseLong(uri.extensionString(BucketURI.API_CALL_ATTEMPT_TIMEOUT_KEY, "10000")))); + if ("crt".equals(implementation)) { + int healthTimeout = Integer.parseInt(uri.extensionString("crtMinimumThroughputTimeoutSeconds", "2")); + if (healthTimeout < 2) { + throw new IllegalArgumentException("crtMinimumThroughputTimeoutSeconds must be at least 2"); + } + builder.crtTargetThroughputGbps(Double.parseDouble(uri.extensionString("crtTargetThroughputInGbps", "10"))) + .crtMinimumThroughputBps(Long.parseLong(uri.extensionString("crtMinimumThroughputInBps", "33554432"))) + .crtMinimumThroughputTimeout(Duration.ofSeconds(healthTimeout)) + .crtMaxNativeMemoryBytes(Long.parseLong(uri.extensionString("crtMaxNativeMemoryLimitInBytes", "-1"))); + } + return builder.build(); + } + + /** + * Creates a builder with Netty, CRT health defaults of 32 MiB/s and two seconds, and a 10 Gbps transfer target. + */ + public static Builder builder() { + return new Builder(); + } + + /** + * Builds immutable configurations; subsequent builder changes do not affect previously built snapshots. + */ + public static final class Builder { + private String sdkImplementation = "netty"; + private long crtMinimumThroughputBps = 32L * 1024 * 1024; + private Duration crtMinimumThroughputTimeout = Duration.ofSeconds(2); + private double crtTargetThroughputGbps = 10.0; + private int maxConcurrency = -1; + private long crtMaxNativeMemoryBytes = -1; + private Duration connectionTimeout = Duration.ofSeconds(2); + private int numRetries = 2; + private Duration apiCallTimeout = Duration.ofSeconds(30); + private Duration apiCallAttemptTimeout = Duration.ofSeconds(10); + + private Builder() { + } + + /** + * Selects the S3 SDK implementation, either netty or crt. + */ + public Builder sdkImplementation(String implementation) { + this.sdkImplementation = implementation; + return this; + } + + /** + * Sets the minimum healthy throughput of each CRT connection in bytes per second. + */ + public Builder crtMinimumThroughputBps(long throughputBps) { + this.crtMinimumThroughputBps = throughputBps; + return this; + } + + /** + * Sets how long a CRT connection may remain below the throughput threshold. + */ + public Builder crtMinimumThroughputTimeout(Duration timeout) { + this.crtMinimumThroughputTimeout = timeout; + return this; + } + + /** + * Sets the CRT transfer throughput target in gigabits per second; this is not a bandwidth limit. + */ + public Builder crtTargetThroughputGbps(double throughputGbps) { + this.crtTargetThroughputGbps = throughputGbps; + return this; + } + + /** Selects a connection cap, or -1 to inherit AutoMQ's existing client concurrency. */ + public Builder maxConcurrency(int maxConcurrency) { + this.maxConcurrency = maxConcurrency; + return this; + } + + /** Sets a per-client native memory budget, or -1 to let CRT calculate it. */ + public Builder crtMaxNativeMemoryBytes(long memoryBytes) { + this.crtMaxNativeMemoryBytes = memoryBytes; + return this; + } + + /** Sets the timeout for establishing a connection, independent of object completion time. */ + public Builder connectionTimeout(Duration timeout) { + this.connectionTimeout = timeout; + return this; + } + + /** Sets retries per HTTP request, excluding the initial attempt. */ + public Builder numRetries(int numRetries) { + this.numRetries = numRetries; + return this; + } + + /** Sets the Netty SDK deadline for an API call including all attempts. */ + public Builder apiCallTimeout(Duration timeout) { + this.apiCallTimeout = timeout; + return this; + } + + /** Sets the Netty SDK deadline for one API call attempt. */ + public Builder apiCallAttemptTimeout(Duration timeout) { + this.apiCallAttemptTimeout = timeout; + return this; + } + + /** + * Creates a snapshot without changing the process-wide configuration. + */ + public Configuration build() { + return new Configuration(sdkImplementation, crtMinimumThroughputBps, crtMinimumThroughputTimeout, + crtTargetThroughputGbps, maxConcurrency, crtMaxNativeMemoryBytes, + connectionTimeout, numRetries, apiCallTimeout, apiCallAttemptTimeout); + } + } + + void configureCrtBuilder(S3CrtAsyncClientBuilder builder, int inheritedMaxConcurrency) { + builder.maxConcurrency(maxConcurrency == -1 ? inheritedMaxConcurrency : maxConcurrency) + .targetThroughputInGbps(crtTargetThroughputGbps) + // AutoMQ owns multipart uploads and precomputes checksums for individual requests. + .thresholdInBytes(Long.MAX_VALUE) + .retryConfiguration(retry -> retry.numRetries(numRetries)) + .httpConfiguration(http -> http.connectionTimeout(connectionTimeout) + .connectionHealthConfiguration(health -> health.minimumThroughputInBps(crtMinimumThroughputBps) + .minimumThroughputTimeout(crtMinimumThroughputTimeout))); + if (crtMaxNativeMemoryBytes != -1) { + builder.maxNativeMemoryLimitInBytes(crtMaxNativeMemoryBytes); + } + } + } + public static class Builder { private BucketURI bucketURI; private Map tagging; diff --git a/s3stream/src/main/java/com/automq/stream/s3/operator/BucketURI.java b/s3stream/src/main/java/com/automq/stream/s3/operator/BucketURI.java index 0464b6751f..bc5eb4ff7b 100644 --- a/s3stream/src/main/java/com/automq/stream/s3/operator/BucketURI.java +++ b/s3stream/src/main/java/com/automq/stream/s3/operator/BucketURI.java @@ -44,6 +44,9 @@ public class BucketURI { public static final String SECRET_KEY_KEY = "secretKey"; public static final String API_CALL_TIMEOUT_KEY = "apiCallTimeoutMs"; public static final String API_CALL_ATTEMPT_TIMEOUT_KEY = "apiCallAttemptTimeoutMs"; + public static final String CONNECTION_TIMEOUT_KEY = "connectionTimeoutMs"; + public static final String NUM_RETRIES_KEY = "numRetries"; + public static final String MAX_CONCURRENCY_KEY = "maxConcurrency"; private static final String EMPTY_STRING = ""; private final short bucketId; private final String protocol; diff --git a/s3stream/src/test/java/com/automq/stream/s3/operator/AwsObjectStorageTest.java b/s3stream/src/test/java/com/automq/stream/s3/operator/AwsObjectStorageTest.java index b755b67e60..fe7d94be62 100644 --- a/s3stream/src/test/java/com/automq/stream/s3/operator/AwsObjectStorageTest.java +++ b/s3stream/src/test/java/com/automq/stream/s3/operator/AwsObjectStorageTest.java @@ -51,6 +51,173 @@ public class AwsObjectStorageTest { + /** + * CRT keeps PUTs single-request while AutoMQ retains ownership of multipart uploads and precomputed checksums. + */ + @Tag("S3Unit") + @Test + void testCrtDisablesAutomaticMultipart() { + AwsObjectStorage storage = new AwsObjectStorage(mock(S3AsyncClient.class), "bucket", ChecksumAlgorithm.CRC32); + var builder = mock(software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder.class, org.mockito.Mockito.RETURNS_SELF); + when(builder.build()).thenReturn(mock(S3AsyncClient.class)); + try (var factory = org.mockito.Mockito.mockStatic(S3AsyncClient.class)) { + factory.when(S3AsyncClient::crtBuilder).thenReturn(builder); + BucketURI uri = BucketURI.parse("0@s3://bucket?region=us-east-1&sdkImpl=crt"); + try (S3AsyncClient ignored = storage.newS3Client(uri, List.of(), AwsObjectStorage.Configuration.fromBucketURI(uri))) { + org.mockito.Mockito.verify(builder).thresholdInBytes(Long.MAX_VALUE); + } + } finally { + storage.close(); + } + } + + /** + * Each URI supplies its own SDK and tuning options; configuring one bucket leaves another bucket's defaults intact. + */ + @Tag("S3Unit") + @Test + void testPerBucketClientOptions() { + BucketURI uri = BucketURI.parse("0@s3://one?region=us-east-1&sdkImpl=crt&connectionTimeoutMs=1500" + + "&numRetries=0&maxConcurrency=12&crtTargetThroughputInGbps=2.5&crtMinimumThroughputInBps=1048576" + + "&crtMinimumThroughputTimeoutSeconds=3&crtMaxNativeMemoryLimitInBytes=1073741824"); + AwsObjectStorage.Configuration options = AwsObjectStorage.Configuration.fromBucketURI(uri); + Assertions.assertEquals("crt", options.sdkImplementation()); + Assertions.assertEquals(java.time.Duration.ofMillis(1500), options.connectionTimeout()); + Assertions.assertEquals(0, options.numRetries()); + Assertions.assertEquals(12, options.maxConcurrency()); + Assertions.assertEquals(2.5, options.crtTargetThroughputGbps()); + Assertions.assertEquals(1048576L, options.crtMinimumThroughputBps()); + Assertions.assertEquals(java.time.Duration.ofSeconds(3), options.crtMinimumThroughputTimeout()); + Assertions.assertEquals(1073741824L, options.crtMaxNativeMemoryBytes()); + Assertions.assertEquals(AwsObjectStorage.Configuration.builder().build(), + AwsObjectStorage.Configuration.fromBucketURI(BucketURI.parse("1@s3://two?region=us-east-1"))); + } + + /** + * Netty consumes the shared retry setting, preserves API deadlines, and ignores CRT-only settings. + */ + @Tag("S3Unit") + @Test + void testNettySharedRetryOptions() { + AwsObjectStorage storage = new AwsObjectStorage(mock(S3AsyncClient.class), "bucket"); + try { + storage.bucketURI.addExtension("numRetries", "0"); + storage.bucketURI.addExtension("connectionTimeoutMs", "1500"); + storage.bucketURI.addExtension("maxConcurrency", "12"); + storage.bucketURI.addExtension("crtTargetThroughputInGbps", "ignored"); + var configuration = AwsObjectStorage.Configuration.fromBucketURI(storage.bucketURI); + var overrides = storage.clientOverrideConfiguration(configuration); + Assertions.assertEquals(1, overrides.retryStrategy().orElseThrow().maxAttempts()); + Assertions.assertEquals(java.time.Duration.ofMillis(30000), overrides.apiCallTimeout().orElseThrow()); + Assertions.assertEquals(java.time.Duration.ofMillis(10000), overrides.apiCallAttemptTimeout().orElseThrow()); + try (S3AsyncClient client = storage.newS3Client( + BucketURI.parse("0@s3://bucket?region=us-east-1&endpoint=http://localhost:9000&pathStyle=true"), + List.of(() -> AwsBasicCredentials.create("test", "test")), configuration)) { + Assertions.assertEquals("s3", client.serviceName()); + } + } finally { + storage.close(); + } + } + + /** + * Unsupported SDK names and CRT health windows below two seconds fail during URI option parsing. + */ + @Tag("S3Unit") + @Test + void testBucketClientOptionValidation() { + Assertions.assertThrows(IllegalArgumentException.class, () -> AwsObjectStorage.Configuration.fromBucketURI( + BucketURI.parse("0@s3://one?sdkImpl=unknown"))); + Assertions.assertThrows(IllegalArgumentException.class, () -> AwsObjectStorage.Configuration.fromBucketURI( + BucketURI.parse("0@s3://one?sdkImpl=crt&crtMinimumThroughputTimeoutSeconds=1"))); + } + + /** + * Both SDK implementations build with endpoint and path-style overrides using the existing runtime dependencies. + */ + @Tag("S3Unit") + @Test + void testSdkClientSelection() { + AwsObjectStorage storage = new AwsObjectStorage(mock(S3AsyncClient.class), "bucket"); + try { + for (String implementation : List.of("netty", "crt")) { + storage.bucketURI.addExtension("sdkImpl", implementation); + try (S3AsyncClient client = storage.newS3Client( + BucketURI.parse("0@s3://bucket?region=us-east-1&endpoint=http://localhost:9000&pathStyle=true"), + List.of(() -> AwsBasicCredentials.create("test", "test")), + AwsObjectStorage.Configuration.fromBucketURI(storage.bucketURI))) { + Assertions.assertEquals("s3", client.serviceName()); + } + } + } finally { + storage.bucketURI.addExtension("sdkImpl", "netty"); + storage.close(); + } + } + + /** + * Builder mutations preserve previously built snapshots and unmodified fields retain their defaults. + */ + @Tag("S3Unit") + @Test + void testConfigurationSnapshot() { + AwsObjectStorage.Configuration.Builder builder = AwsObjectStorage.Configuration.builder(); + AwsObjectStorage.Configuration defaults = builder.build(); + AwsObjectStorage.Configuration custom = builder.sdkImplementation("crt") + .crtMinimumThroughputBps(1024).crtMinimumThroughputTimeout(java.time.Duration.ofSeconds(3)) + .crtTargetThroughputGbps(2.5).build(); + Assertions.assertEquals("netty", defaults.sdkImplementation()); + Assertions.assertEquals(32L * 1024 * 1024, defaults.crtMinimumThroughputBps()); + Assertions.assertEquals(java.time.Duration.ofSeconds(2), defaults.crtMinimumThroughputTimeout()); + Assertions.assertEquals("crt", custom.sdkImplementation()); + Assertions.assertEquals(10.0, defaults.crtTargetThroughputGbps()); + Assertions.assertEquals(2.5, custom.crtTargetThroughputGbps()); + Assertions.assertEquals(1024, custom.crtMinimumThroughputBps()); + Assertions.assertEquals(java.time.Duration.ofSeconds(3), custom.crtMinimumThroughputTimeout()); + Assertions.assertEquals(-1, defaults.maxConcurrency()); + Assertions.assertEquals(-1L, defaults.crtMaxNativeMemoryBytes()); + Assertions.assertEquals(java.time.Duration.ofSeconds(2), defaults.connectionTimeout()); + Assertions.assertEquals(2, defaults.numRetries()); + } + + /** + * Automatic concurrency inherits the existing cap, while native memory stays unset until explicitly configured. + */ + @Tag("S3Unit") + @Test + void testCrtBuilderConfiguration() { + software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder builder = mock( + software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder.class, org.mockito.Mockito.RETURNS_SELF); + org.mockito.Mockito.doAnswer(invocation -> { + Consumer configure = invocation.getArgument(0); + var http = software.amazon.awssdk.services.s3.crt.S3CrtHttpConfiguration.builder(); + configure.accept(http); + var result = http.build(); + Assertions.assertEquals(java.time.Duration.ofSeconds(2), result.connectionTimeout()); + Assertions.assertEquals(32L * 1024 * 1024, result.healthConfiguration().minimumThroughputInBps()); + Assertions.assertEquals(java.time.Duration.ofSeconds(2), result.healthConfiguration().minimumThroughputTimeout()); + return builder; + }).when(builder).httpConfiguration(org.mockito.ArgumentMatchers + .>any()); + org.mockito.Mockito.doAnswer(invocation -> { + Consumer configure = invocation.getArgument(0); + var retry = software.amazon.awssdk.services.s3.crt.S3CrtRetryConfiguration.builder(); + configure.accept(retry); + Assertions.assertEquals(2, retry.build().numRetries()); + return builder; + }).when(builder).retryConfiguration(org.mockito.ArgumentMatchers + .>any()); + AwsObjectStorage.Configuration.builder().build().configureCrtBuilder(builder, 77); + org.mockito.Mockito.verify(builder).maxConcurrency(77); + org.mockito.Mockito.verify(builder).thresholdInBytes(Long.MAX_VALUE); + org.mockito.Mockito.verify(builder, org.mockito.Mockito.never()).maxNativeMemoryLimitInBytes(any()); + AwsObjectStorage.Configuration.builder().maxConcurrency(12).crtMaxNativeMemoryBytes(1024L * 1024 * 1024) + .build().configureCrtBuilder(builder, 77); + org.mockito.Mockito.verify(builder).maxConcurrency(12); + org.mockito.Mockito.verify(builder).maxNativeMemoryLimitInBytes(1024L * 1024 * 1024); + org.mockito.Mockito.verify(builder, org.mockito.Mockito.times(2)).thresholdInBytes(Long.MAX_VALUE); + } + /** * Given a timeout on a later LIST page, retry restarts pagination and returns only the successful attempt's * objects, without retaining partial results or a stale continuation token. From 86c7fa2cd8fc798cf5c3fd79084fe60c6606a283 Mon Sep 17 00:00:00 2001 From: Robin Han Date: Mon, 12 Oct 2026 22:06:19 +0800 Subject: [PATCH 2/2] docs: keep devkit README unchanged --- devkit/README.md | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/devkit/README.md b/devkit/README.md index 46eb55b0d5..cdabc731a9 100644 --- a/devkit/README.md +++ b/devkit/README.md @@ -426,22 +426,6 @@ num.partitions=4 log.retention.hours=24 ``` -**Select the S3 SDK per bucket** — add camel-case options to each data, ops, or WAL bucket URI: - -```properties -s3.data.buckets=0@s3://automq-data?region=us-east-1&endpoint=http://minio:9000&pathStyle=true&sdkImpl=crt&connectionTimeoutMs=2000&numRetries=2&maxConcurrency=-1&crtTargetThroughputInGbps=10&crtMinimumThroughputInBps=33554432&crtMinimumThroughputTimeoutSeconds=2&crtMaxNativeMemoryLimitInBytes=-1 -``` - -Omitting `sdkImpl` selects Netty. `connectionTimeoutMs`, `numRetries` (excluding the initial attempt), -and `maxConcurrency` apply to both clients. `maxConcurrency=-1` inherits AutoMQ's existing client concurrency -without changing its inflight quotas. CRT-only options are ignored by Netty; omitted options use the values -in the example. `crtMaxNativeMemoryLimitInBytes=-1` leaves native memory budgeting to CRT. -`crtMinimumThroughputTimeoutSeconds` must be at least two. The throughput target is per client and is not -a bandwidth limiter. Existing `apiCallTimeoutMs` and `apiCallAttemptTimeoutMs` remain supported by Netty. -CRT automatic multipart uploads are disabled with a fixed `Long.MAX_VALUE` threshold; AutoMQ's explicit -multipart uploads remain enabled, preserving its request-level precomputed checksums. -Configure each `s3.ops.buckets` and `s3.wal.path` URI separately; settings do not propagate between buckets. - **Add a new feature** — create `config/features/my-feature.properties`: ```properties my.setting=value