From dd6a1a0d9d941a871eed62aa36ef2fd15527da4d Mon Sep 17 00:00:00 2001 From: jencymaryjoseph <35571282+jencymaryjoseph@users.noreply.github.com> Date: Fri, 2 Oct 2026 10:27:05 -0700 Subject: [PATCH 1/4] Isolate S3 CRT metric publishers from event loop --- .../services/s3/S3CrtAsyncClientBuilder.java | 6 +- .../s3/internal/crt/S3CrtAsyncHttpClient.java | 13 +- .../crt/S3CrtMetricPublisherDispatcher.java | 173 ++++++++++ .../crt/S3CrtResponseHandlerAdapter.java | 15 +- .../crt/S3CrtAsyncHttpClientTest.java | 27 +- .../S3CrtMetricPublisherDispatcherTest.java | 311 ++++++++++++++++++ ...3CrtResponseHandlerAdapterMetricsTest.java | 50 ++- .../crt/S3CrtResponseHandlerAdapterTest.java | 10 + 8 files changed, 580 insertions(+), 25 deletions(-) create mode 100644 services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java create mode 100644 services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java index 9a0c9e9bbdd1..7e54dee61ba1 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java @@ -390,7 +390,11 @@ default S3CrtAsyncClientBuilder retryConfiguration(ConsumerThe SDK does not close these publishers when the client is closed; the caller retains ownership and is - * responsible for closing them. + * responsible for closing them. Metric publication is asynchronous and best effort. Closing the client initiates + * shutdown of metric publication, but may return while previously queued metric collections are being delivered. + * The SDK attempts to drain accepted collections for up to five seconds and then drops any that remain queued. If a + * publisher is closed concurrently with this delivery, those metric collections may not be published and a warning + * may be logged. This does not affect the outcome of the S3 operation. * * @param metricPublishers the metric publishers to use * @return this builder for method chaining. diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java index 6c91688a940f..e9cb73c950d7 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java @@ -82,11 +82,13 @@ public final class S3CrtAsyncHttpClient implements SdkAsyncHttpClient { private final S3NativeClientConfiguration s3NativeClientConfiguration; private final S3ClientOptions s3ClientOptions; private final List metricPublishers; + private final S3CrtMetricPublisherDispatcher metricPublisherDispatcher; private S3CrtAsyncHttpClient(Builder builder) { s3NativeClientConfiguration = builder.clientConfiguration; this.s3ClientOptions = createS3ClientOption(); this.metricPublishers = builder.metricPublishers == null ? Collections.emptyList() : builder.metricPublishers; + this.metricPublisherDispatcher = new S3CrtMetricPublisherDispatcher(); this.crtS3Client = new S3Client(s3ClientOptions); } @@ -97,6 +99,7 @@ private S3CrtAsyncHttpClient(Builder builder) { s3NativeClientConfiguration = builder.clientConfiguration; s3ClientOptions = createS3ClientOption(); this.metricPublishers = builder.metricPublishers == null ? Collections.emptyList() : builder.metricPublishers; + this.metricPublisherDispatcher = new S3CrtMetricPublisherDispatcher(); this.crtS3Client = crtS3Client; } @@ -182,6 +185,7 @@ public CompletableFuture execute(AsyncExecuteRequest asyncRequest) { .httpExecutionAttributes(httpExecutionAttributes) .metaRequestFuture(s3MetaRequestFuture) .metricPublishers(effectivePublishers) + .metricPublisherDispatcher(metricPublisherDispatcher) .maxConcurrency(effectivePublishers.isEmpty() ? null : crtS3Client.getMaxActiveConnections()) .build(); @@ -339,8 +343,13 @@ private static HttpRequest toCrtRequest(AsyncExecuteRequest asyncRequest, Comple @Override public void close() { - s3NativeClientConfiguration.close(); - crtS3Client.close(); + CompletableFuture shutdownCompleteFuture = crtS3Client.getShutdownCompleteFuture(); + shutdownCompleteFuture.whenComplete((ignored, throwable) -> metricPublisherDispatcher.startAsyncClose()); + try { + crtS3Client.close(); + } finally { + s3NativeClientConfiguration.close(); + } } public static Builder builder() { diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java new file mode 100644 index 000000000000..1a51ba9eec8d --- /dev/null +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java @@ -0,0 +1,173 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.services.s3.internal.crt; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import software.amazon.awssdk.annotations.SdkInternalApi; +import software.amazon.awssdk.annotations.SdkTestInternalApi; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; +import software.amazon.awssdk.utils.Logger; +import software.amazon.awssdk.utils.SdkAutoCloseable; +import software.amazon.awssdk.utils.ThreadFactoryBuilder; + +/** + * Publishes CRT metric collections away from CRT native callback threads. + * + *

Collections are submitted without blocking to a single bounded queue and published in acceptance order by one + * daemon worker. A slow or blocked publisher can delay the other publishers configured for the client, but it cannot + * block a CRT callback thread or cause the queue to grow without bound. + */ +@SdkInternalApi +final class S3CrtMetricPublisherDispatcher implements SdkAutoCloseable { + private static final Logger log = Logger.loggerFor(S3CrtMetricPublisherDispatcher.class); + private static final int DEFAULT_QUEUE_CAPACITY = 128; + private static final Duration DEFAULT_CLOSE_TIMEOUT = Duration.ofSeconds(5); + private static final ThreadFactory CLOSE_THREAD_FACTORY = + new ThreadFactoryBuilder() + .daemonThreads(true) + .threadNamePrefix("s3-crt-metric-publisher-cleanup") + .build(); + + private final Object lock = new Object(); + private final ThreadPoolExecutor executor; + private final int queueCapacity; + private final Duration closeTimeout; + private final AtomicLong queueFullDropCount = new AtomicLong(); + private final AtomicLong closedDispatcherDropCount = new AtomicLong(); + private final AtomicBoolean asyncCloseStarted = new AtomicBoolean(); + private boolean closed; + + S3CrtMetricPublisherDispatcher() { + this(DEFAULT_QUEUE_CAPACITY, DEFAULT_CLOSE_TIMEOUT); + } + + @SdkTestInternalApi + S3CrtMetricPublisherDispatcher(int queueCapacity, Duration closeTimeout) { + if (queueCapacity <= 0) { + throw new IllegalArgumentException("queueCapacity must be greater than zero"); + } + if (closeTimeout.isNegative()) { + throw new IllegalArgumentException("closeTimeout must not be negative"); + } + this.queueCapacity = queueCapacity; + this.closeTimeout = closeTimeout; + this.executor = new ThreadPoolExecutor( + 1, 1, + 0, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(queueCapacity), + new ThreadFactoryBuilder() + .daemonThreads(true) + .threadNamePrefix("s3-crt-metric-publisher") + .build()); + } + + /** + * Dispatches one collection to all configured publishers as a single queue task. Synchronization establishes one + * acceptance order across concurrent CRT callbacks and coordinates submission with close. Publisher code is never + * invoked while holding the lock. + */ + void dispatch(MetricCollection collection, List publishers) { + synchronized (lock) { + if (closed) { + recordClosedDispatcherDrop(); + return; + } + + try { + executor.execute(() -> publish(collection, publishers)); + } catch (RejectedExecutionException e) { + recordQueueFullDrop(); + } + } + } + + private void publish(MetricCollection collection, List publishers) { + for (MetricPublisher publisher : publishers) { + try { + publisher.publish(collection); + } catch (RuntimeException e) { + log.warn(() -> "Failed to publish CRT S3 metrics through " + publisher.getClass().getName(), e); + } + } + } + + private void recordQueueFullDrop() { + long dropped = queueFullDropCount.incrementAndGet(); + if (isPowerOfTwo(dropped)) { + log.warn(() -> "Dropped CRT S3 metric collections because the metric publisher queue is full. " + + "Queue capacity: " + queueCapacity + ", total dropped: " + dropped + "."); + } + } + + private void recordClosedDispatcherDrop() { + long dropped = closedDispatcherDropCount.incrementAndGet(); + if (isPowerOfTwo(dropped)) { + log.warn(() -> "Dropped CRT S3 metric collections because the metric publisher dispatcher is closed. " + + "Total dropped after close: " + dropped + "."); + } + } + + void startAsyncClose() { + if (asyncCloseStarted.compareAndSet(false, true)) { + CLOSE_THREAD_FACTORY.newThread(this::close).start(); + } + } + + @Override + public void close() { + synchronized (lock) { + if (closed) { + return; + } + closed = true; + executor.shutdown(); + } + + boolean terminated = false; + try { + terminated = executor.awaitTermination(closeTimeout.toNanos(), TimeUnit.NANOSECONDS); + } catch (InterruptedException e) { + int queuedCollectionCount = forceShutdown(); + Thread.currentThread().interrupt(); + log.warn(() -> "Interrupted while draining CRT S3 metric publisher queue. Undelivered queued collections: " + + queuedCollectionCount + "."); + return; + } + + if (!terminated) { + int queuedCollectionCount = forceShutdown(); + log.warn(() -> "Timed out draining CRT S3 metric publisher queue. Undelivered queued collections: " + + queuedCollectionCount + "."); + } + } + + private int forceShutdown() { + return executor.shutdownNow().size(); + } + + private static boolean isPowerOfTwo(long value) { + return (value & (value - 1)) == 0; + } +} diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java index 4f512f536765..941c7db99cae 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java @@ -83,6 +83,7 @@ public final class S3CrtResponseHandlerAdapter implements S3MetaRequestResponseH private final Duration s3MetaRequestTimeout; private final List metricPublishers; private final Integer maxConcurrency; + private final S3CrtMetricPublisherDispatcher metricPublisherDispatcher; private volatile boolean responseHandlingInitiated; @@ -100,6 +101,7 @@ private S3CrtResponseHandlerAdapter(Builder builder) { this.metricPublishers = builder.metricPublishers; this.maxConcurrency = builder.maxConcurrency; this.s3MetaRequestTimeout = builder.s3MetaRequestTimeout; + this.metricPublisherDispatcher = builder.metricPublisherDispatcher; // Registered last: the callback reads responseHandler and s3MetaRequestTimeout, so all fields must be assigned // first in case resultFuture is already complete and the callback runs synchronously here. @@ -132,6 +134,7 @@ public static final class Builder { private Duration s3MetaRequestTimeout = META_REQUEST_TIMEOUT; private List metricPublishers = Collections.emptyList(); private Integer maxConcurrency; + private S3CrtMetricPublisherDispatcher metricPublisherDispatcher; private Builder() { } @@ -171,6 +174,11 @@ public Builder maxConcurrency(Integer maxConcurrency) { return this; } + Builder metricPublisherDispatcher(S3CrtMetricPublisherDispatcher metricPublisherDispatcher) { + this.metricPublisherDispatcher = metricPublisherDispatcher; + return this; + } + public S3CrtResponseHandlerAdapter build() { return new S3CrtResponseHandlerAdapter(this); } @@ -389,7 +397,8 @@ public void onProgress(S3MetaRequestProgress progress) { * Invoked by CRT once per underlying HTTP request (e.g. each ranged part GET / upload part) with that request's * telemetry. We publish each as its own nested ApiCall -> ApiCallAttempt -> HttpClient MetricCollection, mirroring * the node placement of the standard client so all metric publishers render it consistently. Runs on a CRT native - * thread; it only reads the metrics object and publishes, and never throws back into the native callback. + * thread; it reads and copies the metrics object before dispatching the completed collection to the client-scoped + * bounded queue and single worker, and never throws back into the native callback. */ @Override public void onTelemetry(S3RequestMetrics requestMetrics) { @@ -425,9 +434,9 @@ public void onTelemetry(S3RequestMetrics requestMetrics) { reportHttpManagerMetrics(httpClient, requestMetrics.getHttpManagerMetrics()); MetricCollection collection = apiCall.collect(); - metricPublishers.forEach(p -> p.publish(collection)); + metricPublisherDispatcher.dispatch(collection, metricPublishers); } catch (RuntimeException e) { - log.warn(() -> "Failed to publish CRT S3 request metrics", e); + log.warn(() -> "Failed to collect CRT S3 request metrics", e); } } diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java index d5c4848bc780..826f84d9ba51 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption.CRT_MEMORY_BUFFER_DISABLED; @@ -97,6 +98,7 @@ public class S3CrtAsyncHttpClientTest { @BeforeEach public void methodSetup() { s3Client = Mockito.mock(S3Client.class); + when(s3Client.getShutdownCompleteFuture()).thenReturn(CompletableFuture.completedFuture(null)); responseHandler = Mockito.mock(SdkAsyncHttpResponseHandler.class); contentPublisher = Mockito.mock(SdkHttpContentPublisher.class); s3NativeClientConfiguration = S3NativeClientConfiguration.builder() @@ -166,16 +168,21 @@ public void execute_withMetricPublisher_publishesCrtMaxConcurrency() { .s3ClientConfiguration(s3NativeClientConfiguration) .metricPublishers(Collections.singletonList(publisher))); - S3MetaRequestOptions requestOptions = makeRequest(getExecuteRequestBuilder().build()); - requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); - - ArgumentCaptor metricCollectionCaptor = ArgumentCaptor.forClass(MetricCollection.class); - verify(publisher).publish(metricCollectionCaptor.capture()); - MetricCollection attempt = - metricCollectionCaptor.getValue().childrenWithName("ApiCallAttempt").findFirst().orElseThrow(AssertionError::new); - MetricCollection httpClient = - attempt.childrenWithName("HttpClient").findFirst().orElseThrow(AssertionError::new); - assertThat(httpClient.metricValues(HttpMetric.MAX_CONCURRENCY)).containsExactly(expectedMaxConcurrency); + try { + S3MetaRequestOptions requestOptions = makeRequest(getExecuteRequestBuilder().build()); + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + + ArgumentCaptor metricCollectionCaptor = ArgumentCaptor.forClass(MetricCollection.class); + verify(publisher, timeout(1_000)).publish(metricCollectionCaptor.capture()); + MetricCollection attempt = + metricCollectionCaptor.getValue().childrenWithName("ApiCallAttempt").findFirst() + .orElseThrow(AssertionError::new); + MetricCollection httpClient = + attempt.childrenWithName("HttpClient").findFirst().orElseThrow(AssertionError::new); + assertThat(httpClient.metricValues(HttpMetric.MAX_CONCURRENCY)).containsExactly(expectedMaxConcurrency); + } finally { + asyncHttpClient.close(); + } } @Test diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java new file mode 100644 index 000000000000..221634abbc14 --- /dev/null +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java @@ -0,0 +1,311 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.services.s3.internal.crt; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricCollector; +import software.amazon.awssdk.metrics.MetricPublisher; + +public class S3CrtMetricPublisherDispatcherTest { + + @Test + public void dispatch_blockingPublisher_doesNotBlockCallingThread() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(1)); + BlockingPublisher publisher = new BlockingPublisher(); + AtomicReference dispatchThread = new AtomicReference<>(); + + try { + CompletableFuture dispatch = + CompletableFuture.runAsync(() -> { + dispatchThread.set(Thread.currentThread()); + dispatcher.dispatch(collection("one"), Collections.singletonList(publisher)); + }); + + dispatch.get(1, TimeUnit.SECONDS); + assertThat(publisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(publisher.publishThread.get()).isNotSameAs(dispatchThread.get()); + assertThat(publisher.publishThread.get().isDaemon()).isTrue(); + } finally { + publisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void dispatch_blockingPublisher_delaysOtherPublisherWithoutBlockingCaller() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(1)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + CapturingPublisher healthyPublisher = new CapturingPublisher(1); + + try { + dispatcher.dispatch(collection("one"), Arrays.asList(blockingPublisher, healthyPublisher)); + + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.completed.await(50, TimeUnit.MILLISECONDS)).isFalse(); + + blockingPublisher.release.countDown(); + assertThat(healthyPublisher.completed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.collectionNames).containsExactly("one"); + } finally { + blockingPublisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void dispatch_multiplePublishers_preservesGlobalAcceptanceOrder() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(1)); + List publications = new CopyOnWriteArrayList<>(); + CountDownLatch completed = new CountDownLatch(6); + MetricPublisher firstPublisher = recordingPublisher("first", publications, completed); + MetricPublisher secondPublisher = recordingPublisher("second", publications, completed); + List publishers = Arrays.asList(firstPublisher, secondPublisher); + + try { + dispatcher.dispatch(collection("one"), publishers); + dispatcher.dispatch(collection("two"), publishers); + dispatcher.dispatch(collection("three"), publishers); + + assertThat(completed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(publications).containsExactly("first-one", "second-one", + "first-two", "second-two", + "first-three", "second-three"); + } finally { + dispatcher.close(); + } + } + + @Test + public void dispatch_fullQueue_dropsWholeCollection() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofSeconds(1)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + CapturingPublisher healthyPublisher = new CapturingPublisher(2); + List publishers = Arrays.asList(blockingPublisher, healthyPublisher); + + try { + dispatcher.dispatch(collection("one"), publishers); + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + + dispatcher.dispatch(collection("two"), publishers); + dispatcher.dispatch(collection("three"), publishers); + + blockingPublisher.release.countDown(); + + assertThat(healthyPublisher.completed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.collectionNames).containsExactly("one", "two"); + } finally { + blockingPublisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void dispatch_publisherThrows_doesNotAffectRemainingPublishersAndWorkerContinues() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(1)); + ThrowingFirstPublisher throwingPublisher = new ThrowingFirstPublisher(); + CapturingPublisher healthyPublisher = new CapturingPublisher(2); + List publishers = Arrays.asList(throwingPublisher, healthyPublisher); + + try { + dispatcher.dispatch(collection("one"), publishers); + dispatcher.dispatch(collection("two"), publishers); + + assertThat(throwingPublisher.secondCallCompleted.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.completed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.collectionNames).containsExactly("one", "two"); + } finally { + dispatcher.close(); + } + } + + @Test + public void close_drainsQueuedCollectionsWithoutClosingPublishers() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(2)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + MetricPublisher publisherSpy = mock(MetricPublisher.class); + + dispatcher.dispatch(collection("one"), Collections.singletonList(blockingPublisher)); + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + dispatcher.dispatch(collection("two"), Collections.singletonList(publisherSpy)); + + CompletableFuture closeFuture = CompletableFuture.runAsync(dispatcher::close); + blockingPublisher.release.countDown(); + + closeFuture.get(1, TimeUnit.SECONDS); + verify(publisherSpy).publish(collectionNamed("two")); + verify(publisherSpy, never()).close(); + } + + @Test + public void close_blockedPublisher_returnsAfterTimeoutAndDropsQueuedCollections() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofMillis(50)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + MetricPublisher queuedPublisher = mock(MetricPublisher.class); + + try { + dispatcher.dispatch(collection("one"), Collections.singletonList(blockingPublisher)); + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + dispatcher.dispatch(collection("two"), Collections.singletonList(queuedPublisher)); + + long startNanos = System.nanoTime(); + dispatcher.close(); + long closeDurationMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + + assertThat(closeDurationMillis).isLessThan(1_000); + verify(queuedPublisher, never()).publish(org.mockito.ArgumentMatchers.any()); + } finally { + blockingPublisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void startAsyncClose_blockedPublisher_doesNotBlockCallingThread() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofSeconds(1)); + BlockingPublisher publisher = new BlockingPublisher(); + + try { + dispatcher.dispatch(collection("one"), Collections.singletonList(publisher)); + assertThat(publisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + + CompletableFuture.runAsync(dispatcher::startAsyncClose).get(1, TimeUnit.SECONDS); + } finally { + publisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void dispatch_closedDispatcher_dropsWithoutInvokingPublisher() { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofSeconds(1)); + MetricPublisher publisher = mock(MetricPublisher.class); + + dispatcher.close(); + dispatcher.dispatch(collection("one"), Collections.singletonList(publisher)); + + verify(publisher, never()).publish(org.mockito.ArgumentMatchers.any()); + } + + private static MetricPublisher recordingPublisher(String publisherName, + List publications, + CountDownLatch completed) { + return new MetricPublisher() { + @Override + public void publish(MetricCollection metricCollection) { + publications.add(publisherName + "-" + metricCollection.name()); + completed.countDown(); + } + + @Override + public void close() { + } + }; + } + + private static MetricCollection collection(String name) { + return MetricCollector.create(name).collect(); + } + + private static MetricCollection collectionNamed(String name) { + return org.mockito.ArgumentMatchers.argThat(collection -> collection != null && name.equals(collection.name())); + } + + private static final class BlockingPublisher implements MetricPublisher { + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private final AtomicReference publishThread = new AtomicReference<>(); + + @Override + public void publish(MetricCollection metricCollection) { + publishThread.set(Thread.currentThread()); + started.countDown(); + boolean released = false; + while (!released) { + try { + release.await(); + released = true; + } catch (InterruptedException ignored) { + // Deliberately model a publisher that does not respond to interruption. + } + } + } + + @Override + public void close() { + } + } + + private static final class CapturingPublisher implements MetricPublisher { + private final List collectionNames = new CopyOnWriteArrayList<>(); + private final CountDownLatch completed; + + private CapturingPublisher(int expectedCollections) { + this.completed = new CountDownLatch(expectedCollections); + } + + @Override + public void publish(MetricCollection metricCollection) { + collectionNames.add(metricCollection.name()); + completed.countDown(); + } + + @Override + public void close() { + } + } + + private static final class ThrowingFirstPublisher implements MetricPublisher { + private final AtomicInteger calls = new AtomicInteger(); + private final CountDownLatch secondCallCompleted = new CountDownLatch(1); + + @Override + public void publish(MetricCollection metricCollection) { + if (calls.incrementAndGet() == 1) { + throw new IllegalStateException("expected"); + } + secondCallCompleted.countDown(); + } + + @Override + public void close() { + } + } +} diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterMetricsTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterMetricsTest.java index 6b5ba22fab81..efe88eb657e6 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterMetricsTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterMetricsTest.java @@ -22,10 +22,14 @@ import java.net.URI; import java.time.Duration; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import software.amazon.awssdk.core.metrics.CoreMetric; import software.amazon.awssdk.crt.CrtRuntimeException; @@ -54,11 +58,23 @@ public class S3CrtResponseHandlerAdapterMetricsTest { private static final long CONNECTION_ACQUISITION_NANOS = 2_000_000L; private static final String SERVICE_ENDPOINT = "https://s3.us-west-2.amazonaws.com"; private static final int MAX_CONCURRENCY = 10; + private S3CrtMetricPublisherDispatcher metricPublisherDispatcher; + + @BeforeEach + public void setup() { + metricPublisherDispatcher = new S3CrtMetricPublisherDispatcher(); + } + + @AfterEach + public void teardown() { + metricPublisherDispatcher.close(); + } @Test public void onTelemetry_publishesApiCallTreeWithMappedValues() throws Exception { - CapturingPublisher publisher = new CapturingPublisher(); + CapturingPublisher publisher = new CapturingPublisher(1); adapterPublishingTo(publisher).onTelemetry(successfulGetObjectMetrics()); + publisher.awaitExpectedCollections(); assertThat(publisher.collections).hasSize(1); MetricCollection apiCall = publisher.collections.get(0); @@ -98,8 +114,9 @@ public void onTelemetry_unavailableMetric_isSkipped_othersStillPublished() throw // CRT throws this when a datum is not available for the request; only that one metric should be skipped. when(metrics.getBackoffDelayDurationNs()).thenThrow(new CrtRuntimeException(AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE)); - CapturingPublisher publisher = new CapturingPublisher(); + CapturingPublisher publisher = new CapturingPublisher(1); adapterPublishingTo(publisher).onTelemetry(metrics); + publisher.awaitExpectedCollections(); assertThat(publisher.collections).hasSize(1); MetricCollection attempt = childNamed(publisher.collections.get(0), "ApiCallAttempt"); @@ -117,8 +134,9 @@ public void onTelemetry_optionalHttpMetricsUnavailable_areSkipped() throws Excep when(metrics.getResponseStatus()).thenReturn(-1); when(metrics.getHttpManagerMetrics()).thenReturn(null); - CapturingPublisher publisher = new CapturingPublisher(); + CapturingPublisher publisher = new CapturingPublisher(1); adapterPublishingTo(publisher).onTelemetry(metrics); + publisher.awaitExpectedCollections(); assertThat(publisher.collections).hasSize(1); MetricCollection attempt = childNamed(publisher.collections.get(0), "ApiCallAttempt"); @@ -139,8 +157,9 @@ public void onTelemetry_failedAttempt_reportsApiCallUnsuccessful() throws Except S3RequestMetrics metrics = successfulGetObjectMetrics(); when(metrics.isApiCallSuccessful()).thenReturn(false); - CapturingPublisher publisher = new CapturingPublisher(); + CapturingPublisher publisher = new CapturingPublisher(1); adapterPublishingTo(publisher).onTelemetry(metrics); + publisher.awaitExpectedCollections(); MetricCollection apiCall = publisher.collections.get(0); assertThat(apiCall.metricValues(CoreMetric.API_CALL_SUCCESSFUL)).containsExactly(false); @@ -157,20 +176,21 @@ public void onTelemetry_noPublishers_isNoOp_andReadsNoMetrics() { @Test public void onTelemetry_calledPerAttempt_publishesOneCollectionEach() throws Exception { - CapturingPublisher publisher = new CapturingPublisher(); + CapturingPublisher publisher = new CapturingPublisher(3); S3CrtResponseHandlerAdapter adapter = adapterPublishingTo(publisher); S3RequestMetrics metrics = successfulGetObjectMetrics(); adapter.onTelemetry(metrics); adapter.onTelemetry(metrics); adapter.onTelemetry(metrics); + publisher.awaitExpectedCollections(); assertThat(publisher.collections).hasSize(3); } @Test public void onTelemetry_retriedRequest_publishesSeparateCollectionsWithIncreasingRetryCount() throws Exception { - CapturingPublisher publisher = new CapturingPublisher(); + CapturingPublisher publisher = new CapturingPublisher(2); S3CrtResponseHandlerAdapter adapter = adapterPublishingTo(publisher); // First attempt fails (retry count 0), the retry succeeds (retry count 1). CRT delivers these as two separate @@ -183,6 +203,7 @@ public void onTelemetry_retriedRequest_publishesSeparateCollectionsWithIncreasin S3RequestMetrics retryAttempt = successfulGetObjectMetrics(); when(retryAttempt.getRetryCount()).thenReturn(1); adapter.onTelemetry(retryAttempt); + publisher.awaitExpectedCollections(); assertThat(publisher.collections).hasSize(2); assertThat(publisher.collections.get(0).metricValues(CoreMetric.API_CALL_SUCCESSFUL)).containsExactly(false); @@ -217,13 +238,14 @@ private static S3RequestMetrics successfulGetObjectMetrics() throws Exception { return metrics; } - private static S3CrtResponseHandlerAdapter adapterPublishingTo(MetricPublisher... publishers) { + private S3CrtResponseHandlerAdapter adapterPublishingTo(MetricPublisher... publishers) { return S3CrtResponseHandlerAdapter.builder() .executeFuture(new CompletableFuture<>()) .responseHandler(mock(SdkAsyncHttpResponseHandler.class)) .httpExecutionAttributes(SdkHttpExecutionAttributes.builder().build()) .metaRequestFuture(new CompletableFuture<>()) .metricPublishers(Arrays.asList(publishers)) + .metricPublisherDispatcher(metricPublisherDispatcher) .maxConcurrency(MAX_CONCURRENCY) .build(); } @@ -233,15 +255,25 @@ private static MetricCollection childNamed(MetricCollection parent, String name) } private static final class CapturingPublisher implements MetricPublisher { - private final List collections = new ArrayList<>(); + private final List collections = new CopyOnWriteArrayList<>(); + private final CountDownLatch expectedCollections; + + private CapturingPublisher(int expectedCollections) { + this.expectedCollections = new CountDownLatch(expectedCollections); + } @Override public void publish(MetricCollection metricCollection) { collections.add(metricCollection); + expectedCollections.countDown(); } @Override public void close() { } + + private void awaitExpectedCollections() throws InterruptedException { + assertThat(expectedCollections.await(5, TimeUnit.SECONDS)).isTrue(); + } } } diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java index e1bc05bb09c5..b02899aeb40c 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java @@ -33,6 +33,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import org.junit.After; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -64,11 +65,13 @@ public class S3CrtResponseHandlerAdapterTest { @Mock private S3MetaRequestWrapper s3MetaRequest; private CompletableFuture future; + private S3CrtMetricPublisherDispatcher metricPublisherDispatcher; @Before public void setup() { future = new CompletableFuture<>(); sdkResponseHandler = spy(new TestResponseHandler()); + metricPublisherDispatcher = new S3CrtMetricPublisherDispatcher(); responseHandlerAdapter = S3CrtResponseHandlerAdapter.builder() .executeFuture(future) .responseHandler(sdkResponseHandler) @@ -76,9 +79,15 @@ public void setup() { SdkHttpExecutionAttributes.builder().build()) .metaRequestFuture( CompletableFuture.completedFuture(s3MetaRequest)) + .metricPublisherDispatcher(metricPublisherDispatcher) .build(); } + @After + public void teardown() { + metricPublisherDispatcher.close(); + } + @Test public void successfulResponse_shouldCompleteFutureSuccessfully() throws Exception { HttpHeader[] httpHeaders = new HttpHeader[2]; @@ -113,6 +122,7 @@ public void s3MetaRequestNotFinish_shouldFailFuture() throws Exception { .httpExecutionAttributes(SdkHttpExecutionAttributes.builder().build()) .metaRequestFuture(new CompletableFuture<>()) .s3MetaRequestTimeout(Duration.ofMillis(10)) + .metricPublisherDispatcher(metricPublisherDispatcher) .build(); int statusCode = 200; responseHandlerAdapter.onResponseHeaders(statusCode, new HttpHeader[0]); From ad2a311d72586af9cd20d566c78840e34be4adf6 Mon Sep 17 00:00:00 2001 From: jencymaryjoseph <35571282+jencymaryjoseph@users.noreply.github.com> Date: Tue, 27 Oct 2026 10:54:13 -0700 Subject: [PATCH 2/4] Use debug logging for S3 CRT metric publishing --- .../services/s3/S3CrtAsyncClientBuilder.java | 4 ++-- .../crt/S3CrtMetricPublisherDispatcher.java | 18 +++++++++--------- .../crt/S3CrtResponseHandlerAdapter.java | 2 +- 3 files changed, 12 insertions(+), 12 deletions(-) diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java index 7e54dee61ba1..b93849e908d8 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java @@ -393,8 +393,8 @@ default S3CrtAsyncClientBuilder retryConfiguration(Consumer publishe try { publisher.publish(collection); } catch (RuntimeException e) { - log.warn(() -> "Failed to publish CRT S3 metrics through " + publisher.getClass().getName(), e); + log.debug(() -> "Failed to publish CRT S3 metrics through " + publisher.getClass().getName(), e); } } } @@ -116,16 +116,16 @@ private void publish(MetricCollection collection, List publishe private void recordQueueFullDrop() { long dropped = queueFullDropCount.incrementAndGet(); if (isPowerOfTwo(dropped)) { - log.warn(() -> "Dropped CRT S3 metric collections because the metric publisher queue is full. " - + "Queue capacity: " + queueCapacity + ", total dropped: " + dropped + "."); + log.debug(() -> "Dropped CRT S3 metric collections because the metric publisher queue is full. " + + "Queue capacity: " + queueCapacity + ", total dropped: " + dropped + "."); } } private void recordClosedDispatcherDrop() { long dropped = closedDispatcherDropCount.incrementAndGet(); if (isPowerOfTwo(dropped)) { - log.warn(() -> "Dropped CRT S3 metric collections because the metric publisher dispatcher is closed. " - + "Total dropped after close: " + dropped + "."); + log.debug(() -> "Dropped CRT S3 metric collections because the metric publisher dispatcher is closed. " + + "Total dropped after close: " + dropped + "."); } } @@ -151,15 +151,15 @@ public void close() { } catch (InterruptedException e) { int queuedCollectionCount = forceShutdown(); Thread.currentThread().interrupt(); - log.warn(() -> "Interrupted while draining CRT S3 metric publisher queue. Undelivered queued collections: " - + queuedCollectionCount + "."); + log.debug(() -> "Interrupted while draining CRT S3 metric publisher queue. Undelivered queued collections: " + + queuedCollectionCount + "."); return; } if (!terminated) { int queuedCollectionCount = forceShutdown(); - log.warn(() -> "Timed out draining CRT S3 metric publisher queue. Undelivered queued collections: " - + queuedCollectionCount + "."); + log.debug(() -> "Timed out draining CRT S3 metric publisher queue. Undelivered queued collections: " + + queuedCollectionCount + "."); } } diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java index 941c7db99cae..403de28e427d 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java @@ -436,7 +436,7 @@ public void onTelemetry(S3RequestMetrics requestMetrics) { MetricCollection collection = apiCall.collect(); metricPublisherDispatcher.dispatch(collection, metricPublishers); } catch (RuntimeException e) { - log.warn(() -> "Failed to collect CRT S3 request metrics", e); + log.debug(() -> "Failed to collect CRT S3 request metrics", e); } } From fc3c9a3a7e086ad8fea641364300d55e4084abf9 Mon Sep 17 00:00:00 2001 From: jencymaryjoseph <35571282+jencymaryjoseph@users.noreply.github.com> Date: Wed, 28 Oct 2026 12:52:04 -0700 Subject: [PATCH 3/4] Address S3 CRT metric dispatcher review feedback --- .../services/s3/S3CrtAsyncClientBuilder.java | 11 +- .../s3/internal/crt/S3CrtAsyncHttpClient.java | 8 +- .../crt/S3CrtMetricPublisherDispatcher.java | 149 +++++++++++++++--- .../crt/S3CrtAsyncHttpClientTest.java | 101 ++++++++++++ .../S3CrtMetricPublisherDispatcherTest.java | 46 ++++++ 5 files changed, 285 insertions(+), 30 deletions(-) diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java index b93849e908d8..33986b25ba7f 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java @@ -390,11 +390,12 @@ default S3CrtAsyncClientBuilder retryConfiguration(ConsumerThe SDK does not close these publishers when the client is closed; the caller retains ownership and is - * responsible for closing them. Metric publication is asynchronous and best effort. Closing the client initiates - * shutdown of metric publication, but may return while previously queued metric collections are being delivered. - * The SDK attempts to drain accepted collections for up to five seconds and then drops any that remain queued. If a - * publisher is closed concurrently with this delivery, those metric collections may not be published and a debug - * message may be logged. This does not affect the outcome of the S3 operation. + * responsible for closing them. Metric publication is asynchronous and best effort. Closing the client waits up to + * five seconds for metric collections already accepted by the dispatcher. Collections that arrive while native CRT + * shutdown is in progress are drained asynchronously for up to five seconds after CRT shutdown completes. Closing + * the client may therefore return before those final collections are delivered. If a publisher is closed + * concurrently with this delivery, those metric collections may not be published and a debug message may be logged. + * This does not affect the outcome of the S3 operation. * * @param metricPublishers the metric publishers to use * @return this builder for method chaining. diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java index e9cb73c950d7..1f4f4db7f2f3 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java @@ -346,9 +346,13 @@ public void close() { CompletableFuture shutdownCompleteFuture = crtS3Client.getShutdownCompleteFuture(); shutdownCompleteFuture.whenComplete((ignored, throwable) -> metricPublisherDispatcher.startAsyncClose()); try { - crtS3Client.close(); + metricPublisherDispatcher.flushPendingMetrics(); } finally { - s3NativeClientConfiguration.close(); + try { + s3NativeClientConfiguration.close(); + } finally { + crtS3Client.close(); + } } } diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java index 4aa93639481c..32322b533c51 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java @@ -18,7 +18,7 @@ import java.time.Duration; import java.util.List; import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -42,22 +42,29 @@ @SdkInternalApi final class S3CrtMetricPublisherDispatcher implements SdkAutoCloseable { private static final Logger log = Logger.loggerFor(S3CrtMetricPublisherDispatcher.class); + // Uses CloudWatchMetricPublisher's 128-entry queue as an existing SDK precedent. This limits each client + // to retaining at most 128 queued metric collections awaiting publication. private static final int DEFAULT_QUEUE_CAPACITY = 128; + // Allows time for queued metrics to be published while preventing a slow publisher from delaying shutdown + // indefinitely. private static final Duration DEFAULT_CLOSE_TIMEOUT = Duration.ofSeconds(5); + // Delay between attempts to enqueue a flush marker when the queue is full; retries stop when + // DEFAULT_CLOSE_TIMEOUT expires. + private static final long FLUSH_RETRY_DELAY_NANOS = TimeUnit.MILLISECONDS.toNanos(10); private static final ThreadFactory CLOSE_THREAD_FACTORY = new ThreadFactoryBuilder() .daemonThreads(true) .threadNamePrefix("s3-crt-metric-publisher-cleanup") .build(); - private final Object lock = new Object(); private final ThreadPoolExecutor executor; private final int queueCapacity; private final Duration closeTimeout; private final AtomicLong queueFullDropCount = new AtomicLong(); private final AtomicLong closedDispatcherDropCount = new AtomicLong(); + private final AtomicBoolean dispatchAttempted = new AtomicBoolean(); private final AtomicBoolean asyncCloseStarted = new AtomicBoolean(); - private boolean closed; + private final AtomicBoolean closed = new AtomicBoolean(); S3CrtMetricPublisherDispatcher() { this(DEFAULT_QUEUE_CAPACITY, DEFAULT_CLOSE_TIMEOUT); @@ -80,26 +87,46 @@ final class S3CrtMetricPublisherDispatcher implements SdkAutoCloseable { new ThreadFactoryBuilder() .daemonThreads(true) .threadNamePrefix("s3-crt-metric-publisher") - .build()); + .build(), + this::handleRejectedTask); } /** - * Dispatches one collection to all configured publishers as a single queue task. Synchronization establishes one - * acceptance order across concurrent CRT callbacks and coordinates submission with close. Publisher code is never - * invoked while holding the lock. + * Dispatches one collection to all configured publishers as a single queue task. Queue admission is non-blocking, + * and publisher code is only invoked by the executor worker. */ void dispatch(MetricCollection collection, List publishers) { - synchronized (lock) { - if (closed) { - recordClosedDispatcherDrop(); - return; - } + dispatchAttempted.set(true); + executor.execute(new PublishTask(collection, publishers)); + } - try { - executor.execute(() -> publish(collection, publishers)); - } catch (RejectedExecutionException e) { - recordQueueFullDrop(); + /** + * Waits for collections accepted before the marker task to be published, up to the configured close timeout. + */ + void flushPendingMetrics() { + if (!dispatchAttempted.get() || executor.isShutdown()) { + return; + } + + long deadlineNanos = System.nanoTime() + closeTimeout.toNanos(); + FlushMarkerTask flushMarker = submitFlushMarker(deadlineNanos); + if (flushMarker == null) { + return; + } + + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { + logFlushTimeout(); + return; + } + + try { + if (!flushMarker.await(remainingNanos)) { + logFlushTimeout(); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.debug(() -> "Interrupted while waiting for queued CRT S3 metrics to be published."); } } @@ -113,6 +140,45 @@ private void publish(MetricCollection collection, List publishe } } + private FlushMarkerTask submitFlushMarker(long deadlineNanos) { + while (!executor.isShutdown()) { + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { + logFlushTimeout(); + return null; + } + + FlushMarkerTask flushMarker = new FlushMarkerTask(); + executor.execute(flushMarker); + if (!flushMarker.rejected()) { + return flushMarker; + } + + try { + TimeUnit.NANOSECONDS.sleep(Math.min(remainingNanos, FLUSH_RETRY_DELAY_NANOS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.debug(() -> "Interrupted while waiting to flush queued CRT S3 metrics."); + return null; + } + } + return null; + } + + private void handleRejectedTask(Runnable task, ThreadPoolExecutor pool) { + if (task instanceof FlushMarkerTask) { + ((FlushMarkerTask) task).reject(); + } else if (pool.isShutdown()) { + recordClosedDispatcherDrop(); + } else { + recordQueueFullDrop(); + } + } + + private void logFlushTimeout() { + log.debug(() -> "Timed out waiting for queued CRT S3 metrics to be published."); + } + private void recordQueueFullDrop() { long dropped = queueFullDropCount.incrementAndGet(); if (isPowerOfTwo(dropped)) { @@ -137,13 +203,10 @@ void startAsyncClose() { @Override public void close() { - synchronized (lock) { - if (closed) { - return; - } - closed = true; - executor.shutdown(); + if (!closed.compareAndSet(false, true)) { + return; } + executor.shutdown(); boolean terminated = false; try { @@ -164,10 +227,50 @@ public void close() { } private int forceShutdown() { - return executor.shutdownNow().size(); + return (int) executor.shutdownNow() + .stream() + .filter(PublishTask.class::isInstance) + .count(); } private static boolean isPowerOfTwo(long value) { return (value & (value - 1)) == 0; } + + private final class PublishTask implements Runnable { + private final MetricCollection collection; + private final List publishers; + + private PublishTask(MetricCollection collection, List publishers) { + this.collection = collection; + this.publishers = publishers; + } + + @Override + public void run() { + publish(collection, publishers); + } + } + + private static final class FlushMarkerTask implements Runnable { + private final CountDownLatch completion = new CountDownLatch(1); + private volatile boolean rejected; + + @Override + public void run() { + completion.countDown(); + } + + private void reject() { + rejected = true; + } + + private boolean rejected() { + return rejected; + } + + private boolean await(long timeoutNanos) throws InterruptedException { + return completion.await(timeoutNanos, TimeUnit.NANOSECONDS); + } + } } diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java index 826f84d9ba51..308db450368d 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java @@ -44,6 +44,10 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Stream; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -465,6 +469,82 @@ public void closeHttpClient_shouldCloseUnderlyingResources() { s3NativeClientConfiguration.close(); } + @Test + public void close_withQueuedMetrics_waitsBeforeClosingCrtClient() throws Exception { + CompletableFuture shutdownCompleteFuture = new CompletableFuture<>(); + when(s3Client.getShutdownCompleteFuture()).thenReturn(shutdownCompleteFuture); + BlockingMetricPublisher publisher = new BlockingMetricPublisher(); + asyncHttpClient = new S3CrtAsyncHttpClient( + s3Client, + S3CrtAsyncHttpClient.builder() + .s3ClientConfiguration(s3NativeClientConfiguration) + .metricPublishers(Collections.singletonList(publisher))); + S3MetaRequestOptions requestOptions = makeRequest(getExecuteRequestBuilder().build()); + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + assertThat(publisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + + CompletableFuture closeFuture = CompletableFuture.runAsync(asyncHttpClient::close); + try { + verify(s3Client, timeout(1_000)).getShutdownCompleteFuture(); + assertThat(closeFuture.isDone()).isFalse(); + verify(s3Client, Mockito.never()).close(); + + publisher.release.countDown(); + closeFuture.get(1, TimeUnit.SECONDS); + assertThat(publisher.publishCount).hasValue(2); + verify(s3Client).close(); + } finally { + publisher.release.countDown(); + shutdownCompleteFuture.complete(null); + } + } + + @Test + public void close_metricDispatcherClosesOnlyAfterCrtShutdownCompletes() throws Exception { + CompletableFuture shutdownCompleteFuture = new CompletableFuture<>(); + when(s3Client.getShutdownCompleteFuture()).thenReturn(shutdownCompleteFuture); + AtomicInteger publishCount = new AtomicInteger(); + AtomicReference publishThread = new AtomicReference<>(); + CountDownLatch published = new CountDownLatch(1); + MetricPublisher publisher = new MetricPublisher() { + @Override + public void publish(MetricCollection metricCollection) { + publishThread.set(Thread.currentThread()); + publishCount.incrementAndGet(); + published.countDown(); + } + + @Override + public void close() { + } + }; + asyncHttpClient = new S3CrtAsyncHttpClient( + s3Client, + S3CrtAsyncHttpClient.builder() + .s3ClientConfiguration(s3NativeClientConfiguration) + .metricPublishers(Collections.singletonList(publisher))); + S3MetaRequestOptions requestOptions = makeRequest(getExecuteRequestBuilder().build()); + + try { + asyncHttpClient.close(); + + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + assertThat(published.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(publishCount).hasValue(1); + assertThat(publishThread.get().isAlive()).isTrue(); + + shutdownCompleteFuture.complete(null); + publishThread.get().join(1_000); + assertThat(publishThread.get().isAlive()).isFalse(); + + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + assertThat(publishCount).hasValue(1); + } finally { + shutdownCompleteFuture.complete(null); + } + } + @Test void build_shouldPassThroughParameters() { String signingRegion = "us-west-2"; @@ -741,6 +821,27 @@ void execute_whenMakeMetaRequestThrows_shouldCloseAdapter() { Mockito.verify(adapter).close(); } + private static final class BlockingMetricPublisher implements MetricPublisher { + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private final AtomicInteger publishCount = new AtomicInteger(); + + @Override + public void publish(MetricCollection metricCollection) { + started.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + publishCount.incrementAndGet(); + } + + @Override + public void close() { + } + } + private AsyncExecuteRequest.Builder getExecuteRequestBuilder() { return getExecuteRequestBuilder(443); } diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java index 221634abbc14..8439b5cc56f5 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java @@ -132,6 +132,38 @@ public void dispatch_fullQueue_dropsWholeCollection() throws Exception { } } + @Test + public void flushPendingMetrics_fullQueue_retriesUntilMarkerIsAccepted() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofSeconds(2)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + MetricPublisher queuedPublisher = mock(MetricPublisher.class); + AtomicReference flushThread = new AtomicReference<>(); + + CompletableFuture flushFuture = null; + try { + dispatcher.dispatch(collection("one"), Collections.singletonList(blockingPublisher)); + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + dispatcher.dispatch(collection("two"), Collections.singletonList(queuedPublisher)); + + flushFuture = CompletableFuture.runAsync(() -> { + flushThread.set(Thread.currentThread()); + dispatcher.flushPendingMetrics(); + }); + + assertThat(waitForThreadState(flushThread, Thread.State.TIMED_WAITING, Duration.ofSeconds(1))).isTrue(); + assertThat(flushFuture.isDone()).isFalse(); + + blockingPublisher.release.countDown(); + + flushFuture.get(1, TimeUnit.SECONDS); + verify(queuedPublisher).publish(collectionNamed("two")); + } finally { + blockingPublisher.release.countDown(); + dispatcher.close(); + } + } + @Test public void dispatch_publisherThrows_doesNotAffectRemainingPublishersAndWorkerContinues() throws Exception { S3CrtMetricPublisherDispatcher dispatcher = @@ -248,6 +280,20 @@ private static MetricCollection collectionNamed(String name) { return org.mockito.ArgumentMatchers.argThat(collection -> collection != null && name.equals(collection.name())); } + private static boolean waitForThreadState(AtomicReference threadReference, + Thread.State state, + Duration timeout) throws InterruptedException { + long deadlineNanos = System.nanoTime() + timeout.toNanos(); + while (System.nanoTime() < deadlineNanos) { + Thread thread = threadReference.get(); + if (thread != null && thread.getState() == state) { + return true; + } + TimeUnit.MILLISECONDS.sleep(1); + } + return false; + } + private static final class BlockingPublisher implements MetricPublisher { private final CountDownLatch started = new CountDownLatch(1); private final CountDownLatch release = new CountDownLatch(1); From ea9a637b4511e5fa6dc1d9bf58a73b31c26bc24e Mon Sep 17 00:00:00 2001 From: jencymaryjoseph <35571282+jencymaryjoseph@users.noreply.github.com> Date: Fri, 30 Oct 2026 10:53:40 -0700 Subject: [PATCH 4/4] Address S3 CRT metric dispatcher review comments --- .../awssdk/services/s3/S3CrtAsyncClientBuilder.java | 10 ++++------ .../internal/crt/S3CrtMetricPublisherDispatcher.java | 4 ++++ 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java index 33986b25ba7f..a594e50f189c 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java @@ -390,12 +390,10 @@ default S3CrtAsyncClientBuilder retryConfiguration(ConsumerThe SDK does not close these publishers when the client is closed; the caller retains ownership and is - * responsible for closing them. Metric publication is asynchronous and best effort. Closing the client waits up to - * five seconds for metric collections already accepted by the dispatcher. Collections that arrive while native CRT - * shutdown is in progress are drained asynchronously for up to five seconds after CRT shutdown completes. Closing - * the client may therefore return before those final collections are delivered. If a publisher is closed - * concurrently with this delivery, those metric collections may not be published and a debug message may be logged. - * This does not affect the outcome of the S3 operation. + * responsible for closing them. Metric publication is asynchronous and best effort. Closing the client waits for a + * bounded time while pending metric collections are published, but may return before collections generated during + * client shutdown are delivered. If a publisher is closed concurrently with this delivery, those metric collections + * may not be published and a debug message may be logged. This does not affect the outcome of the S3 operation. * * @param metricPublishers the metric publishers to use * @return this builder for method chaining. diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java index 32322b533c51..8350ec81b9f2 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java @@ -197,6 +197,10 @@ private void recordClosedDispatcherDrop() { void startAsyncClose() { if (asyncCloseStarted.compareAndSet(false, true)) { + if (!dispatchAttempted.get()) { + close(); + return; + } CLOSE_THREAD_FACTORY.newThread(this::close).start(); } }