fix(sdk-core): avoid duplicate concurrent endpoint-discovery refresh calls

.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."
+}

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();

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<String, EndpointDiscoveryEndpoint> internalCache =
+            (Map<String, EndpointDiscoveryEndpoint>) 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);
+    }
+
 }