diff --git a/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/AliBlobStore.java b/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/AliBlobStore.java index 789a60f8e..954304985 100644 --- a/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/AliBlobStore.java +++ b/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/AliBlobStore.java @@ -998,11 +998,19 @@ private static OSSClient buildOSSClient(Builder builder) { builder, creds, retryer, - (proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout) -> - Apache5HttpClientBuilder.create() - .options(AliTransformer.toHttpClientOptions( - proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout)) - .build()); + (proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout, + disableConnectionReaper) -> { + Apache5HttpClientBuilder httpClientBuilder = + Apache5HttpClientBuilder.create() + .options(AliTransformer.toHttpClientOptions( + proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout)); + // The idle-connection reaper is a setting on the Apache5 client builder itself. Leave + // it at the builder default unless the caller explicitly set the flag. + if (disableConnectionReaper != null) { + httpClientBuilder.useReaper(!disableConnectionReaper); + } + return httpClientBuilder.build(); + }); return clientBuilder.build(); } diff --git a/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/OssClientFactory.java b/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/OssClientFactory.java index 9bf7fa5a3..01f0c9032 100644 --- a/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/OssClientFactory.java +++ b/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/OssClientFactory.java @@ -27,11 +27,15 @@ private OssClientFactory() {} * Builds the transport {@link HttpClient} for a specific client kind (sync or async) from the * resolved connection options. This is the only part of client construction that differs between * the sync and async paths, so it is supplied by the caller as a small strategy. + * + *

{@code disableConnectionReaper} is applied here rather than via {@code HttpClientOptions} + * because the idle-connection reaper is a setting on the Apache5 client builder itself, not on + * the options object. When {@code null}, the builder's default reaper behavior is retained. */ @FunctionalInterface public interface HttpClientFactory { HttpClient create(String proxyHost, Duration readWriteTimeout, - Integer maxConnections, Duration idleConnectionTimeout); + Integer maxConnections, Duration idleConnectionTimeout, Boolean disableConnectionReaper); } /** @@ -89,18 +93,23 @@ public static , T> void configure( Duration readWriteTimeout = resolveReadWriteTimeout(mcjBuilder.getRetryConfig(), mcjBuilder.getSocketTimeout()); - // Connection-pool size and idle-connection timeout are only settable via HttpClientOptions, - // not on the OSS client builder. When the caller sets either, build an explicit transport - // client from those options (carrying proxyHost + readWriteTimeout forward so nothing the - // builder would otherwise set is lost). When neither is set, leave the SDK to construct its - // own default client and set readWriteTimeout directly, preserving the prior behavior. - if (mcjBuilder.getMaxConnections() != null || mcjBuilder.getIdleConnectionTimeout() != null) { + // Connection-pool size, idle-connection timeout, and the idle-connection reaper are only + // reachable by supplying an explicit transport client: the first two live on + // HttpClientOptions and the reaper lives on the Apache5 client builder, neither of which the + // OSS client builder exposes. When the caller sets any of them, build an explicit transport + // client (carrying proxyHost + readWriteTimeout forward so nothing the builder would otherwise + // set is lost). When none is set, leave the SDK to construct its own default client and set + // readWriteTimeout directly, preserving the prior behavior. + if (mcjBuilder.getMaxConnections() != null + || mcjBuilder.getIdleConnectionTimeout() != null + || mcjBuilder.getDisableConnectionReaper() != null) { clientBuilder.httpClient( httpClientFactory.create( proxyHost, readWriteTimeout, mcjBuilder.getMaxConnections(), - mcjBuilder.getIdleConnectionTimeout())); + mcjBuilder.getIdleConnectionTimeout(), + mcjBuilder.getDisableConnectionReaper())); } else if (readWriteTimeout != null) { clientBuilder.readWriteTimeout(readWriteTimeout); } diff --git a/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStore.java b/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStore.java index 18447d286..f86525b68 100644 --- a/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStore.java +++ b/blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStore.java @@ -967,11 +967,19 @@ public AsyncBlobStore build() { this, creds, retryer, - (proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout) -> - Apache5AsyncHttpClientBuilder.create() - .options(AliTransformer.toHttpClientOptions( - proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout)) - .build()); + (proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout, + disableConnectionReaper) -> { + Apache5AsyncHttpClientBuilder httpClientBuilder = + Apache5AsyncHttpClientBuilder.create() + .options(AliTransformer.toHttpClientOptions( + proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout)); + // The idle-connection reaper is a setting on the Apache5 client builder itself. + // Leave it at the builder default unless the caller explicitly set the flag. + if (disableConnectionReaper != null) { + httpClientBuilder.useReaper(!disableConnectionReaper); + } + return httpClientBuilder.build(); + }); async = asyncBuilder.build(); } @@ -983,11 +991,19 @@ public AsyncBlobStore build() { this, creds, retryer, - (proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout) -> - Apache5HttpClientBuilder.create() - .options(AliTransformer.toHttpClientOptions( - proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout)) - .build()); + (proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout, + disableConnectionReaper) -> { + Apache5HttpClientBuilder httpClientBuilder = + Apache5HttpClientBuilder.create() + .options(AliTransformer.toHttpClientOptions( + proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout)); + // The idle-connection reaper is a setting on the Apache5 client builder itself. + // Leave it at the builder default unless the caller explicitly set the flag. + if (disableConnectionReaper != null) { + httpClientBuilder.useReaper(!disableConnectionReaper); + } + return httpClientBuilder.build(); + }); sync = syncBuilder.build(); } diff --git a/blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/AliBlobStoreTest.java b/blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/AliBlobStoreTest.java index 8a1be5566..00e6d0109 100644 --- a/blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/AliBlobStoreTest.java +++ b/blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/AliBlobStoreTest.java @@ -11,6 +11,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -68,6 +69,8 @@ import com.aliyun.sdk.service.oss2.models.UploadPartRequest; import com.aliyun.sdk.service.oss2.models.UploadPartResult; import com.aliyun.sdk.service.oss2.paginator.ListObjectVersionsIterable; +import com.aliyun.sdk.service.oss2.transport.apache5client.Apache5HttpClient; +import com.aliyun.sdk.service.oss2.transport.apache5client.Apache5HttpClientBuilder; import com.salesforce.multicloudj.blob.driver.BlobIdentifier; import com.salesforce.multicloudj.blob.driver.BlobInfo; import com.salesforce.multicloudj.blob.driver.BlobMetadata; @@ -134,6 +137,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import org.mockito.MockedStatic; public class AliBlobStoreTest { @@ -2273,4 +2277,88 @@ void testBuildOSSClient_withSocketTimeoutOnly_usesReadWriteTimeoutBranch() { builder.withSocketTimeout(Duration.ofSeconds(30)); assertNotNull(builder.build()); } + + private static CredentialsOverrider sessionCredentialsOverrider() { + StsCredentials creds = new StsCredentials("key-1", "secret-1", "token-1"); + return new CredentialsOverrider.Builder(CredentialsType.SESSION) + .withSessionCredentials(creds) + .build(); + } + + /** + * Builds a sync store (without injecting a client) while intercepting + * {@code Apache5HttpClientBuilder.create()} so the reaper flag applied to the OSS SDK's Apache5 + * HTTP client can be asserted. + */ + private Apache5HttpClientBuilder captureApache5BuilderFor(AliBlobStore.Builder builder) { + Apache5HttpClientBuilder apacheBuilder = mock(Apache5HttpClientBuilder.class); + when(apacheBuilder.options(any())).thenReturn(apacheBuilder); + when(apacheBuilder.useReaper(true)).thenReturn(apacheBuilder); + when(apacheBuilder.useReaper(false)).thenReturn(apacheBuilder); + when(apacheBuilder.build()).thenReturn(mock(Apache5HttpClient.class)); + + try (MockedStatic apacheStatic = + mockStatic(Apache5HttpClientBuilder.class)) { + apacheStatic.when(Apache5HttpClientBuilder::create).thenReturn(apacheBuilder); + builder.build(); + } + return apacheBuilder; + } + + @Test + void testDisableConnectionReaperWiredIntoSyncClient() { + AliBlobStore.Builder builder = + (AliBlobStore.Builder) + new AliBlobStore.Builder() + .withBucket("bucket-1") + .withRegion("cn-shanghai") + .withCredentialsOverrider(sessionCredentialsOverrider()) + .withDisableConnectionReaper(true); + + Apache5HttpClientBuilder apacheBuilder = captureApache5BuilderFor(builder); + + // disable=true must translate to useReaper(false) + verify(apacheBuilder).useReaper(false); + } + + @Test + void testEnableConnectionReaperWiredIntoSyncClient() { + AliBlobStore.Builder builder = + (AliBlobStore.Builder) + new AliBlobStore.Builder() + .withBucket("bucket-1") + .withRegion("cn-shanghai") + .withCredentialsOverrider(sessionCredentialsOverrider()) + .withDisableConnectionReaper(false); + + Apache5HttpClientBuilder apacheBuilder = captureApache5BuilderFor(builder); + + // disable=false must translate to useReaper(true) + verify(apacheBuilder).useReaper(true); + } + + @Test + void testReaperUntouchedWhenUnsetSyncClient() { + AliBlobStore.Builder builder = + (AliBlobStore.Builder) + new AliBlobStore.Builder() + .withBucket("bucket-1") + .withRegion("cn-shanghai") + .withCredentialsOverrider(sessionCredentialsOverrider()); + + Apache5HttpClientBuilder apacheBuilder = mock(Apache5HttpClientBuilder.class); + when(apacheBuilder.options(any())).thenReturn(apacheBuilder); + when(apacheBuilder.useReaper(true)).thenReturn(apacheBuilder); + when(apacheBuilder.useReaper(false)).thenReturn(apacheBuilder); + when(apacheBuilder.build()).thenReturn(mock(Apache5HttpClient.class)); + + try (MockedStatic apacheStatic = + mockStatic(Apache5HttpClientBuilder.class)) { + apacheStatic.when(Apache5HttpClientBuilder::create).thenReturn(apacheBuilder); + builder.build(); + } + + verify(apacheBuilder, never()).useReaper(true); + verify(apacheBuilder, never()).useReaper(false); + } } diff --git a/blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStoreTest.java b/blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStoreTest.java index 852961af3..7870b6018 100644 --- a/blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStoreTest.java +++ b/blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStoreTest.java @@ -10,6 +10,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -61,6 +62,10 @@ import com.aliyun.sdk.service.oss2.transfermanager.DownloadError; import com.aliyun.sdk.service.oss2.transfermanager.DownloadResult; import com.aliyun.sdk.service.oss2.transfermanager.Downloader; +import com.aliyun.sdk.service.oss2.transport.apache5client.Apache5AsyncHttpClient; +import com.aliyun.sdk.service.oss2.transport.apache5client.Apache5AsyncHttpClientBuilder; +import com.aliyun.sdk.service.oss2.transport.apache5client.Apache5HttpClient; +import com.aliyun.sdk.service.oss2.transport.apache5client.Apache5HttpClientBuilder; import com.salesforce.multicloudj.blob.ali.AliTransformerSupplier; import com.salesforce.multicloudj.blob.async.driver.AsyncBlobStore; import com.salesforce.multicloudj.blob.driver.BlobIdentifier; @@ -109,6 +114,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.mockito.ArgumentCaptor; +import org.mockito.MockedStatic; public class AliAsyncBlobStoreTest { @@ -1875,4 +1881,79 @@ void testBuildClients_withOnlyIdleConnectionTimeout_buildsSuccessfully() { builder.withIdleConnectionTimeout(Duration.ofSeconds(45)); assertNotNull(builder.build()); } + + private static CredentialsOverrider sessionCredentialsOverrider() { + StsCredentials creds = new StsCredentials("key-1", "secret-1", "token-1"); + return new CredentialsOverrider.Builder(CredentialsType.SESSION) + .withSessionCredentials(creds) + .build(); + } + + @Test + void testDisableConnectionReaperWiredIntoAsyncAndSyncClients() { + AliAsyncBlobStore.Builder builder = new AliAsyncBlobStore.Builder(); + builder.withBucket("bucket-1"); + builder.withRegion("cn-shanghai"); + builder.withCredentialsOverrider(sessionCredentialsOverrider()); + builder.withDisableConnectionReaper(true); + + Apache5AsyncHttpClientBuilder asyncApache = mock(Apache5AsyncHttpClientBuilder.class); + when(asyncApache.options(any())).thenReturn(asyncApache); + when(asyncApache.useReaper(true)).thenReturn(asyncApache); + when(asyncApache.useReaper(false)).thenReturn(asyncApache); + when(asyncApache.build()).thenReturn(mock(Apache5AsyncHttpClient.class)); + + Apache5HttpClientBuilder syncApache = mock(Apache5HttpClientBuilder.class); + when(syncApache.options(any())).thenReturn(syncApache); + when(syncApache.useReaper(true)).thenReturn(syncApache); + when(syncApache.useReaper(false)).thenReturn(syncApache); + when(syncApache.build()).thenReturn(mock(Apache5HttpClient.class)); + + try (MockedStatic asyncStatic = + mockStatic(Apache5AsyncHttpClientBuilder.class); + MockedStatic syncStatic = + mockStatic(Apache5HttpClientBuilder.class)) { + asyncStatic.when(Apache5AsyncHttpClientBuilder::create).thenReturn(asyncApache); + syncStatic.when(Apache5HttpClientBuilder::create).thenReturn(syncApache); + builder.build(); + } + + // disable=true must translate to useReaper(false) on both the async and sync pools. + verify(asyncApache).useReaper(false); + verify(syncApache).useReaper(false); + } + + @Test + void testReaperUntouchedWhenUnsetAsyncAndSyncClients() { + AliAsyncBlobStore.Builder builder = new AliAsyncBlobStore.Builder(); + builder.withBucket("bucket-1"); + builder.withRegion("cn-shanghai"); + builder.withCredentialsOverrider(sessionCredentialsOverrider()); + + Apache5AsyncHttpClientBuilder asyncApache = mock(Apache5AsyncHttpClientBuilder.class); + when(asyncApache.options(any())).thenReturn(asyncApache); + when(asyncApache.useReaper(true)).thenReturn(asyncApache); + when(asyncApache.useReaper(false)).thenReturn(asyncApache); + when(asyncApache.build()).thenReturn(mock(Apache5AsyncHttpClient.class)); + + Apache5HttpClientBuilder syncApache = mock(Apache5HttpClientBuilder.class); + when(syncApache.options(any())).thenReturn(syncApache); + when(syncApache.useReaper(true)).thenReturn(syncApache); + when(syncApache.useReaper(false)).thenReturn(syncApache); + when(syncApache.build()).thenReturn(mock(Apache5HttpClient.class)); + + try (MockedStatic asyncStatic = + mockStatic(Apache5AsyncHttpClientBuilder.class); + MockedStatic syncStatic = + mockStatic(Apache5HttpClientBuilder.class)) { + asyncStatic.when(Apache5AsyncHttpClientBuilder::create).thenReturn(asyncApache); + syncStatic.when(Apache5HttpClientBuilder::create).thenReturn(syncApache); + builder.build(); + } + + verify(asyncApache, never()).useReaper(true); + verify(asyncApache, never()).useReaper(false); + verify(syncApache, never()).useReaper(true); + verify(syncApache, never()).useReaper(false); + } } diff --git a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java index 553ab182f..f675ce07d 100644 --- a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java +++ b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java @@ -115,7 +115,9 @@ protected static boolean shouldConfigureHttpClient(Builder builder) { || builder.getSocketTimeout() != null || builder.getIdleConnectionTimeout() != null || builder.getUseSystemPropertyProxyValues() != null - || builder.getUseEnvironmentVariableProxyValues() != null; + || builder.getUseEnvironmentVariableProxyValues() != null + || builder.getDisableConnectionReaper() != null + || builder.getTcpKeepAlive() != null; } @Override @@ -703,6 +705,16 @@ public Builder self() { return this; } + @Override + public Boolean getDisableConnectionReaper() { + return super.getDisableConnectionReaper(); + } + + @Override + public Boolean getTcpKeepAlive() { + return super.getTcpKeepAlive(); + } + /** Helper function to generate the client */ private static S3Client buildS3Client(Builder builder) { Region regionObj = Region.of(builder.getRegion()); @@ -769,6 +781,12 @@ private static SdkHttpClient generateHttpClient(Builder builder) { if (builder.getIdleConnectionTimeout() != null) { httpClientBuilder.connectionMaxIdleTime(builder.getIdleConnectionTimeout()); } + if (builder.getDisableConnectionReaper() != null) { + httpClientBuilder.useIdleConnectionReaper(!builder.getDisableConnectionReaper()); + } + if (builder.getTcpKeepAlive() != null) { + httpClientBuilder.tcpKeepAlive(builder.getTcpKeepAlive()); + } return httpClientBuilder.build(); } diff --git a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStore.java b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStore.java index 5074af9d5..acd3319f6 100644 --- a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStore.java +++ b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStore.java @@ -589,6 +589,16 @@ public Builder() { providerId(AwsConstants.PROVIDER_ID); } + @Override + public Boolean getDisableConnectionReaper() { + return super.getDisableConnectionReaper(); + } + + @Override + public Boolean getTcpKeepAlive() { + return super.getTcpKeepAlive(); + } + private static S3AsyncClient buildS3Client(Builder builder) { Region regionObj = Region.of(builder.getRegion()); @@ -659,7 +669,9 @@ private static void applyCommonConfig( || config.getSocketTimeout() != null || config.getIdleConnectionTimeout() != null || config.getUseSystemPropertyProxyValues() != null - || config.getUseEnvironmentVariableProxyValues() != null) { + || config.getUseEnvironmentVariableProxyValues() != null + || config.getDisableConnectionReaper() != null + || config.getTcpKeepAlive() != null) { NettyNioAsyncHttpClient.Builder httpClientBuilder = NettyNioAsyncHttpClient.builder(); @@ -700,6 +712,14 @@ private static void applyCommonConfig( httpClientBuilder.connectionMaxIdleTime(config.getIdleConnectionTimeout()); } + if (config.getDisableConnectionReaper() != null) { + httpClientBuilder.useIdleConnectionReaper(!config.getDisableConnectionReaper()); + } + + if (config.getTcpKeepAlive() != null) { + httpClientBuilder.tcpKeepAlive(config.getTcpKeepAlive()); + } + builder.httpClient(httpClientBuilder.build()); } diff --git a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsBlobStoreTest.java b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsBlobStoreTest.java index a98a8fcea..bf513b695 100644 --- a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsBlobStoreTest.java +++ b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsBlobStoreTest.java @@ -91,7 +91,9 @@ import software.amazon.awssdk.core.exception.SdkClientException; import software.amazon.awssdk.core.sync.RequestBody; import software.amazon.awssdk.core.sync.ResponseTransformer; +import software.amazon.awssdk.http.SdkHttpClient; import software.amazon.awssdk.http.SdkHttpResponse; +import software.amazon.awssdk.http.apache.ApacheHttpClient; import software.amazon.awssdk.retries.api.RetryStrategy; import software.amazon.awssdk.services.s3.S3Client; import software.amazon.awssdk.services.s3.S3ClientBuilder; @@ -289,6 +291,24 @@ void testShouldConfigureHttpClient() { AwsBlobStore.shouldConfigureHttpClient( (AwsBlobStore.Builder) builderWithUseEnvVarProxyValues)); + var builderWithDisableReaper = + new AwsBlobStore.Builder() + .withTransformerSupplier(transformerSupplier) + .withBucket("bucket-1") + .withRegion("us-east-2") + .withDisableConnectionReaper(true); + assertTrue( + AwsBlobStore.shouldConfigureHttpClient((AwsBlobStore.Builder) builderWithDisableReaper)); + + var builderWithTcpKeepAlive = + new AwsBlobStore.Builder() + .withTransformerSupplier(transformerSupplier) + .withBucket("bucket-1") + .withRegion("us-east-2") + .withTcpKeepAlive(true); + assertTrue( + AwsBlobStore.shouldConfigureHttpClient((AwsBlobStore.Builder) builderWithTcpKeepAlive)); + var builderWithNoOverrides = new AwsBlobStore.Builder() .withTransformerSupplier(transformerSupplier) @@ -298,6 +318,77 @@ void testShouldConfigureHttpClient() { AwsBlobStore.shouldConfigureHttpClient((AwsBlobStore.Builder) builderWithNoOverrides)); } + /** + * Builds a sync client with the given builder while intercepting + * {@code ApacheHttpClient.builder()} so the pool-hardening flags applied to the Apache HTTP + * client builder can be asserted. + */ + private ApacheHttpClient.Builder captureApacheHttpClientFor(AwsBlobStore.Builder builder) { + ApacheHttpClient.Builder apacheBuilder = mock(ApacheHttpClient.Builder.class); + when(apacheBuilder.useIdleConnectionReaper(any())).thenReturn(apacheBuilder); + when(apacheBuilder.tcpKeepAlive(any())).thenReturn(apacheBuilder); + when(apacheBuilder.maxConnections(any())).thenReturn(apacheBuilder); + when(apacheBuilder.socketTimeout(any())).thenReturn(apacheBuilder); + when(apacheBuilder.connectionMaxIdleTime(any())).thenReturn(apacheBuilder); + when(apacheBuilder.proxyConfiguration(any())).thenReturn(apacheBuilder); + when(apacheBuilder.build()).thenReturn(mock(SdkHttpClient.class)); + + try (MockedStatic apacheStatic = mockStatic(ApacheHttpClient.class)) { + apacheStatic.when(ApacheHttpClient::builder).thenReturn(apacheBuilder); + builder.build(); + } + return apacheBuilder; + } + + @Test + void testDisableConnectionReaperWiredIntoSyncHttpClient() { + AwsBlobStore.Builder builder = + (AwsBlobStore.Builder) + new AwsBlobStore.Builder() + .withTransformerSupplier(transformerSupplier) + .withBucket("bucket-1") + .withRegion("us-east-2") + .withDisableConnectionReaper(true); + + ApacheHttpClient.Builder apacheBuilder = captureApacheHttpClientFor(builder); + + // disable=true must translate to useIdleConnectionReaper(false) + verify(apacheBuilder).useIdleConnectionReaper(false); + verify(apacheBuilder, times(0)).tcpKeepAlive(any()); + } + + @Test + void testTcpKeepAliveWiredIntoSyncHttpClient() { + AwsBlobStore.Builder builder = + (AwsBlobStore.Builder) + new AwsBlobStore.Builder() + .withTransformerSupplier(transformerSupplier) + .withBucket("bucket-1") + .withRegion("us-east-2") + .withTcpKeepAlive(true); + + ApacheHttpClient.Builder apacheBuilder = captureApacheHttpClientFor(builder); + + verify(apacheBuilder).tcpKeepAlive(true); + verify(apacheBuilder, times(0)).useIdleConnectionReaper(any()); + } + + @Test + void testPoolHardeningFlagsUntouchedWhenUnset() { + AwsBlobStore.Builder builder = + (AwsBlobStore.Builder) + new AwsBlobStore.Builder() + .withTransformerSupplier(transformerSupplier) + .withBucket("bucket-1") + .withRegion("us-east-2") + .withMaxConnections(50); + + ApacheHttpClient.Builder apacheBuilder = captureApacheHttpClientFor(builder); + + verify(apacheBuilder, times(0)).useIdleConnectionReaper(any()); + verify(apacheBuilder, times(0)).tcpKeepAlive(any()); + } + @Test void testExceptionHandling() { AwsServiceException awsServiceException = diff --git a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStoreTest.java b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStoreTest.java index 8b1f251d5..68af4f7da 100644 --- a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStoreTest.java +++ b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStoreTest.java @@ -97,6 +97,8 @@ import software.amazon.awssdk.core.internal.async.ByteArrayAsyncResponseTransformer; import software.amazon.awssdk.core.internal.async.InputStreamResponseTransformer; import software.amazon.awssdk.http.SdkHttpResponse; +import software.amazon.awssdk.http.async.SdkAsyncHttpClient; +import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient; import software.amazon.awssdk.retries.api.RetryStrategy; import software.amazon.awssdk.services.s3.S3AsyncClient; import software.amazon.awssdk.services.s3.S3AsyncClientBuilder; @@ -2031,6 +2033,77 @@ void testBuildS3AsyncClientWithoutRetryConfig() { assertEquals(BUCKET, store.getBucket()); } + /** + * Builds an async client with the given builder while intercepting + * {@code NettyNioAsyncHttpClient.builder()} so the pool-hardening flags applied to the Netty HTTP + * client builder can be asserted. + */ + private NettyNioAsyncHttpClient.Builder captureNettyHttpClientFor( + AwsAsyncBlobStore.Builder builder) { + NettyNioAsyncHttpClient.Builder nettyBuilder = mock(NettyNioAsyncHttpClient.Builder.class); + when(nettyBuilder.useIdleConnectionReaper(any())).thenReturn(nettyBuilder); + when(nettyBuilder.tcpKeepAlive(any())).thenReturn(nettyBuilder); + when(nettyBuilder.maxConcurrency(any())).thenReturn(nettyBuilder); + when(nettyBuilder.writeTimeout(any())).thenReturn(nettyBuilder); + when(nettyBuilder.readTimeout(any())).thenReturn(nettyBuilder); + when(nettyBuilder.connectionMaxIdleTime(any())).thenReturn(nettyBuilder); + when(nettyBuilder.proxyConfiguration(any())).thenReturn(nettyBuilder); + when(nettyBuilder.build()).thenReturn(mock(SdkAsyncHttpClient.class)); + + try (MockedStatic nettyStatic = + mockStatic(NettyNioAsyncHttpClient.class)) { + nettyStatic.when(NettyNioAsyncHttpClient::builder).thenReturn(nettyBuilder); + builder.build(); + } + return nettyBuilder; + } + + @Test + void testDisableConnectionReaperWiredIntoAsyncHttpClient() { + AwsAsyncBlobStore.Builder builder = + (AwsAsyncBlobStore.Builder) + new AwsAsyncBlobStore.Builder() + .withBucket(BUCKET) + .withRegion(REGION) + .withDisableConnectionReaper(true); + + NettyNioAsyncHttpClient.Builder nettyBuilder = captureNettyHttpClientFor(builder); + + // disable=true must translate to useIdleConnectionReaper(false) + verify(nettyBuilder).useIdleConnectionReaper(false); + verify(nettyBuilder, times(0)).tcpKeepAlive(any()); + } + + @Test + void testTcpKeepAliveWiredIntoAsyncHttpClient() { + AwsAsyncBlobStore.Builder builder = + (AwsAsyncBlobStore.Builder) + new AwsAsyncBlobStore.Builder() + .withBucket(BUCKET) + .withRegion(REGION) + .withTcpKeepAlive(true); + + NettyNioAsyncHttpClient.Builder nettyBuilder = captureNettyHttpClientFor(builder); + + verify(nettyBuilder).tcpKeepAlive(true); + verify(nettyBuilder, times(0)).useIdleConnectionReaper(any()); + } + + @Test + void testPoolHardeningFlagsUntouchedWhenUnsetAsync() { + AwsAsyncBlobStore.Builder builder = + (AwsAsyncBlobStore.Builder) + new AwsAsyncBlobStore.Builder() + .withBucket(BUCKET) + .withRegion(REGION) + .withMaxConnections(50); + + NettyNioAsyncHttpClient.Builder nettyBuilder = captureNettyHttpClientFor(builder); + + verify(nettyBuilder, times(0)).useIdleConnectionReaper(any()); + verify(nettyBuilder, times(0)).tcpKeepAlive(any()); + } + @Test void testBuildS3AsyncClientWithUseSystemPropertyProxyValues() { var store = diff --git a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobStoreBuilder.java b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobStoreBuilder.java index d5bc0bd17..9c4aeb239 100644 --- a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobStoreBuilder.java +++ b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobStoreBuilder.java @@ -41,6 +41,8 @@ public abstract class BlobStoreBuilder implements SdkProvi private Boolean useEnvironmentVariableProxyValues; private String quotaProjectId; private TracingPolicy tracingPolicy; + private Boolean disableConnectionReaper; + private Boolean tcpKeepAlive; public BlobStoreBuilder providerId(String providerId) { this.providerId = providerId; @@ -132,6 +134,45 @@ public BlobStoreBuilder withIdleConnectionTimeout(Duration idleConnectionTime return this; } + /** + * Method to disable the HTTP client's background idle-connection reaper. Under bursty workloads + * the reaper can close pooled connections just before a traffic spike, forcing fresh TCP/TLS + * handshakes that add tail latency; disabling it keeps connections warm and available. + * + *

When left unset, the underlying SDK default is retained and client behavior is unchanged. + * + *

Provider support: AWS, GCP, and Alibaba Cloud. This maps to each provider's HTTP client + * idle-connection-reaper setting for both the synchronous and asynchronous connection pools. + * + * @param disable {@code true} to disable the idle-connection reaper + * @return An instance of self + */ + public BlobStoreBuilder withDisableConnectionReaper(boolean disable) { + this.disableConnectionReaper = disable; + return this; + } + + /** + * Method to enable TCP keep-alive on pooled connections. When enabled, the OS periodically probes + * idle connections so dead peers are detected and evicted before they are handed to a request, + * reducing sporadic failures on long-lived pools behind load balancers or NAT gateways. + * + *

When left unset, the underlying SDK default is retained and client behavior is unchanged. + * The actual keep-alive interval is governed by OS-level TCP settings. + * + *

Provider support: AWS and GCP. This maps to the HTTP client's TCP keep-alive socket setting + * for both the synchronous and asynchronous connection pools. The Alibaba Cloud SDK does not + * expose a TCP keep-alive setting, so it ignores this value; supplying it is a safe no-op and + * never throws. + * + * @param enable {@code true} to enable TCP keep-alive + * @return An instance of self + */ + public BlobStoreBuilder withTcpKeepAlive(boolean enable) { + this.tcpKeepAlive = enable; + return this; + } + /** * Method to supply credentialsOverrider * diff --git a/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStore.java b/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStore.java index 83eac433e..f9bb78cfd 100644 --- a/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStore.java +++ b/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStore.java @@ -126,6 +126,7 @@ import lombok.Getter; import org.apache.http.HttpHost; import org.apache.http.client.config.RequestConfig; +import org.apache.http.config.SocketConfig; import org.apache.http.conn.HttpClientConnectionManager; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClientBuilder; @@ -1721,7 +1722,8 @@ private static CloseableHttpClient buildHttpClient(Builder builder) { HttpClientBuilder httpClientBuilder = HttpClientBuilder.create(); httpClientBuilder.setDefaultRequestConfig(buildRequestConfig(builder)); httpClientBuilder.setConnectionManager(buildConnectionManager(builder)); - if (builder.getIdleConnectionTimeout() != null) { + boolean reaperEnabled = !Boolean.TRUE.equals(builder.getDisableConnectionReaper()); + if (reaperEnabled && builder.getIdleConnectionTimeout() != null) { httpClientBuilder.evictIdleConnections( builder.getIdleConnectionTimeout().toMillis(), TimeUnit.MILLISECONDS); } @@ -1735,6 +1737,16 @@ private static HttpClientConnectionManager buildConnectionManager(Builder builde ? builder.getMaxConnections() : DEFAULT_MAX_CONNECTIONS; connectionManager.setMaxTotal(maxConns); connectionManager.setDefaultMaxPerRoute(maxConns); + // TCP keep-alive is a socket-level option on the pooled connections. When enabled, the OS + // periodically probes idle sockets so dead peers are detected and evicted before a request is + // handed a stale connection. Left unset, the connection manager's default socket config is + // retained and behavior is unchanged. + if (builder.getTcpKeepAlive() != null) { + connectionManager.setDefaultSocketConfig( + SocketConfig.custom() + .setSoKeepAlive(builder.getTcpKeepAlive()) + .build()); + } return connectionManager; } diff --git a/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStoreTest.java b/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStoreTest.java index 3faad6a7f..4fc087782 100644 --- a/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStoreTest.java +++ b/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStoreTest.java @@ -127,6 +127,8 @@ import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.http.config.SocketConfig; +import org.apache.http.impl.client.HttpClientBuilder; import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -4587,4 +4589,100 @@ void testDoListBlobVersions_noSuchElementException() { assertThrows(NoSuchElementException.class, versions::next); } + + @Test + void testTcpKeepAliveEnabledSetsSocketConfig() throws Exception { + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) + new GcpBlobStore.Builder() + .withBucket(TEST_BUCKET) + .withRegion("us-west-2") + .withTcpKeepAlive(true); + + PoolingHttpClientConnectionManager manager = invokeBuildConnectionManager(builder); + + SocketConfig socketConfig = manager.getDefaultSocketConfig(); + assertTrue(socketConfig.isSoKeepAlive()); + } + + @Test + void testTcpKeepAliveDisabledSetsSocketConfigFalse() throws Exception { + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) + new GcpBlobStore.Builder() + .withBucket(TEST_BUCKET) + .withRegion("us-west-2") + .withTcpKeepAlive(false); + + PoolingHttpClientConnectionManager manager = invokeBuildConnectionManager(builder); + + assertFalse(manager.getDefaultSocketConfig().isSoKeepAlive()); + } + + @Test + void testTcpKeepAliveUnsetLeavesSocketConfigDefault() throws Exception { + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) + new GcpBlobStore.Builder() + .withBucket(TEST_BUCKET) + .withRegion("us-west-2") + .withMaxConnections(50); + + PoolingHttpClientConnectionManager manager = invokeBuildConnectionManager(builder); + + assertNull(manager.getDefaultSocketConfig()); + } + + /** + * Invokes the private static {@code buildHttpClient(Builder)} while intercepting + * {@code HttpClientBuilder.create()} so the idle-connection eviction wiring can be asserted + * without opening real connections. + */ + private HttpClientBuilder captureHttpClientBuilderFor(GcpBlobStore.Builder builder) + throws Exception { + HttpClientBuilder httpClientBuilder = mock(HttpClientBuilder.class); + when(httpClientBuilder.build()) + .thenReturn(mock(org.apache.http.impl.client.CloseableHttpClient.class)); + + Method method = + GcpBlobStore.Builder.class.getDeclaredMethod( + "buildHttpClient", GcpBlobStore.Builder.class); + method.setAccessible(true); + try (MockedStatic httpStatic = Mockito.mockStatic(HttpClientBuilder.class)) { + httpStatic.when(HttpClientBuilder::create).thenReturn(httpClientBuilder); + method.invoke(null, builder); + } + return httpClientBuilder; + } + + @Test + void testDisableConnectionReaperSuppressesIdleEviction() throws Exception { + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) + new GcpBlobStore.Builder() + .withBucket(TEST_BUCKET) + .withRegion("us-west-2") + .withIdleConnectionTimeout(Duration.ofMinutes(5)) + .withDisableConnectionReaper(true); + + HttpClientBuilder httpClientBuilder = captureHttpClientBuilderFor(builder); + + verify(httpClientBuilder, never()).evictIdleConnections(anyLong(), any(TimeUnit.class)); + } + + @Test + void testReaperEnabledHonorsIdleConnectionTimeout() throws Exception { + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) + new GcpBlobStore.Builder() + .withBucket(TEST_BUCKET) + .withRegion("us-west-2") + .withIdleConnectionTimeout(Duration.ofMinutes(5)) + .withDisableConnectionReaper(false); + + HttpClientBuilder httpClientBuilder = captureHttpClientBuilderFor(builder); + + verify(httpClientBuilder) + .evictIdleConnections(Duration.ofMinutes(5).toMillis(), TimeUnit.MILLISECONDS); + } }