Skip to content

blob: harden HTTP connection pools across clouds to cut tail latency under bursty traffic #533

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hkhiri wants to merge 1 commit into
base: main
Choose a base branch
from
Open

blob: harden HTTP connection pools across clouds to cut tail latency under bursty traffic #533

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/AliBlobStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

Expand Down
25 changes: 17 additions & 8 deletions blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/OssClientFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
* <p>{@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);

@LihaoLiuXs LihaoLiuXs Aug 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This SAM is being reshaped concurrently by #532, and the two changes are not composable as written.

This PR adds a 5th positional parameter here and widens the gate below (|| getDisableConnectionReaper() != null) so the flag forces an explicitly constructed transport. #532 widens that same gate for metricsPublisher and adds AliInstrumentedHttpClientFactory.instrument(...), whose javadoc states it deliberately wraps "the one the SDK builder already produced, so all of the SDK's default transport behavior (TLS trust config, idle-connection reaping, proxy, timeouts, pool sizing) is preserved untouched" — the assumption this PR invalidates.

Whichever lands second has to reconcile by hand inside the AliBlobStore/AliAsyncBlobStore lambdas: useReaper(...) must be applied to the builder and the concrete built client passed through instrument(...). The instrument overloads are typed on Apache5HttpClient/Apache5AsyncHttpClient rather than HttpClient, so the composition order is load-bearing and it is easy to silently drop the reaper flag or the metrics wrapper while resolving the conflict.

Suggested fix: replace the growing positional list with a small options/config object (proxyHost, readWriteTimeout, maxConnections, idleConnectionTimeout, disableConnectionReaper, …) so both changes extend one type instead of the same signature, and agree on a merge order.

}

/**
Expand Down Expand Up @@ -89,18 +93,23 @@ public static <B extends BaseClientBuilder<B, T>, 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);
}
Expand Down
36 changes: 26 additions & 10 deletions blob/blob-ali/src/main/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

Expand All @@ -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();
}

Expand Down
88 changes: 88 additions & 0 deletions blob/blob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/AliBlobStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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 {

Expand Down Expand Up @@ -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<Apache5HttpClientBuilder> 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<Apache5HttpClientBuilder> apacheStatic =
mockStatic(Apache5HttpClientBuilder.class)) {
apacheStatic.when(Apache5HttpClientBuilder::create).thenReturn(apacheBuilder);
builder.build();
}

verify(apacheBuilder, never()).useReaper(true);
verify(apacheBuilder, never()).useReaper(false);
}
}
81 changes: 81 additions & 0 deletions ...ob-ali/src/test/java/com/salesforce/multicloudj/blob/ali/async/AliAsyncBlobStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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 {

Expand Down Expand Up @@ -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<Apache5AsyncHttpClientBuilder> asyncStatic =
mockStatic(Apache5AsyncHttpClientBuilder.class);
MockedStatic<Apache5HttpClientBuilder> 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<Apache5AsyncHttpClientBuilder> asyncStatic =
mockStatic(Apache5AsyncHttpClientBuilder.class);
MockedStatic<Apache5HttpClientBuilder> 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);
}
}
20 changes: 19 additions & 1 deletion blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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();
}

Expand Down
Loading
Loading