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 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); + } + }