Merge pull request #94 from JaidenAshmore/issue/91_batching_message_retriever_error_safety

JaidenAshmore · web-flow · commit 442eb34b4c57 · 2019-05-22T22:09:25.000-05:00
refs #91: Make BatchingMessageRetriever safer by catching exceptions
diff --git a/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/resolver/batching/BatchingMessageResolver.java b/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);
             }
         }
     }
diff --git a/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/retriever/batching/BatchingMessageRetriever.java b/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/retriever/batching/BatchingMessageRetriever.java
@@ -1,22 +1,24 @@
 package com.jashmore.sqs.retriever.batching;
 
 import static com.jashmore.sqs.retriever.batching.BatchingMessageRetrieverConstants.DEFAULT_BACKOFF_TIME_IN_MS;
+import static com.jashmore.sqs.retriever.batching.BatchingMessageRetrieverConstants.DEFAULT_BATCHING_TRIGGER;
 
 import com.google.common.annotations.VisibleForTesting;
 
 import com.jashmore.sqs.QueueProperties;
 import com.jashmore.sqs.aws.AwsConstants;
+import com.jashmore.sqs.broker.concurrent.ConcurrentMessageBrokerProperties;
 import com.jashmore.sqs.retriever.AsyncMessageRetriever;
 import com.jashmore.sqs.retriever.MessageRetriever;
-import com.jashmore.sqs.util.RetrieverUtils;
+import com.jashmore.sqs.util.properties.PropertyUtils;
+import com.jashmore.sqs.util.retriever.RetrieverUtils;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 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.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -52,11 +54,11 @@ public BatchingMessageRetriever(final QueueProperties queueProperties,
 
     @VisibleForTesting
     BatchingMessageRetriever(final QueueProperties queueProperties,
-                                    final SqsAsyncClient sqsAsyncClient,
-                                    final BatchingMessageRetrieverProperties properties,
-                                    final AtomicInteger numberWaitingForMessages,
-                                    final BlockingQueue<Message> messagesDownloaded,
-                                    final Object shouldObtainMessagesLock) {
+                             final SqsAsyncClient sqsAsyncClient,
+                             final BatchingMessageRetrieverProperties properties,
+                             final AtomicInteger numberWaitingForMessages,
+                             final BlockingQueue<Message> messagesDownloaded,
+                             final Object shouldObtainMessagesLock) {
         this.queueProperties = queueProperties;
         this.sqsAsyncClient = sqsAsyncClient;
         this.properties = properties;
@@ -84,9 +86,10 @@ public Message retrieveMessage() throws InterruptedException {
     private void incrementWaitingCountAndNotify() {
         synchronized (shouldObtainMessagesLock) {
             final int currentThreads = numberWaitingForMessages.incrementAndGet();
-            if (currentThreads >= properties.getNumberOfThreadsWaitingTrigger()) {
+            final int numberOfThreadsWaitingTrigger = getNumberOfThreadsWaitingTrigger();
+            if (currentThreads >= numberOfThreadsWaitingTrigger) {
                 log.debug("Maximum number of threads({}) waiting has arrived requesting any sleeping threads to wake up to process",
-                        properties.getNumberOfThreadsWaitingTrigger());
+                        numberOfThreadsWaitingTrigger);
                 // notify that we should grab a message
                 shouldObtainMessagesLock.notifyAll();
             }
@@ -97,30 +100,30 @@ private void incrementWaitingCountAndNotify() {
     public void run() {
         log.debug("Started background thread");
         while (true) {
-            final int numberOfMessagesToObtain;
-            synchronized (shouldObtainMessagesLock) {
-                final int triggerValue = properties.getNumberOfThreadsWaitingTrigger();
-                if ((numberWaitingForMessages.get() - messagesDownloaded.size()) < triggerValue) {
-                    try {
-                        waitForEnoughThreadsToRequestMessages(getPollingPeriodInMs(triggerValue));
-                    } catch (InterruptedException exception) {
-                        log.debug("Thread interrupted while waiting for messages");
-                        break;
+            try {
+                final int numberOfMessagesToObtain;
+                synchronized (shouldObtainMessagesLock) {
+                    final int triggerValue = getNumberOfThreadsWaitingTrigger();
+                    if ((numberWaitingForMessages.get() - messagesDownloaded.size()) < triggerValue) {
+                        try {
+                            waitForEnoughThreadsToRequestMessages(getPollingPeriodInMs());
+                        } catch (InterruptedException exception) {
+                            log.debug("Thread interrupted while waiting for messages");
+                            break;
+                        }
                     }
+                    numberOfMessagesToObtain = Math.min(numberWaitingForMessages.get() - messagesDownloaded.size(),
+                            AwsConstants.MAX_NUMBER_OF_MESSAGES_FROM_SQS);
                 }
-                numberOfMessagesToObtain = Math.min(numberWaitingForMessages.get() - messagesDownloaded.size(),
-                        AwsConstants.MAX_NUMBER_OF_MESSAGES_FROM_SQS);
-            }
 
-            if (numberOfMessagesToObtain <= 0) {
-                log.debug("Requesting 0 messages");
-                // We don't want to go request out if there are no messages to retrieve
-                continue;
-            }
+                if (numberOfMessagesToObtain <= 0) {
+                    log.debug("Requesting 0 messages");
+                    // We don't want to go request out if there are no messages to retrieve
+                    continue;
+                }
 
-            log.debug("Requesting {} messages", numberOfMessagesToObtain);
+                log.debug("Requesting {} messages", numberOfMessagesToObtain);
 
-            try {
                 final ReceiveMessageResponse response;
                 try {
                     response = sqsAsyncClient.receiveMessage(buildReceiveMessageRequest(numberOfMessagesToObtain))
@@ -140,9 +143,11 @@ public void run() {
             } catch (final Throwable throwable) {
                 log.error("Error thrown trying to obtain messages", throwable);
                 try {
-                    backoff(RetrieverUtils.safelyGetBackoffTime(properties.getErrorBackoffTimeInMilliseconds(), DEFAULT_BACKOFF_TIME_IN_MS));
+                    final long errorBackoffTimeInMilliseconds = getErrorBackoffTimeInMilliseconds();
+                    log.error("Error thrown while organising threads to process messages. Backing off for {}ms", errorBackoffTimeInMilliseconds, throwable);
+                    backoff(errorBackoffTimeInMilliseconds);
                 } catch (final InterruptedException interruptedException) {
-                    log.trace("Thread interrupted during error backoff thread sleeping");
+                    log.debug("Thread interrupted during backoff period");
                     break;
                 }
             }
@@ -155,6 +160,33 @@ void waitForEnoughThreadsToRequestMessages(final long waitPeriodInMs) throws Int
         shouldObtainMessagesLock.wait(waitPeriodInMs);
     }
 
+    /**
+     * Get the number of seconds that the thread should wait when there was an error trying to organise a thread to process.
+     *
+     * @return the backoff time in milliseconds
+     * @see ConcurrentMessageBrokerProperties#getErrorBackoffTimeInMilliseconds() for more information
+     */
+    private long getErrorBackoffTimeInMilliseconds() {
+        return PropertyUtils.safelyGetPositiveOrZeroLongValue(
+                "errorBackoffTimeInMilliseconds",
+                properties::getErrorBackoffTimeInMilliseconds,
+                DEFAULT_BACKOFF_TIME_IN_MS
+        );
+    }
+
+    /**
+     * Safely get the total number of threads requiring messages before it sends a batch request for messages.
+     *
+     * @return the total number of threads for the batching trigger
+     */
+    private int getNumberOfThreadsWaitingTrigger() {
+        return PropertyUtils.safelyGetIntegerValue(
+                "numberOfThreadsWaitingTrigger",
+                properties::getNumberOfThreadsWaitingTrigger,
+                DEFAULT_BATCHING_TRIGGER
+        );
+    }
+
     @VisibleForTesting
     void backoff(final long backoffTimeInMs) throws InterruptedException {
         Thread.sleep(backoffTimeInMs);
@@ -171,12 +203,12 @@ private ReceiveMessageRequest buildReceiveMessageRequest(final int numberOfMessa
                 .builder()
                 .queueUrl(queueProperties.getQueueUrl())
                 .maxNumberOfMessages(numberOfMessagesToObtain)
-                .waitTimeSeconds(RetrieverUtils.safelyGetWaitTimeInSeconds(properties.getMessageWaitTimeInSeconds()));
+                .waitTimeSeconds(RetrieverUtils.safelyGetWaitTimeInSeconds(properties::getMessageWaitTimeInSeconds));
 
         final Integer visibilityTimeoutInSeconds = properties.getVisibilityTimeoutInSeconds();
         if (visibilityTimeoutInSeconds != null) {
             if (visibilityTimeoutInSeconds < 0) {
-                log.warn("Non-positive visibilityTimeoutInSeconds provided: ", visibilityTimeoutInSeconds);
+                log.warn("Non-positive visibilityTimeoutInSeconds provided: {}", visibilityTimeoutInSeconds);
             } else {
                 requestBuilder.visibilityTimeout(visibilityTimeoutInSeconds);
             }
@@ -190,15 +222,13 @@ private ReceiveMessageRequest buildReceiveMessageRequest(final int numberOfMessa
      * could cause this retriever to block forever if the number of threads never reaches
      * {@link BatchingMessageRetrieverProperties#getNumberOfThreadsWaitingTrigger()}.
      *
-     * @param triggerValue the number of threads that would trigger the batch retrieval of messages
      * @return the polling period in ms
      */
-    private long getPollingPeriodInMs(final int triggerValue) {
-        return Optional.ofNullable(properties.getMessageRetrievalPollingPeriodInMs())
-                .orElseGet(() -> {
-                    log.warn("No polling period specifically set, defaulting to zero which will have the thread blocked until {} threads request messages",
-                            triggerValue);
-                    return 0L;
-                });
+    private long getPollingPeriodInMs() {
+        return PropertyUtils.safelyGetLongValue(
+                "messageRetrievalPollingPeriodInMs",
+                properties::getMessageRetrievalPollingPeriodInMs,
+                0L
+        );
     }
 }
diff --git a/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/retriever/batching/BatchingMessageRetrieverConstants.java b/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/retriever/batching/BatchingMessageRetrieverConstants.java
@@ -8,4 +8,9 @@ public class BatchingMessageRetrieverConstants {
      * The default amount of time to sleep the thread when there was an error obtaining messages.
      */
     public static final int DEFAULT_BACKOFF_TIME_IN_MS = 10_000;
+
+    /**
+     * The default number of threads requesting messages for it trigger a request for messages.
+     */
+    public static final int DEFAULT_BATCHING_TRIGGER = 1;
 }
diff --git a/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/retriever/prefetch/PrefetchingMessageRetriever.java b/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.RetrieverUtils;
+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;
@@ -157,12 +157,12 @@ private ReceiveMessageRequest buildReceiveMessageRequest() {
         final ReceiveMessageRequest.Builder requestBuilder = ReceiveMessageRequest
                 .builder()
                 .queueUrl(queueProperties.getQueueUrl())
-                .waitTimeSeconds(RetrieverUtils.safelyGetWaitTimeInSeconds(properties.getMessageWaitTimeInSeconds()))
+                .waitTimeSeconds(RetrieverUtils.safelyGetWaitTimeInSeconds(properties::getMessageWaitTimeInSeconds))
                 .maxNumberOfMessages(numberOfMessagesToObtain);
         final Integer visibilityTimeoutInSeconds = properties.getVisibilityTimeoutForMessagesInSeconds();
         if (visibilityTimeoutInSeconds != null) {
             if (visibilityTimeoutInSeconds < 0) {
-                log.warn("Non-positive visibilityTimeoutInSeconds provided: ", visibilityTimeoutInSeconds);
+                log.warn("Non-positive visibilityTimeoutInSeconds provided: {}", visibilityTimeoutInSeconds);
             } else {
                 requestBuilder.visibilityTimeout(visibilityTimeoutInSeconds);
             }
@@ -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.safelyGetPositiveOrZeroIntegerValue(
+                "errorBackoffTimeInMilliseconds",
+                properties::getErrorBackoffTimeInMilliseconds,
+                DEFAULT_ERROR_BACKOFF_TIMEOUT_IN_MILLISECONDS
+        );
     }
 }
diff --git a/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/util/RetrieverUtils.java b/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/util/RetrieverUtils.java
diff --git a/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/util/retriever/RetrieverUtils.java b/java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/util/retriever/RetrieverUtils.java
@@ -0,0 +1,39 @@
+package com.jashmore.sqs.util.retriever;
+
+import static com.jashmore.sqs.aws.AwsConstants.MAX_SQS_RECEIVE_WAIT_TIME_IN_SECONDS;
+
+import com.jashmore.sqs.aws.AwsConstants;
+import com.jashmore.sqs.util.properties.PropertyUtils;
+import lombok.experimental.UtilityClass;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.function.Supplier;
+
+/**
+ * Utility functions that are helpful for the {@link com.jashmore.sqs.retriever.MessageRetriever}.
+ */
+@Slf4j
+@UtilityClass
+public class RetrieverUtils {
+    /**
+     * Gets the wait time in seconds, defaulting to {@link AwsConstants#MAX_SQS_RECEIVE_WAIT_TIME_IN_SECONDS} if it is not present or invalid.
+     *
+     * @param unsafeWaitTimeInSeconds the original wait time in seconds
+     * @return the amount of time to wait for messages from SQS
+     */
+    public int safelyGetWaitTimeInSeconds(final Supplier<Integer> unsafeWaitTimeInSeconds) {
+        final int waitTimeInSeconds = PropertyUtils.safelyGetPositiveIntegerValue(
+                "messageWaitTimeInSeconds",
+                unsafeWaitTimeInSeconds,
+                MAX_SQS_RECEIVE_WAIT_TIME_IN_SECONDS
+        );
+
+        if (waitTimeInSeconds > MAX_SQS_RECEIVE_WAIT_TIME_IN_SECONDS) {
+            log.warn("messageWaitTimeInSeconds provided({}) is greater than AWS maximum({}), using default instead", waitTimeInSeconds,
+                    AwsConstants.MAX_SQS_RECEIVE_WAIT_TIME_IN_SECONDS);
+            return MAX_SQS_RECEIVE_WAIT_TIME_IN_SECONDS;
+        }
+
+        return waitTimeInSeconds;
+    }
+}
