Merge pull request #93 from JaidenAshmore/issue/91_concurrent_message_broker_error_safety

refs #91: Added more safety to the ConcurrentMessageBroker if the properties throw exceptions.

Author: JaidenAshmore

examples/java-dynamic-sqs-listener-core-examples/src/main/java/com/jashmore/sqs/examples/ConcurrentBrokerExample.java

@@ -49,6 +49,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.IntStream;
 import javax.validation.constraints.Min;
+import javax.validation.constraints.PositiveOrZero;
 
 /**
  * This example shows the core framework being used to processing messages place onto the queue with a dynamic level of concurrency via the
@@ -63,7 +64,7 @@
  */
 @Slf4j
 public class ConcurrentBrokerExample {
-    private static final int CONCURRENCY_LEVEL_PERIOD_IN_MS = 5000;
+    private static final long CONCURRENCY_LEVEL_PERIOD_IN_MS = 5000L;
     private static final int CONCURRENCY_LEVEL_LIMIT = 10;
 
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
@@ -130,14 +131,19 @@ public Integer getConcurrencyLevel() {
                     }
 
                     @Override
-                    public @Min(0) Integer getPreferredConcurrencyPollingRateInMilliseconds() {
+                    public @Min(0) Long getPreferredConcurrencyPollingRateInMilliseconds() {
                         return CONCURRENCY_LEVEL_PERIOD_IN_MS;
                     }
 
                     @Override
-                    public String threadNameFormat() {
+                    public String getThreadNameFormat() {
                         return "my-message-listener-thread-%d";
                     }
+
+                    @Override
+                    public @PositiveOrZero Long getErrorBackoffTimeInMilliseconds() {
+                        return 0L;
+                    }
                 })
         );
 

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/broker/concurrent/CachingConcurrentMessageBrokerProperties.java

@@ -7,7 +7,9 @@
 import net.jcip.annotations.ThreadSafe;
 
 import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
 import javax.validation.constraints.Min;
+import javax.validation.constraints.PositiveOrZero;
 
 /**
  * Implementation that will cache the values as the methods to retrieve the values may be costly.
@@ -24,7 +26,8 @@ public class CachingConcurrentMessageBrokerProperties implements ConcurrentMessa
     private static final int SINGLE_CACHE_VALUE_KEY = 0;
 
     private final LoadingCache<Integer, Integer> cachedConcurrencyLevel;
-    private final LoadingCache<Integer, Integer> cachedPreferredConcurrencyPollingRateInSeconds;
+    private final LoadingCache<Integer, Long> cachedPreferredConcurrencyPollingRateInSeconds;
+    private final LoadingCache<Integer, Long> cachedErrorBackoffTimeInMilliseconds;
     private final String threadNameFormat;
 
     /**
@@ -43,7 +46,11 @@ public CachingConcurrentMessageBrokerProperties(final int cachingTimeoutInMs,
                 .expireAfterWrite(cachingTimeoutInMs, TimeUnit.MILLISECONDS)
                 .build(CacheLoader.from(delegateProperties::getPreferredConcurrencyPollingRateInMilliseconds));
 
-        this.threadNameFormat = delegateProperties.threadNameFormat();
+        this.cachedErrorBackoffTimeInMilliseconds = CacheBuilder.newBuilder()
+                .expireAfterWrite(cachingTimeoutInMs, TimeUnit.MILLISECONDS)
+                .build(CacheLoader.from(delegateProperties::getErrorBackoffTimeInMilliseconds));
+
+        this.threadNameFormat = delegateProperties.getThreadNameFormat();
     }
 
     @Override
@@ -52,12 +59,18 @@ public CachingConcurrentMessageBrokerProperties(final int cachingTimeoutInMs,
     }
 
     @Override
-    public @Min(0) Integer getPreferredConcurrencyPollingRateInMilliseconds() {
+    public @Min(0) Long getPreferredConcurrencyPollingRateInMilliseconds() {
         return cachedPreferredConcurrencyPollingRateInSeconds.getUnchecked(SINGLE_CACHE_VALUE_KEY);
     }
 
     @Override
-    public String threadNameFormat() {
+    public String getThreadNameFormat() {
         return threadNameFormat;
     }
+
+    @Nullable
+    @Override
+    public @PositiveOrZero Long getErrorBackoffTimeInMilliseconds() {
+        return cachedErrorBackoffTimeInMilliseconds.getUnchecked(SINGLE_CACHE_VALUE_KEY);
+    }
 }

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/broker/concurrent/ConcurrentMessageBroker.java

@@ -1,14 +1,17 @@
 package com.jashmore.sqs.broker.concurrent;
 
+import static com.jashmore.sqs.broker.concurrent.ConcurrentMessageBrokerConstants.DEFAULT_BACKOFF_TIME_IN_MS;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
 
+import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 import com.jashmore.sqs.broker.MessageBroker;
 import com.jashmore.sqs.processor.MessageProcessor;
 import com.jashmore.sqs.retriever.MessageRetriever;
 import com.jashmore.sqs.util.ResizableSemaphore;
+import com.jashmore.sqs.util.properties.PropertyUtils;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import lombok.extern.slf4j.Slf4j;
 
@@ -46,30 +49,41 @@ public void run() {
         final ResizableSemaphore concurrentMessagesBeingProcessedSemaphore = new ResizableSemaphore(0);
 
         while (!Thread.currentThread().isInterrupted()) {
-            updateConcurrencyLevelIfChanged(concurrentMessagesBeingProcessedSemaphore);
-
-            final long numberOfMillisecondsToObtainPermit = properties.getPreferredConcurrencyPollingRateInMilliseconds();
             try {
-                final boolean obtainedPermit = concurrentMessagesBeingProcessedSemaphore.tryAcquire(numberOfMillisecondsToObtainPermit, MILLISECONDS);
-                if (!obtainedPermit) {
+                updateConcurrencyLevelIfChanged(concurrentMessagesBeingProcessedSemaphore);
+
+                final long numberOfMillisecondsToObtainPermit = getNumberOfMillisecondsToObtainPermit();
+                try {
+                    final boolean obtainedPermit = concurrentMessagesBeingProcessedSemaphore.tryAcquire(numberOfMillisecondsToObtainPermit, MILLISECONDS);
+                    if (!obtainedPermit) {
+                        continue;
+                    }
+                } catch (final InterruptedException interruptedException) {
+                    log.debug("Interrupted exception caught while adding more listeners, shutting down!");
+                    Thread.currentThread().interrupt();
                     continue;
                 }
-            } catch (final InterruptedException interruptedException) {
-                log.info("Interrupted exception caught while adding more listeners, shutting down!");
-                Thread.currentThread().interrupt();
-                continue;
-            }
 
-            CompletableFuture.supplyAsync(() -> {
+                CompletableFuture.supplyAsync(() -> {
+                    try {
+                        return messageRetriever.retrieveMessage();
+                    } catch (final InterruptedException exception) {
+                        log.debug("Thread interrupted waiting for a message");
+                        throw new RuntimeException("Failure to get message");
+                    }
+                }, concurrentThreadsExecutor)
+                        .thenAcceptAsync(messageProcessor::processMessage, messageProcessingThreadsExecutor)
+                        .whenComplete((ignoredResult, throwable) -> concurrentMessagesBeingProcessedSemaphore.release());
+            } catch (final Throwable throwable) {
                 try {
-                    return messageRetriever.retrieveMessage();
-                } catch (final InterruptedException exception) {
-                    log.trace("Thread interrupted waiting for a message");
-                    throw new RuntimeException("Failure to get message");
+                    final long errorBackoffTimeInMilliseconds = getErrorBackoffTimeInMilliseconds();
+                    log.error("Error thrown while organising threads to process messages. Backing off for {}ms", errorBackoffTimeInMilliseconds, throwable);
+                    Thread.sleep(errorBackoffTimeInMilliseconds);
+                } catch (final InterruptedException interruptedException) {
+                    log.debug("Thread interrupted during backoff period");
+                    Thread.currentThread().interrupt();
                 }
-            }, concurrentThreadsExecutor)
-                    .thenAcceptAsync(messageProcessor::processMessage, messageProcessingThreadsExecutor)
-                    .whenComplete((ignoredResult, throwable) -> concurrentMessagesBeingProcessedSemaphore.release());
+            }
         }
 
         log.debug("Shutting down message controller");
@@ -88,6 +102,7 @@ public void run() {
      */
     private void updateConcurrencyLevelIfChanged(final ResizableSemaphore resizableSemaphore) {
         final int newConcurrencyLevel = properties.getConcurrencyLevel();
+        Preconditions.checkArgument(newConcurrencyLevel >= 0, "concurrencyLevel should be non-negative");
 
         if (resizableSemaphore.getMaximumPermits() != newConcurrencyLevel) {
             log.debug("Changing concurrency from {} to {}", resizableSemaphore.getMaximumPermits(), newConcurrencyLevel);
@@ -111,13 +126,54 @@ private void shutdownConcurrentThreads(final ExecutorService concurrentThreadsEx
         }
     }
 
+    /**
+     * Build the {@link ExecutorService} that will be used for the threads that are processing the messages.
+     *
+     * <p>This will provide the logic to name the threads to make it easier to debug multiple different message listeners in the system.
+     *
+     * @return the executor service that will be used for processing messages
+     */
     private ExecutorService buildMessageProcessingExecutorService() {
-        final ThreadFactoryBuilder threadFactoryBuilder = new ThreadFactoryBuilder();
+        try {
+            final ThreadFactoryBuilder threadFactoryBuilder = new ThreadFactoryBuilder();
 
-        if (properties.threadNameFormat() != null) {
-            threadFactoryBuilder.setNameFormat(properties.threadNameFormat());
+            final String threadNameFormat = properties.getThreadNameFormat();
+            if (threadNameFormat != null) {
+                threadFactoryBuilder.setNameFormat(threadNameFormat);
+            }
+
+            return Executors.newCachedThreadPool(threadFactoryBuilder.build());
+        } catch (final Throwable throwable) {
+            log.error("Error thrown building message processing executor service, returning default");
+            return Executors.newCachedThreadPool();
         }
+    }
 
-        return Executors.newCachedThreadPool(threadFactoryBuilder.build());
+    /**
+     * 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 number of milliseconds that should wait to get a permit for creating a new thread.
+     *
+     * @return the number of milliseconds to wait
+     * @see ConcurrentMessageBrokerProperties#getPreferredConcurrencyPollingRateInMilliseconds() for more information
+     */
+    private long getNumberOfMillisecondsToObtainPermit() {
+        return PropertyUtils.safelyGetPositiveLongValue(
+                "preferredConcurrencyPollingRateInMilliseconds",
+                properties::getPreferredConcurrencyPollingRateInMilliseconds,
+                DEFAULT_BACKOFF_TIME_IN_MS
+        );
     }
 }

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/broker/concurrent/ConcurrentMessageBrokerConstants.java

@@ -0,0 +1,16 @@
+package com.jashmore.sqs.broker.concurrent;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class ConcurrentMessageBrokerConstants {
+    /**
+     * The default amount of time to sleep the thread when there was an error organising the message processing threads.
+     */
+    public static final int DEFAULT_BACKOFF_TIME_IN_MS = 10_000;
+
+    /**
+     * The default amount of time the thread should wait for a thread to process a message before it tries again and checks the available concurrency.
+     */
+    public static final long DEFAULT_CONCURRENCY_POLLING_IN_MS = 60_000L;
+}

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/broker/concurrent/ConcurrentMessageBrokerProperties.java

@@ -2,8 +2,10 @@
 
 import com.jashmore.sqs.retriever.MessageRetriever;
 
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.NotThreadSafe;
 import javax.validation.constraints.Min;
+import javax.validation.constraints.PositiveOrZero;
 
 /**
  * Properties for dynamically configuring how the {@link ConcurrentMessageBroker} is able to process messages concurrently.
@@ -68,7 +70,7 @@ public interface ConcurrentMessageBrokerProperties {
      * @return the number of milliseconds between polls for the concurrency level
      */
     @Min(0)
-    Integer getPreferredConcurrencyPollingRateInMilliseconds();
+    Long getPreferredConcurrencyPollingRateInMilliseconds();
 
     /**
      * The String format for the name of the children threads that process the message.
@@ -80,5 +82,20 @@ public interface ConcurrentMessageBrokerProperties {
      *
      * @return the format for the name of threads that process messages
      */
-    String threadNameFormat();
+    @Nullable
+    String getThreadNameFormat();
+
+    /**
+     * The number of milliseconds that the background thread for organising concurrent threads should backoff before attempting again.
+     *
+     * <p>This is needed to stop the background thread from trying again and again over and over causing a flood of error log messages that may make it
+     * difficult to debug.
+     *
+     * <p>If this value is null, negative or zero, {@link ConcurrentMessageBrokerConstants#DEFAULT_BACKOFF_TIME_IN_MS} will be used as the backoff period.
+     *
+     * @return the number of milliseconds to sleep the thread after an error is thrown
+     */
+    @Nullable
+    @PositiveOrZero
+    Long getErrorBackoffTimeInMilliseconds();
 }

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/broker/concurrent/StaticConcurrentMessageBrokerProperties.java

@@ -1,5 +1,7 @@
 package com.jashmore.sqs.broker.concurrent;
 
+import static com.jashmore.sqs.broker.concurrent.ConcurrentMessageBrokerConstants.DEFAULT_CONCURRENCY_POLLING_IN_MS;
+
 import com.google.common.base.Preconditions;
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -9,8 +11,11 @@
 import lombok.ToString;
 import net.jcip.annotations.ThreadSafe;
 
+import java.util.Locale;
 import java.util.Optional;
+import javax.annotation.Nullable;
 import javax.validation.constraints.Min;
+import javax.validation.constraints.PositiveOrZero;
 
 /**
  * Implementation that stores the value as non-mutable field values and therefore will return the same value on every call.
@@ -24,20 +29,18 @@
 @Builder(toBuilder = true)
 @ThreadSafe
 public final class StaticConcurrentMessageBrokerProperties implements ConcurrentMessageBrokerProperties {
-    private static final Integer DEFAULT_CONCURRENCY_POLLING_IN_MS = 60_000;
-
     @NonNull
     private final Integer concurrencyLevel;
-
     @NonNull
-    private final Integer preferredConcurrencyPollingRateInMilliseconds;
-
+    private final Long preferredConcurrencyPollingRateInMilliseconds;
     private final String threadNameFormat;
+    private final Long errorBackoffTimeInMilliseconds;
 
     @SuppressFBWarnings("RV_RETURN_VAL")
     public StaticConcurrentMessageBrokerProperties(final Integer concurrencyLevel,
-                                                   final Integer preferredConcurrencyPollingRateInMilliseconds,
-                                                   final String threadNameFormat) {
+                                                   final Long preferredConcurrencyPollingRateInMilliseconds,
+                                                   final String threadNameFormat,
+                                                   final Long errorBackoffTimeInMilliseconds) {
         Preconditions.checkArgument(concurrencyLevel == null || concurrencyLevel >= 0, "concurrencyLevel should be greater than or equal to zero");
         Preconditions.checkArgument(preferredConcurrencyPollingRateInMilliseconds == null || preferredConcurrencyPollingRateInMilliseconds >= 0,
                 "preferredConcurrencyPollingRateInMilliseconds should be greater than or equal to zero");
@@ -46,12 +49,14 @@ public StaticConcurrentMessageBrokerProperties(final Integer concurrencyLevel,
                 .orElse(0);
         this.preferredConcurrencyPollingRateInMilliseconds = Optional.ofNullable(preferredConcurrencyPollingRateInMilliseconds)
                 .orElse(DEFAULT_CONCURRENCY_POLLING_IN_MS);
+        this.errorBackoffTimeInMilliseconds = errorBackoffTimeInMilliseconds;
 
         this.threadNameFormat = threadNameFormat;
         if (threadNameFormat != null) {
+
             // Test that the thread name is in the correct format
             //noinspection ResultOfMethodCallIgnored
-            String.format(threadNameFormat, 0);
+            String.format(Locale.ROOT, threadNameFormat, 1);
         }
     }
 
@@ -61,12 +66,18 @@ public StaticConcurrentMessageBrokerProperties(final Integer concurrencyLevel,
     }
 
     @Override
-    public @Min(0) Integer getPreferredConcurrencyPollingRateInMilliseconds() {
+    public @Min(0) Long getPreferredConcurrencyPollingRateInMilliseconds() {
         return preferredConcurrencyPollingRateInMilliseconds;
     }
 
     @Override
-    public String threadNameFormat() {
+    public String getThreadNameFormat() {
         return threadNameFormat;
     }
+
+    @Nullable
+    @Override
+    public @PositiveOrZero Long getErrorBackoffTimeInMilliseconds() {
+        return errorBackoffTimeInMilliseconds;
+    }
 }

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/retriever/batching/BatchingMessageRetrieverProperties.java

@@ -59,7 +59,7 @@ public interface BatchingMessageRetrieverProperties {
      * set on the SQS queue will be used.
      *
      * @return the visibility timeout for the message
-     * @see ReceiveMessageRequest#visibilityTimeout for where this is applied against
+     * @see ReceiveMessageRequest#visibilityTimeout() for where this is applied against
      */
     @Nullable
     @Positive
@@ -73,7 +73,7 @@ public interface BatchingMessageRetrieverProperties {
      * <p>If this value is null, the {@link AwsConstants#MAX_SQS_RECEIVE_WAIT_TIME_IN_SECONDS} will be used when requesting messages from SQS.
      *
      * @return the wait time in seconds for obtaining messages
-     * @see ReceiveMessageRequest#waitTimeSeconds for the usage
+     * @see ReceiveMessageRequest#waitTimeSeconds()  for the usage
      */
     @Nullable
     @PositiveOrZero