Merge branch 'master' into dongie/gh5755

Author: dagnir

.changes/next-release/bugfix-AmazonS3-4025803.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "Fix bug in S3 Multipart uploads with FileAsyncRequestBody - ensure that concurrency is limited correctly by bufferSizeInBytes"
+}

codegen/src/main/java/software/amazon/awssdk/codegen/customization/processors/LowercaseShapeValidatorProcessor.java

@@ -18,6 +18,10 @@
 import software.amazon.awssdk.codegen.customization.CodegenCustomizationProcessor;
 import software.amazon.awssdk.codegen.model.intermediate.IntermediateModel;
 import software.amazon.awssdk.codegen.model.service.ServiceModel;
+import software.amazon.awssdk.codegen.validation.ModelInvalidException;
+import software.amazon.awssdk.codegen.validation.ValidationEntry;
+import software.amazon.awssdk.codegen.validation.ValidationErrorId;
+import software.amazon.awssdk.codegen.validation.ValidationErrorSeverity;
 
 /**
  * A processor that validates shape names in service models to ensure they start with uppercase letters.
@@ -31,11 +35,12 @@ public void preprocess(ServiceModel serviceModel) {
 
         serviceModel.getShapes().forEach((shapeName, shape) -> {
             if ("structure".equals(shape.getType()) && Character.isLowerCase(shapeName.charAt(0))) {
-                throw new IllegalStateException(
-                    String.format("Shape name '%s' starts with a lowercase character." +
-                                  "Shape names must start with an uppercase character." +
-                                  "Please update the shape name in your service model",
-                                  shapeName));
+                String errorMsg = String.format("Shape name '%s' starts with a lowercase character. Shape names must start with"
+                                                + " an uppercase character. Please update the shape name in your service model",
+                                                shapeName);
+                ValidationEntry entry = ValidationEntry.create(ValidationErrorId.INVALID_IDENTIFIER_NAME,
+                                                               ValidationErrorSeverity.DANGER, errorMsg);
+                throw ModelInvalidException.fromEntry(entry);
             }
         });
     }

codegen/src/test/java/software/amazon/awssdk/codegen/customization/processors/LowercaseShapeValidatorProcessorTest.java

@@ -23,6 +23,10 @@
 import org.junit.jupiter.api.Test;
 import software.amazon.awssdk.codegen.model.service.ServiceModel;
 import software.amazon.awssdk.codegen.utils.ModelLoaderUtils;
+import software.amazon.awssdk.codegen.validation.ModelInvalidException;
+import software.amazon.awssdk.codegen.validation.ValidationEntry;
+import software.amazon.awssdk.codegen.validation.ValidationErrorId;
+import software.amazon.awssdk.codegen.validation.ValidationErrorSeverity;
 
 public class LowercaseShapeValidatorProcessorTest {
 
@@ -40,7 +44,14 @@ public static void setUp() throws IOException {
     @Test
     public void preprocess_serviceWithLowercaseShape_throwsException() {
         assertThatThrownBy(() -> processor.preprocess(serviceModel))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessageContaining("Shape name 'lowercaseshape' starts with a lowercase character");
+            .isInstanceOf(ModelInvalidException.class)
+            .hasMessageContaining("Shape name 'lowercaseshape' starts with a lowercase character")
+            .matches(e -> {
+                ModelInvalidException modelInvalid = (ModelInvalidException) e;
+                ValidationEntry entry = modelInvalid.validationEntries().get(0);
+
+                return entry.getErrorId() == ValidationErrorId.INVALID_IDENTIFIER_NAME &&
+                       entry.getSeverity() == ValidationErrorSeverity.DANGER;
+            }, "Validation entry details are correct");
     }
 }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodySplitHelper.java

@@ -166,6 +166,7 @@ private final class FileAsyncRequestBodyWrapper implements AsyncRequestBody {
 
         private final FileAsyncRequestBody fileAsyncRequestBody;
         private final SimplePublisher<AsyncRequestBody> simplePublisher;
+        private final AtomicBoolean hasCompleted = new AtomicBoolean(false);
 
         FileAsyncRequestBodyWrapper(FileAsyncRequestBody fileAsyncRequestBody,
                                     SimplePublisher<AsyncRequestBody> simplePublisher) {
@@ -175,16 +176,22 @@ private final class FileAsyncRequestBodyWrapper implements AsyncRequestBody {
 
         @Override
         public void subscribe(Subscriber<? super ByteBuffer> s) {
-            fileAsyncRequestBody.doAfterOnComplete(() -> startNextRequestBody(simplePublisher))
+            fileAsyncRequestBody.doAfterOnComplete(this::startNextIfNeeded)
                                 // The reason we still need to call startNextRequestBody when the subscription is
                                 // cancelled is that upstream could cancel the subscription even though the stream has
                                 // finished successfully before onComplete. If this happens, doAfterOnComplete callback
                                 // will never be invoked, and if the current buffer is full, the publisher will stop
                                 // sending new FileAsyncRequestBody, leading to uncompleted future.
-                                .doAfterOnCancel(() -> startNextRequestBody(simplePublisher))
+                                .doAfterOnCancel(this::startNextIfNeeded)
                                 .subscribe(s);
         }
 
+        private void startNextIfNeeded() {
+            if (hasCompleted.compareAndSet(false, true)) {
+                startNextRequestBody(simplePublisher);
+            }
+        }
+
         @Override
         public Optional<Long> contentLength() {
             return fileAsyncRequestBody.contentLength();

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodySplitHelperTest.java

@@ -92,7 +92,7 @@ private static Runnable verifyConcurrentRequests(FileAsyncRequestBodySplitHelper
             if (concurrency > maxConcurrency.get()) {
                 maxConcurrency.set(concurrency);
             }
-            assertThat(helper.numAsyncRequestBodiesInFlight()).hasValueLessThan(10);
+            assertThat(helper.numAsyncRequestBodiesInFlight()).hasValueBetween(0,10);
         };
     }
 }