refs #91: Add some more safety for BatchingMessageResolver and PrefetchingMessageRetriever

Author: jashmoreAtlassian

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/resolver/batching/BatchingMessageResolver.java

@@ -83,16 +83,20 @@ public CompletableFuture<?> resolveMessage(final Message message) {
     public void run() {
         boolean continueProcessing = true;
         while (continueProcessing) {
-            final List<MessageResolutionBean> batchOfMessagesToResolve = new LinkedList<>();
             try {
-                Queues.drain(messagesToBeResolved, batchOfMessagesToResolve, getBatchSize(), getBufferingTimeInMs(), TimeUnit.MILLISECONDS);
-            } catch (final InterruptedException interruptedException) {
-                // Do nothing, we still want to send the current batch of messages
-                continueProcessing = false;
-            }
-
-            if (!batchOfMessagesToResolve.isEmpty()) {
-                submitMessageDeletionBatch(batchOfMessagesToResolve);
+                final List<MessageResolutionBean> batchOfMessagesToResolve = new LinkedList<>();
+                try {
+                    Queues.drain(messagesToBeResolved, batchOfMessagesToResolve, getBatchSize(), getBufferingTimeInMs(), TimeUnit.MILLISECONDS);
+                } catch (final InterruptedException interruptedException) {
+                    // Do nothing, we still want to send the current batch of messages
+                    continueProcessing = false;
+                }
+
+                if (!batchOfMessagesToResolve.isEmpty()) {
+                    submitMessageDeletionBatch(batchOfMessagesToResolve);
+                }
+            } catch (final Throwable throwable) {
+                log.error("Exception thrown when retrieving messages", throwable);
             }
         }
     }

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/retriever/prefetch/PrefetchingMessageRetriever.java

@@ -8,14 +8,14 @@
 import com.jashmore.sqs.QueueProperties;
 import com.jashmore.sqs.aws.AwsConstants;
 import com.jashmore.sqs.retriever.AsyncMessageRetriever;
+import com.jashmore.sqs.util.properties.PropertyUtils;
 import com.jashmore.sqs.util.retriever.RetrieverUtils;
 import lombok.extern.slf4j.Slf4j;
 import software.amazon.awssdk.services.sqs.SqsAsyncClient;
 import software.amazon.awssdk.services.sqs.model.Message;
 import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
 import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
 
-import java.util.Optional;
 import java.util.concurrent.BlockingDeque;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -181,16 +181,10 @@ private ReceiveMessageRequest buildReceiveMessageRequest() {
     @SuppressWarnings("Duplicates")
     @VisibleForTesting()
     int getBackoffTimeInMs() {
-        return Optional.ofNullable(properties.getErrorBackoffTimeInMilliseconds())
-                .filter(backoffPeriod -> {
-                    if (backoffPeriod < 0) {
-                        log.warn("Non-positive errorBackoffTimeInMilliseconds provided({}), using default({}) instead", backoffPeriod,
-                                DEFAULT_ERROR_BACKOFF_TIMEOUT_IN_MILLISECONDS);
-                        return false;
-                    }
-
-                    return true;
-                })
-                .orElse(DEFAULT_ERROR_BACKOFF_TIMEOUT_IN_MILLISECONDS);
+        return PropertyUtils.safelyGetPositiveIntegerValue(
+                "errorBackoffTimeInMilliseconds",
+                properties::getErrorBackoffTimeInMilliseconds,
+                DEFAULT_ERROR_BACKOFF_TIMEOUT_IN_MILLISECONDS
+        );
     }
 }