Address comments

Author: zoewangg

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

@@ -31,6 +31,7 @@
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.internal.util.Mimetype;
 import software.amazon.awssdk.core.internal.util.NoopSubscription;
+import software.amazon.awssdk.utils.Logger;
 import software.amazon.awssdk.utils.builder.SdkBuilder;
 
 /**
@@ -41,6 +42,7 @@
  */
 @SdkInternalApi
 public final class FileAsyncRequestBody implements AsyncRequestBody {
+    private static final Logger log = Logger.loggerFor(FileAsyncRequestBody.class);
 
     /**
      * Default size (in bytes) of ByteBuffer chunks read from the file and delivered to the subscriber.
@@ -241,7 +243,7 @@ public void completed(Integer result, ByteBuffer attachment) {
                     if (result > 0) {
                         attachment.flip();
                         position.addAndGet(attachment.remaining());
-                        subscriber.onNext(attachment);
+                        signalOnNext(attachment);
 
                         synchronized (lock) {
                             // If we have more permits, queue up another read.
@@ -270,21 +272,31 @@ private void closeFile() {
             try {
                 inputChannel.close();
             } catch (IOException e) {
-                signalOnError(e);
+                log.warn(() -> "Failed to close the file", e);
             }
         }
 
-        private void signalOnComplete() {
+        private void signalOnNext(ByteBuffer attachment) {
             if (!done) {
-                done = true;
-                subscriber.onComplete();
+                subscriber.onNext(attachment);
+            }
+        }
+
+        private void signalOnComplete() {
+            synchronized (this) {
+                if (!done) {
+                    done = true;
+                    subscriber.onComplete();
+                }
             }
         }
 
         private void signalOnError(Throwable t) {
-            if (!done) {
-                done = true;
-                subscriber.onError(t);
+            synchronized (this) {
+                if (!done) {
+                    done = true;
+                    subscriber.onError(t);
+                }
             }
         }
     }