Handled review comments

Author: joviegas

services/sqs/src/main/java/software/amazon/awssdk/services/sqs/batchmanager/BatchOverrideConfiguration.java

@@ -36,30 +36,28 @@ public final class BatchOverrideConfiguration implements ToCopyableBuilder<Batch
     private final Integer maxBatchItems;
     private final Integer maxBatchKeys;
     private final Integer maxBufferSize;
-    private final Duration maxBatchOpenInMs;
+    private final Duration maxBatchOpen;
     private final Duration visibilityTimeout;
-    private final Integer longPollWaitTimeoutSeconds;
-    private final Duration minReceiveWaitTimeMs;
+    private final Duration longPollWaitTimeout;
+    private final Duration minReceiveWaitTime;
     private final Integer maxDoneReceiveBatches;
     private final List<String> receiveAttributeNames;
     private final List<String> receiveMessageAttributeNames;
     private final Boolean adaptivePrefetching;
     private final Integer maxInflightReceiveBatches;
-    private final Boolean longPoll;
 
     public BatchOverrideConfiguration(Builder builder) {
         this.maxBatchItems = Validate.isPositiveOrNull(builder.maxBatchItems, "maxBatchItems");
         this.maxBatchKeys = Validate.isPositiveOrNull(builder.maxBatchKeys, "maxBatchKeys");
         this.maxBufferSize = Validate.isPositiveOrNull(builder.maxBufferSize, "maxBufferSize");
-        this.maxBatchOpenInMs = Validate.isPositiveOrNull(builder.maxBatchOpenInMs, "maxBatchOpenInMs");
-        this.visibilityTimeout = Validate.isPositiveOrNull(builder.visibilityTimeout, "visibilityTimeoutSeconds");
-        this.longPollWaitTimeoutSeconds = Validate.isPositiveOrNull(builder.longPollWaitTimeoutSeconds,
-                                                                    "longPollWaitTimeoutSeconds");
-        this.minReceiveWaitTimeMs = Validate.isPositiveOrNull(builder.minReceiveWaitTimeMs, "minReceiveWaitTimeMs");
+        this.maxBatchOpen = Validate.isPositiveOrNull(builder.maxBatchOpen, "maxBatchOpen");
+        this.visibilityTimeout = Validate.isPositiveOrNull(builder.visibilityTimeout, "visibilityTimeout");
+        this.longPollWaitTimeout = Validate.isPositiveOrNull(builder.longPollWaitTimeout,
+                                                             "longPollWaitTimeout");
+        this.minReceiveWaitTime = Validate.isPositiveOrNull(builder.minReceiveWaitTime, "minReceiveWaitTime");
         this.receiveAttributeNames = builder.receiveAttributeNames;
         this.receiveMessageAttributeNames = builder.receiveMessageAttributeNames;
         this.adaptivePrefetching = builder.adaptivePrefetching;
-        this.longPoll = builder.longPoll;
         this.maxInflightReceiveBatches = builder.maxInflightReceiveBatches;
         this.maxDoneReceiveBatches = builder.maxDoneReceiveBatches;
     }
@@ -97,8 +95,8 @@ public Optional<Integer> maxDoneReceiveBatches() {
      * @return the optional maximum amount of time (in milliseconds) that an outgoing call waits to be batched with messages of
      * the same type.
      */
-    public Optional<Duration> maxBatchOpenInMs() {
-        return Optional.ofNullable(maxBatchOpenInMs);
+    public Optional<Duration> maxBatchOpen() {
+        return Optional.ofNullable(maxBatchOpen);
     }
 
     /**
@@ -112,15 +110,15 @@ public Optional<Duration> visibilityTimeout() {
      * @return the amount of time, in seconds, the receive call will block on the server waiting for messages to arrive if the
      * queue is empty when the receive call is first made.
      */
-    public Optional<Integer> longPollWaitTimeoutSeconds() {
-        return Optional.ofNullable(longPollWaitTimeoutSeconds);
+    public Optional<Duration> longPollWaitTimeout() {
+        return Optional.ofNullable(longPollWaitTimeout);
     }
 
     /**
      * @return the minimum wait time for incoming receive message requests.
      */
-    public Optional<Duration> minReceiveWaitTimeMs() {
-        return Optional.ofNullable(minReceiveWaitTimeMs);
+    public Optional<Duration> minReceiveWaitTime() {
+        return Optional.ofNullable(minReceiveWaitTime);
     }
 
     /**
@@ -144,13 +142,6 @@ public Optional<Boolean> adaptivePrefetching() {
         return Optional.ofNullable(adaptivePrefetching);
     }
 
-    /**
-     * @return the option for long polling.
-     */
-    public Optional<Boolean> longPoll() {
-        return Optional.ofNullable(longPoll);
-    }
-
     /**
      * @return the maximum number of concurrent receive message batches.
      */
@@ -163,16 +154,15 @@ public Builder toBuilder() {
         return new Builder().maxBatchItems(maxBatchItems)
                             .maxBatchKeys(maxBatchKeys)
                             .maxBufferSize(maxBufferSize)
-                            .maxBatchOpenInMs(maxBatchOpenInMs)
+                            .maxBatchOpen(maxBatchOpen)
                             .visibilityTimeout(visibilityTimeout)
-                            .longPollWaitTimeoutSeconds(longPollWaitTimeoutSeconds)
-                            .minReceiveWaitTimeMs(minReceiveWaitTimeMs)
+                            .longPollWaitTimeout(longPollWaitTimeout)
+                            .minReceiveWaitTime(minReceiveWaitTime)
                             .maxInflightReceiveBatches(maxInflightReceiveBatches)
                             .receiveAttributeNames(receiveAttributeNames)
                             .receiveMessageAttributeNames(receiveMessageAttributeNames)
                             .adaptivePrefetching(adaptivePrefetching)
-                            .maxDoneReceiveBatches(maxDoneReceiveBatches)
-                            .longPoll(longPoll);
+                            .maxDoneReceiveBatches(maxDoneReceiveBatches);
     }
 
     @Override
@@ -181,16 +171,15 @@ public String toString() {
                        .add("maxBatchItems", maxBatchItems)
                        .add("maxBatchKeys", maxBatchKeys)
                        .add("maxBufferSize", maxBufferSize)
-                       .add("maxBatchOpenInMs", maxBatchOpenInMs.toMillis())
-                       .add("visibilityTimeoutSeconds", visibilityTimeout)
-                       .add("longPollWaitTimeoutSeconds", longPollWaitTimeoutSeconds)
-                       .add("minReceiveWaitTimeMs", minReceiveWaitTimeMs)
+                       .add("maxBatchOpen", maxBatchOpen.toMillis())
+                       .add("visibilityTimeout", visibilityTimeout)
+                       .add("longPollWaitTimeout", longPollWaitTimeout)
+                       .add("minReceiveWaitTime", minReceiveWaitTime)
                        .add("receiveAttributeNames", receiveAttributeNames)
                        .add("receiveMessageAttributeNames", receiveMessageAttributeNames)
                        .add("adaptivePrefetching", adaptivePrefetching)
                        .add("maxInflightReceiveBatches", maxInflightReceiveBatches)
                        .add("maxDoneReceiveBatches", maxDoneReceiveBatches)
-                       .add("longPoll", longPoll)
                        .build();
     }
 
@@ -215,19 +204,19 @@ public boolean equals(Object o) {
             return false;
         }
 
-        if (maxBatchOpenInMs != null ? !maxBatchOpenInMs.equals(that.maxBatchOpenInMs) : that.maxBatchOpenInMs != null) {
+        if (maxBatchOpen != null ? !maxBatchOpen.equals(that.maxBatchOpen) : that.maxBatchOpen != null) {
             return false;
         }
         if (visibilityTimeout != null ? !visibilityTimeout.equals(that.visibilityTimeout) :
             that.visibilityTimeout != null) {
             return false;
         }
-        if (longPollWaitTimeoutSeconds != null ? !longPollWaitTimeoutSeconds.equals(that.longPollWaitTimeoutSeconds) :
-            that.longPollWaitTimeoutSeconds != null) {
+        if (longPollWaitTimeout != null ? !longPollWaitTimeout.equals(that.longPollWaitTimeout) :
+            that.longPollWaitTimeout != null) {
             return false;
         }
-        if (minReceiveWaitTimeMs != null ? !minReceiveWaitTimeMs.equals(that.minReceiveWaitTimeMs) :
-            that.minReceiveWaitTimeMs != null) {
+        if (minReceiveWaitTime != null ? !minReceiveWaitTime.equals(that.minReceiveWaitTime) :
+            that.minReceiveWaitTime != null) {
             return false;
         }
         if (receiveAttributeNames != null ? !receiveAttributeNames.equals(that.receiveAttributeNames) :
@@ -242,9 +231,6 @@ public boolean equals(Object o) {
             that.adaptivePrefetching != null) {
             return false;
         }
-        if (longPoll != null ? !longPoll.equals(that.longPoll) : that.longPoll != null) {
-            return false;
-        }
         if (maxInflightReceiveBatches != null ? !maxInflightReceiveBatches.equals(that.maxInflightReceiveBatches) :
             that.maxInflightReceiveBatches != null) {
             return false;
@@ -258,14 +244,13 @@ public int hashCode() {
         int result = maxBatchItems != null ? maxBatchItems.hashCode() : 0;
         result = 31 * result + (maxBatchKeys != null ? maxBatchKeys.hashCode() : 0);
         result = 31 * result + (maxBufferSize != null ? maxBufferSize.hashCode() : 0);
-        result = 31 * result + (maxBatchOpenInMs != null ? maxBatchOpenInMs.hashCode() : 0);
+        result = 31 * result + (maxBatchOpen != null ? maxBatchOpen.hashCode() : 0);
         result = 31 * result + (visibilityTimeout != null ? visibilityTimeout.hashCode() : 0);
-        result = 31 * result + (longPollWaitTimeoutSeconds != null ? longPollWaitTimeoutSeconds.hashCode() : 0);
-        result = 31 * result + (minReceiveWaitTimeMs != null ? minReceiveWaitTimeMs.hashCode() : 0);
+        result = 31 * result + (longPollWaitTimeout != null ? longPollWaitTimeout.hashCode() : 0);
+        result = 31 * result + (minReceiveWaitTime != null ? minReceiveWaitTime.hashCode() : 0);
         result = 31 * result + (receiveAttributeNames != null ? receiveAttributeNames.hashCode() : 0);
         result = 31 * result + (receiveMessageAttributeNames != null ? receiveMessageAttributeNames.hashCode() : 0);
         result = 31 * result + (adaptivePrefetching != null ? adaptivePrefetching.hashCode() : 0);
-        result = 31 * result + (longPoll != null ? longPoll.hashCode() : 0);
         result = 31 * result + (maxInflightReceiveBatches != null ? maxInflightReceiveBatches.hashCode() : 0);
         result = 31 * result + (maxDoneReceiveBatches != null ? maxDoneReceiveBatches.hashCode() : 0);
         return result;
@@ -276,16 +261,15 @@ public static final class Builder implements CopyableBuilder<Builder, BatchOverr
         private Integer maxBatchItems;
         private Integer maxBatchKeys;
         private Integer maxBufferSize;
-        private Duration maxBatchOpenInMs;
+        private Duration maxBatchOpen;
         private Duration visibilityTimeout;
-        private Integer longPollWaitTimeoutSeconds;
-        private Duration minReceiveWaitTimeMs;
+        private Duration longPollWaitTimeout;
+        private Duration minReceiveWaitTime;
         private Integer maxDoneReceiveBatches;
         private Integer maxInflightReceiveBatches;
         private List<String> receiveAttributeNames = Collections.emptyList();
         private List<String> receiveMessageAttributeNames = Collections.emptyList();
         private Boolean adaptivePrefetching;
-        private Boolean longPoll;
 
         private Builder() {
         }
@@ -331,11 +315,11 @@ public Builder maxBufferSize(Integer maxBufferSize) {
          * Define the maximum amount of time (in milliseconds) that an outgoing call waits for other requests before sending out a
          * batch request.
          *
-         * @param maxBatchOpenInMs The new maxBatchOpenInMs value.
+         * @param maxBatchOpen The new maxBatchOpen value.
          * @return This object for method chaining.
          */
-        public Builder maxBatchOpenInMs(Duration maxBatchOpenInMs) {
-            this.maxBatchOpenInMs = maxBatchOpenInMs;
+        public Builder maxBatchOpen(Duration maxBatchOpen) {
+            this.maxBatchOpen = maxBatchOpen;
             return this;
         }
 
@@ -356,11 +340,11 @@ public Builder visibilityTimeout(Duration visibilityTimeout) {
          * Define the amount of time, in seconds, the receive call will block on the server waiting for messages to arrive if the
          * queue is empty when the receive call is first made. This setting has no effect if long polling is disabled.
          *
-         * @param longPollWaitTimeoutSeconds The new longPollWaitTimeoutSeconds value.
+         * @param longPollWaitTimeout The new longPollWaitTimeout value.
          * @return This object for method chaining.
          */
-        public Builder longPollWaitTimeoutSeconds(Integer longPollWaitTimeoutSeconds) {
-            this.longPollWaitTimeoutSeconds = longPollWaitTimeoutSeconds;
+        public Builder longPollWaitTimeout(Duration longPollWaitTimeout) {
+            this.longPollWaitTimeout = longPollWaitTimeout;
             return this;
         }
 
@@ -369,11 +353,11 @@ public Builder longPollWaitTimeoutSeconds(Integer longPollWaitTimeoutSeconds) {
          * easily waste CPU time busy-waiting against empty local buffers. Avoid setting this to 0 unless you are confident
          * threads will do useful work in-between each call to receive messages!
          *
-         * @param minReceiveWaitTimeMs The new minReceiveWaitTimeMs value.
+         * @param minReceiveWaitTime The new minReceiveWaitTime value.
          * @return This object for method chaining.
          */
-        public Builder minReceiveWaitTimeMs(Duration minReceiveWaitTimeMs) {
-            this.minReceiveWaitTimeMs = minReceiveWaitTimeMs;
+        public Builder minReceiveWaitTime(Duration minReceiveWaitTime) {
+            this.minReceiveWaitTime = minReceiveWaitTime;
             return this;
         }
 
@@ -445,17 +429,6 @@ public Builder adaptivePrefetching(Boolean adaptivePrefetching) {
             return this;
         }
 
-        /**
-         * Define the option for long polling. Specify "true" for receive requests to use long polling.
-         *
-         * @param longPoll The new longPoll value.
-         * @return This object for method chaining.
-         */
-        public Builder longPoll(Boolean longPoll) {
-            this.longPoll = longPoll;
-            return this;
-        }
-
         public BatchOverrideConfiguration build() {
             return new BatchOverrideConfiguration(this);
         }

services/sqs/src/main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/AsyncReceiveMessageBatch.java

@@ -17,13 +17,10 @@
 
 
 import java.time.Duration;
-import java.time.temporal.ChronoUnit;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import software.amazon.awssdk.annotations.SdkInternalApi;
@@ -34,6 +31,7 @@
 import software.amazon.awssdk.services.sqs.model.Message;
 import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
 import software.amazon.awssdk.utils.Logger;
+import software.amazon.awssdk.utils.NumericUtils;
 
 /**
  * The {@code AsyncReceiveMessageBatch} class forms a {@link  ReceiveMessageRequest} request based on configuration settings,
@@ -49,20 +47,18 @@
 public class AsyncReceiveMessageBatch {
 
     private static final Logger log = Logger.loggerFor(AsyncReceiveMessageBatch.class);
-
-    private final ScheduledExecutorService scheduledExecutorService;
     private final String queueUrl;
     private final SqsAsyncClient asyncClient;
     private final Duration visibilityTimeout;
     private final ResponseBatchConfiguration config;
-    private final AtomicBoolean open = new AtomicBoolean(false);
     private volatile Throwable exception;
-    private List<Message> messages;
+    private volatile List<Message> messages = new CopyOnWriteArrayList<>();
     private long visibilityDeadlineNano;
 
-    public AsyncReceiveMessageBatch(ScheduledExecutorService scheduledExecutorService, String queueUrl,
-                                    SqsAsyncClient asyncClient, Duration visibilityTimeout, ResponseBatchConfiguration config) {
-        this.scheduledExecutorService = scheduledExecutorService;
+    public AsyncReceiveMessageBatch(String queueUrl,
+                                    SqsAsyncClient asyncClient,
+                                    Duration visibilityTimeout,
+                                    ResponseBatchConfiguration config) {
         this.queueUrl = queueUrl;
         this.asyncClient = asyncClient;
         this.visibilityTimeout = visibilityTimeout;
@@ -77,20 +73,19 @@ public CompletableFuture<AsyncReceiveMessageBatch> asyncReceiveMessage() {
                                  .messageAttributeNames(config.receiveMessageAttributeNames())
                                  .messageAttributeNames(config.receiveMessageAttributeNames());
 
-        request.visibilityTimeout((int) this.visibilityTimeout.get(ChronoUnit.SECONDS));
+        request.visibilityTimeout(NumericUtils.saturatedCast(this.visibilityTimeout.getSeconds()));
 
-        if (config.longPoll()) {
-            request.waitTimeSeconds(config.longPollWaitTimeoutSeconds());
+        if (config.longPollWaitTimeout() != null) {
+            request.waitTimeSeconds(NumericUtils.saturatedCast(config.longPollWaitTimeout().getSeconds()));
         }
         try {
             return asyncClient.receiveMessage(request.build())
                               .handle((response, throwable) -> {
                                   if (throwable != null) {
                                       setException(throwable);
                                   } else {
-                                      messages = new ArrayList<>(response.messages());
+                                      messages = new CopyOnWriteArrayList<>(response.messages());
                                   }
-                                  open.set(true);
                                   return this;
                               });
         } finally {
@@ -104,7 +99,6 @@ public boolean isEmpty() {
     }
 
     public Throwable getException() {
-        checkIfOpen();
         return exception;
     }
 
@@ -113,7 +107,6 @@ public void setException(Throwable exception) {
     }
 
     public Message removeMessage() {
-        checkIfOpen();
         if (isExpired()) {
             clear();
             return null;
@@ -161,12 +154,6 @@ private CompletableFuture<ChangeMessageVisibilityBatchResponse> nackMessages() {
         return asyncClient.changeMessageVisibilityBatch(batchRequest);
     }
 
-    private void checkIfOpen() {
-        if (!open.get()) {
-            throw new IllegalStateException("Batch is not open");
-        }
-    }
-
     public Integer messagesSize() {
         return messages != null ? messages.size() : 0;
     }