From 6753636f5391e85313a7563d01f0ee2a1a496308 Mon Sep 17 00:00:00 2001 From: Aditya Jain Date: Sat, 15 Aug 2026 15:31:52 -0700 Subject: [PATCH 1/2] fix(sdk-core): avoid duplicate concurrent endpoint-discovery refresh calls EndpointDiscoveryRefreshCache#returnCachedOrDefaultEndpoint decides whether to kick off a background refresh of an expired cached endpoint with a plain check-then-act: read the cached entry's expirationTime, and if it's in the past, cache.put() a bumped-expiration copy and call refreshCacheAsync. Neither the read nor the put/decision is guarded, unlike the sibling "no cached entry yet" branch a few lines above, which correctly uses cache.putIfAbsent as a compare-and-swap so only one caller wins and triggers discovery. Every thread that reads the same expired entry before any of them writes back independently decides it's expired and independently calls refreshCacheAsync, firing duplicate calls against the real endpoint-discovery API for the same cache key -- exactly at the moment of highest concurrent load, right when an entry expires. Guard the refresh decision with cache.replace(key, oldValue, newValue), a compare-and-swap against the exact stale value each caller read, so only the thread that actually wins the race replaces the entry and triggers the refresh. --- .../EndpointDiscoveryRefreshCache.java | 13 +++- .../EndpointDiscoveryRefreshCacheTest.java | 69 +++++++++++++++++++ 2 files changed, 80 insertions(+), 2 deletions(-) diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/endpointdiscovery/EndpointDiscoveryRefreshCache.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/endpointdiscovery/EndpointDiscoveryRefreshCache.java index 988e49ac3ea8..a62cc6900437 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/endpointdiscovery/EndpointDiscoveryRefreshCache.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/endpointdiscovery/EndpointDiscoveryRefreshCache.java @@ -136,8 +136,17 @@ private URI returnCachedOrDefaultEndpoint(String key, EndpointDiscoveryEndpoint } if (endpoint.expirationTime().isBefore(Instant.now())) { - cache.put(key, endpoint.toBuilder().expirationTime(Instant.now().plusSeconds(60)).build()); - refreshCacheAsync(request, key); + EndpointDiscoveryEndpoint refreshedEndpoint = + endpoint.toBuilder().expirationTime(Instant.now().plusSeconds(60)).build(); + // CAS on the stale value each caller read: only the thread that actually wins the + // race to replace it kicks off a background refresh. Without this, every thread that + // reads the same expired entry before any of them writes back would independently + // decide it's expired and each fire its own refreshCacheAsync call, causing a + // thundering herd of duplicate endpoint-discovery requests right when the cache entry + // expires under concurrent load. + if (cache.replace(key, endpoint, refreshedEndpoint)) { + refreshCacheAsync(request, key); + } } return endpoint.endpoint(); diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/endpointdiscovery/EndpointDiscoveryRefreshCacheTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/endpointdiscovery/EndpointDiscoveryRefreshCacheTest.java index 8daa4f0909b4..1b4eb3b2d7aa 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/endpointdiscovery/EndpointDiscoveryRefreshCacheTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/endpointdiscovery/EndpointDiscoveryRefreshCacheTest.java @@ -22,10 +22,18 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import java.lang.reflect.Field; import java.net.URI; +import java.time.Instant; +import java.util.Map; import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -89,4 +97,65 @@ public void getAsync_future_cancelled() { } + @Test + public void get_concurrentCallsOnExpiredEntry_onlyRefreshesOnce() throws Exception { + // Regression test: returnCachedOrDefaultEndpoint used to check-then-act on an expired + // cache entry (isBefore(now) followed by an unguarded cache.put) with no synchronization, + // unlike the sibling "no cached entry yet" branch a few lines above it, which correctly + // uses cache.putIfAbsent as a compare-and-swap. Every thread that read the same expired + // entry before any of them wrote back independently decided to kick off a background + // refresh, causing duplicate discoverEndpoint calls under concurrent load right when an + // entry expires. There should only ever be exactly one refresh call per expiration. + AtomicInteger discoveryCalls = new AtomicInteger(0); + when(mockClient.discoverEndpoint(any())).thenAnswer(invocation -> { + discoveryCalls.incrementAndGet(); + return CompletableFuture.completedFuture( + EndpointDiscoveryEndpoint.builder() + .endpoint(testURI) + .expirationTime(Instant.now().plusSeconds(60)) + .build()); + }); + + EndpointDiscoveryRequest request = EndpointDiscoveryRequest.builder() + .required(false) + .cacheKey(requestCacheKey) + .defaultEndpoint(testURI) + .build(); + + // Prime the cache with an already-expired entry, simulating the moment right after a + // real cache entry expires under concurrent load. + Field cacheField = EndpointDiscoveryRefreshCache.class.getDeclaredField("cache"); + cacheField.setAccessible(true); + @SuppressWarnings("unchecked") + Map internalCache = + (Map) cacheField.get(endpointDiscoveryRefreshCache); + internalCache.put(accessKey + ":" + requestCacheKey, + EndpointDiscoveryEndpoint.builder() + .endpoint(URI.create("stale_endpoint")) + .expirationTime(Instant.now().minusSeconds(1)) + .build()); + + int threadCount = 50; + ExecutorService pool = Executors.newFixedThreadPool(threadCount); + CountDownLatch ready = new CountDownLatch(threadCount); + CountDownLatch go = new CountDownLatch(1); + for (int i = 0; i < threadCount; i++) { + pool.submit(() -> { + ready.countDown(); + try { + go.await(); + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + } + endpointDiscoveryRefreshCache.get(accessKey, request); + }); + } + ready.await(); + go.countDown(); + pool.shutdown(); + assertThat(pool.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + + assertThat(discoveryCalls.get()).isEqualTo(1); + } + } From b8d89cd94176ac066ecac14c5b96b43e7215eacb Mon Sep 17 00:00:00 2001 From: Aditya Jain Date: Sat, 15 Aug 2026 15:32:18 -0700 Subject: [PATCH 2/2] Add changelog entry for endpoint-discovery refresh race fix --- .changes/next-release/bugfix-AWSSDKforJavav2-08fe1ef1.json | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changes/next-release/bugfix-AWSSDKforJavav2-08fe1ef1.json diff --git a/.changes/next-release/bugfix-AWSSDKforJavav2-08fe1ef1.json b/.changes/next-release/bugfix-AWSSDKforJavav2-08fe1ef1.json new file mode 100644 index 000000000000..aa8871547365 --- /dev/null +++ b/.changes/next-release/bugfix-AWSSDKforJavav2-08fe1ef1.json @@ -0,0 +1,5 @@ +{ + "type": "bugfix", + "category": "AWS SDK for Java v2", + "description": "Fixed a race condition in `EndpointDiscoveryRefreshCache` where multiple concurrent threads reading the same expired cache entry could each independently trigger a background endpoint-discovery refresh call for the same key, instead of only one." +} \ No newline at end of file