Handle 200 OK errors in PutObject (multipart) in S3 CRT client (#7084)

Author: alextwoods

.changes/next-release/bugfix-AmazonS3-4405f32.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "Handle 200 OK errors in PutObject (multipart) in S3 CRT client"
+}

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java

@@ -21,6 +21,9 @@
 
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -37,6 +40,8 @@
 import software.amazon.awssdk.crt.s3.S3FinishedResponseContext;
 import software.amazon.awssdk.crt.s3.S3MetaRequestProgress;
 import software.amazon.awssdk.crt.s3.S3MetaRequestResponseHandler;
+import software.amazon.awssdk.http.AbortableInputStream;
+import software.amazon.awssdk.http.SdkHttpFullResponse;
 import software.amazon.awssdk.http.SdkHttpResponse;
 import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler;
 import software.amazon.awssdk.services.s3.model.S3Exception;
@@ -220,6 +225,15 @@ private void handleServiceError(int responseStatus, HttpHeader[] headers, byte[]
             s3Exception.addSuppressed(sdkClientException);
             failResponseHandlerAndFuture(s3Exception);
             notifyResponsePublisherErrorIfNeeded(s3Exception);
+        } else if (responseStatus == 200) {
+            // handleServiceError is only called when crtCode != SUCCESS. If the response status is 200, this is the
+            // S3 "200 with error in body" case (e.g. CompleteMultipartUpload returning HTTP 200 with <Error> XML).
+            // We wrap the response in ErrorFlaggedSdkHttpResponse which overrides isSuccessful() to return false.
+            // This causes the downstream SDK pipeline (XmlResponseParserUtils, DecorateErrorFromResponseBodyUnmarshaller)
+            // to parse the body and detect the error, producing a properly modeled S3Exception.
+            SdkHttpFullResponse errorFlagged = new ErrorFlaggedSdkHttpResponse(errorResponse.build());
+            initiateResponseHandling(errorFlagged);
+            onErrorResponseComplete(errorPayload);
         } else {
             initiateResponseHandling(errorResponse.build());
             onErrorResponseComplete(errorPayload);
@@ -303,4 +317,148 @@ private static SdkHttpResponse.Builder populateSdkHttpResponse(SdkHttpResponse.B
 
     private static class NoOpPublisherListener implements PublisherListener<S3MetaRequestProgress> {
     }
+
+    /**
+     * An {@link SdkHttpFullResponse} wrapper that overrides {@link #isSuccessful()} to always return {@code false}.
+     * This forces the SDK response pipeline ({@code XmlResponseParserUtils}, {@code DecorateErrorFromResponseBodyUnmarshaller})
+     * to parse the response body even for HTTP 200 responses, enabling detection of S3's "200 with error in body" pattern.
+     *
+     * <p>The {@link #toBuilder()} method returns a builder whose {@code build()} also produces an
+     * {@code ErrorFlaggedSdkHttpResponse}, ensuring the override survives the rebuild cycle in
+     * {@code AsyncResponseHandler}.</p>
+     */
+    static final class ErrorFlaggedSdkHttpResponse implements SdkHttpFullResponse {
+        private final SdkHttpFullResponse delegate;
+
+        ErrorFlaggedSdkHttpResponse(SdkHttpResponse delegate) {
+            // Ensure we have a full response to delegate to
+            if (delegate instanceof SdkHttpFullResponse) {
+                this.delegate = (SdkHttpFullResponse) delegate;
+            } else {
+                this.delegate = SdkHttpFullResponse.builder()
+                                                   .statusCode(delegate.statusCode())
+                                                   .headers(delegate.headers())
+                                                   .build();
+            }
+        }
+
+        private ErrorFlaggedSdkHttpResponse(SdkHttpFullResponse delegate, @SuppressWarnings("unused") boolean internal) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public boolean isSuccessful() {
+            return false;
+        }
+
+        @Override
+        public int statusCode() {
+            return delegate.statusCode();
+        }
+
+        @Override
+        public Optional<String> statusText() {
+            return delegate.statusText();
+        }
+
+        @Override
+        public Map<String, List<String>> headers() {
+            return delegate.headers();
+        }
+
+        @Override
+        public Optional<AbortableInputStream> content() {
+            return delegate.content();
+        }
+
+        @Override
+        public Builder toBuilder() {
+            return new ErrorFlaggedBuilder(delegate.toBuilder());
+        }
+    }
+
+    /**
+     * A builder that wraps a standard {@link SdkHttpFullResponse.Builder} but produces an
+     * {@link ErrorFlaggedSdkHttpResponse} on {@code build()}, preserving the {@code isSuccessful()=false} override.
+     */
+    private static final class ErrorFlaggedBuilder implements SdkHttpFullResponse.Builder {
+        private final SdkHttpFullResponse.Builder delegate;
+
+        ErrorFlaggedBuilder(SdkHttpFullResponse.Builder delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public SdkHttpFullResponse build() {
+            return new ErrorFlaggedSdkHttpResponse(delegate.build(), true);
+        }
+
+        @Override
+        public String statusText() {
+            return delegate.statusText();
+        }
+
+        @Override
+        public SdkHttpFullResponse.Builder statusText(String statusText) {
+            delegate.statusText(statusText);
+            return this;
+        }
+
+        @Override
+        public int statusCode() {
+            return delegate.statusCode();
+        }
+
+        @Override
+        public SdkHttpFullResponse.Builder statusCode(int statusCode) {
+            delegate.statusCode(statusCode);
+            return this;
+        }
+
+        @Override
+        public Map<String, List<String>> headers() {
+            return delegate.headers();
+        }
+
+        @Override
+        public SdkHttpFullResponse.Builder putHeader(String headerName, List<String> headerValues) {
+            delegate.putHeader(headerName, headerValues);
+            return this;
+        }
+
+        @Override
+        public SdkHttpFullResponse.Builder appendHeader(String headerName, String headerValue) {
+            delegate.appendHeader(headerName, headerValue);
+            return this;
+        }
+
+        @Override
+        public SdkHttpFullResponse.Builder headers(Map<String, List<String>> headers) {
+            delegate.headers(headers);
+            return this;
+        }
+
+        @Override
+        public SdkHttpFullResponse.Builder removeHeader(String headerName) {
+            delegate.removeHeader(headerName);
+            return this;
+        }
+
+        @Override
+        public SdkHttpFullResponse.Builder clearHeaders() {
+            delegate.clearHeaders();
+            return this;
+        }
+
+        @Override
+        public AbortableInputStream content() {
+            return delegate.content();
+        }
+
+        @Override
+        public SdkHttpFullResponse.Builder content(AbortableInputStream content) {
+            delegate.content(content);
+            return this;
+        }
+    }
 }