Update doc and update async code path to report writeThroughput for non-streaming operations as well

Author: zoewangg

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/MakeAsyncHttpRequestStage.java

@@ -132,9 +132,10 @@ private CompletableFuture<Response<OutputT>> executeHttpRequest(SdkHttpFullReque
 
         CompletableFuture<Response<OutputT>> responseHandlerFuture = responseHandler.prepare();
 
-        SdkHttpContentPublisher requestProvider = context.requestProvider() == null
+        SdkHttpContentPublisher basePublisher = context.requestProvider() == null
                                                   ? new SimpleHttpContentPublisher(request)
-                                                  : createTrackingContentPublisher(context);
+                                                  : new SdkHttpContentPublisherAdapter(context.requestProvider());
+        SdkHttpContentPublisher requestProvider = wrapWithMetricsTracking(basePublisher, context);
         // Set content length if it hasn't been set already.
         SdkHttpFullRequest requestWithContentLength = getRequestWithContentLength(request, requestProvider);
 
@@ -222,10 +223,11 @@ private boolean isFullDuplex(ExecutionAttributes executionAttributes) {
                executionAttributes.getAttribute(SdkInternalExecutionAttribute.IS_FULL_DUPLEX);
     }
 
-    private SdkHttpContentPublisher createTrackingContentPublisher(RequestExecutionContext context) {
+    private SdkHttpContentPublisher wrapWithMetricsTracking(SdkHttpContentPublisher publisher,
+                                                            RequestExecutionContext context) {
         RequestBodyMetrics metrics = context.executionAttributes()
             .getAttribute(InternalCoreExecutionAttribute.REQUEST_BODY_METRICS);
-        return new SdkHttpContentPublisherAdapter(context.requestProvider(), metrics);
+        return new TrackingHttpContentPublisher(publisher, metrics);
     }
 
     private SdkHttpFullRequest getRequestWithContentLength(SdkHttpFullRequest request, SdkHttpContentPublisher requestProvider) {
@@ -271,20 +273,40 @@ private void completeResponseFuture(CompletableFuture<Response<OutputT>> respons
      * {@link SdkHttpContentPublisher} which the HTTP client SPI expects.
      */
     private static final class SdkHttpContentPublisherAdapter implements SdkHttpContentPublisher {
-
         private final AsyncRequestBody asyncRequestBody;
-        private final Publisher<ByteBuffer> trackingPublisher;
 
-        private SdkHttpContentPublisherAdapter(AsyncRequestBody asyncRequestBody, RequestBodyMetrics metrics) {
+        private SdkHttpContentPublisherAdapter(AsyncRequestBody asyncRequestBody) {
             this.asyncRequestBody = asyncRequestBody;
-            this.trackingPublisher = new BytesWrittenTrackingPublisher(asyncRequestBody, metrics);
         }
 
         @Override
         public Optional<Long> contentLength() {
             return asyncRequestBody.contentLength();
         }
 
+        @Override
+        public void subscribe(Subscriber<? super ByteBuffer> s) {
+            asyncRequestBody.subscribe(s);
+        }
+    }
+
+    /**
+     * Wraps an {@link SdkHttpContentPublisher} with write throughput tracking.
+     */
+    private static final class TrackingHttpContentPublisher implements SdkHttpContentPublisher {
+        private final SdkHttpContentPublisher delegate;
+        private final Publisher<ByteBuffer> trackingPublisher;
+
+        private TrackingHttpContentPublisher(SdkHttpContentPublisher delegate, RequestBodyMetrics metrics) {
+            this.delegate = delegate;
+            this.trackingPublisher = new BytesWrittenTrackingPublisher(delegate, metrics);
+        }
+
+        @Override
+        public Optional<Long> contentLength() {
+            return delegate.contentLength();
+        }
+
         @Override
         public void subscribe(Subscriber<? super ByteBuffer> s) {
             trackingPublisher.subscribe(s);

core/sdk-core/src/main/java/software/amazon/awssdk/core/metrics/CoreMetric.java

@@ -158,14 +158,17 @@ public final class CoreMetric {
 
     /**
      * The write throughput of the client, defined as
-     * {@code RequestBytesWritten / (LastByteWrittenTime - FirstByteWrittenTime)},
-     * where FirstByteWrittenTime is when the first byte is read from the request body and LastByteWrittenTime is when
-     * the last byte is read. This value is in bytes per second.
+     * {@code RequestBytesWritten / (LastByteWrittenTime - FirstByteWrittenTime)}.
+     * This value is in bytes per second.
      * <p>
-     * This metric measures the rate at which bytes are read from the request body stream. It excludes connection setup,
-     * TLS handshake time, and server processing time.
+     * This metric measures the rate at which the SDK provides the request body to the HTTP client.
+     * It excludes connection setup, TLS handshake time, and server processing time.
      * <p>
-     * Note: This metric does not account for buffering in  the HTTP client layer. The actual network transmission rate may
+     * Note: This metric only measures the request body, not HTTP headers. For requests with small
+     * payloads where the body size is comparable to the headers size, this metric may not accurately
+     * reflect overall network throughput.
+     * <p>
+     * Note: This metric does not account for buffering in the HTTP client layer. The actual network transmission rate may
      * be lower if the HTTP client buffers data before sending. This metric represents an upper bound of the network
      * throughput.
      * <p>

test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/metrics/AsyncWriteThroughputMetricTest.java

@@ -102,4 +102,23 @@ public void operationWithNoInputOrOutput_noRequestBody_writeThroughputNotReporte
         List<Double> writeThroughputValues = attemptMetrics.get(0).metricValues(CoreMetric.WRITE_THROUGHPUT);
         assertThat(writeThroughputValues).isEmpty();
     }
+
+    @Test
+    public void nonStreamingOperation_withRequestBody_writeThroughputReported() {
+        stubFor(post(anyUrl())
+                    .willReturn(aResponse().withStatus(200).withBody("{}")));
+
+        client.allTypes(r -> r.stringMember("test")).join();
+
+        ArgumentCaptor<MetricCollection> collectionCaptor = ArgumentCaptor.forClass(MetricCollection.class);
+        verify(mockPublisher).publish(collectionCaptor.capture());
+
+        MetricCollection capturedCollection = collectionCaptor.getValue();
+        List<MetricCollection> attemptMetrics = capturedCollection.children();
+
+        assertThat(attemptMetrics).hasSize(1);
+        List<Double> writeThroughputValues = attemptMetrics.get(0).metricValues(CoreMetric.WRITE_THROUGHPUT);
+        assertThat(writeThroughputValues).hasSize(1);
+        assertThat(writeThroughputValues.get(0)).isGreaterThan(0);
+    }
 }

test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/metrics/SyncWriteThroughputMetricTest.java

@@ -102,4 +102,23 @@ public void operationWithNoInputOrOutput_noRequestBody_writeThroughputNotReporte
         List<Double> writeThroughputValues = attemptMetrics.get(0).metricValues(CoreMetric.WRITE_THROUGHPUT);
         assertThat(writeThroughputValues).isEmpty();
     }
+
+    @Test
+    public void nonStreamingOperation_withRequestBody_writeThroughputReported() {
+        stubFor(post(anyUrl())
+                    .willReturn(aResponse().withStatus(200).withBody("{}")));
+
+        client.allTypes(r -> r.stringMember("test"));
+
+        ArgumentCaptor<MetricCollection> collectionCaptor = ArgumentCaptor.forClass(MetricCollection.class);
+        verify(mockPublisher).publish(collectionCaptor.capture());
+
+        MetricCollection capturedCollection = collectionCaptor.getValue();
+        List<MetricCollection> attemptMetrics = capturedCollection.children();
+
+        assertThat(attemptMetrics).hasSize(1);
+        List<Double> writeThroughputValues = attemptMetrics.get(0).metricValues(CoreMetric.WRITE_THROUGHPUT);
+        assertThat(writeThroughputValues).hasSize(1);
+        assertThat(writeThroughputValues.get(0)).isGreaterThan(0);
+    }
 }