refs #231: Converts properties to use Durations and not Numbers (#235)

This converts a lot of properties from property values like timeoutInSeconds
as timeout using the Java duration type. This makes the code easier to read
and reduces errors in submitting the wrong values.

Author: JaidenAshmore

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

@@ -1,6 +1,8 @@
 package com.jashmore.sqs.broker.concurrent;
 
-import static com.jashmore.sqs.broker.concurrent.ConcurrentMessageBrokerConstants.DEFAULT_BACKOFF_TIME_IN_MS;
+import static com.jashmore.sqs.broker.concurrent.ConcurrentMessageBrokerConstants.DEFAULT_BACKOFF_TIME;
+import static com.jashmore.sqs.broker.concurrent.ConcurrentMessageBrokerConstants.DEFAULT_CONCURRENCY_POLLING;
+import static com.jashmore.sqs.util.properties.PropertyUtils.safelyGetPositiveOrZeroDuration;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 import com.jashmore.sqs.broker.MessageBroker;
@@ -9,6 +11,7 @@
 import lombok.extern.slf4j.Slf4j;
 import software.amazon.awssdk.services.sqs.model.Message;
 
+import java.time.Duration;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
@@ -53,8 +56,8 @@ public void processMessages(final ExecutorService messageProcessingExecutorServi
             try {
                 updateConcurrencyLevelIfChanged(concurrentMessagesBeingProcessedSemaphore);
 
-                final long numberOfMillisecondsToObtainPermit = getNumberOfMillisecondsToObtainPermit();
-                final boolean obtainedPermit = concurrentMessagesBeingProcessedSemaphore.tryAcquire(numberOfMillisecondsToObtainPermit, MILLISECONDS);
+                final Duration permitWaitTime = getPermitWaitTime();
+                final boolean obtainedPermit = concurrentMessagesBeingProcessedSemaphore.tryAcquire(permitWaitTime.toMillis(), MILLISECONDS);
                 if (!obtainedPermit) {
                     continue;
                 }
@@ -74,10 +77,11 @@ public void processMessages(final ExecutorService messageProcessingExecutorServi
                     throw runtimeException;
                 }
             } catch (final RuntimeException runtimeException) {
-                final long errorBackoffTimeInMilliseconds = getErrorBackoffTimeInMilliseconds();
-                log.error("Error thrown while organising threads to process messages. Backing off for {}ms", errorBackoffTimeInMilliseconds,
+                final Duration errorBackoffTime = safelyGetPositiveOrZeroDuration("errorBackoffTime", properties::getErrorBackoffTime, DEFAULT_BACKOFF_TIME);
+                final long errorBackoffTimeInMs = errorBackoffTime.toMillis();
+                log.error("Error thrown while organising threads to process messages. Backing off for {}ms", errorBackoffTimeInMs,
                         runtimeException);
-                Thread.sleep(errorBackoffTimeInMilliseconds);
+                Thread.sleep(errorBackoffTimeInMs);
             }
         }
         log.debug("Ending processing of messages");
@@ -86,15 +90,16 @@ public void processMessages(final ExecutorService messageProcessingExecutorServi
     /**
      * 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#getConcurrencyPollingRateInMilliseconds() for more information
+     * @return the duration to wait for an available permit
+     * @see ConcurrentMessageBrokerProperties#getConcurrencyPollingRate() for more information
      */
-    private long getNumberOfMillisecondsToObtainPermit() {
-        return PropertyUtils.safelyGetPositiveLongValue(
-                "numberOfMillisecondsToObtainPermit",
-                properties::getConcurrencyPollingRateInMilliseconds,
-                DEFAULT_BACKOFF_TIME_IN_MS
-        );
+    private Duration getPermitWaitTime() {
+        final Duration pollingRate = properties.getConcurrencyPollingRate();
+        if (pollingRate != null && !pollingRate.isNegative()) {
+            return pollingRate;
+        }
+
+        return DEFAULT_CONCURRENCY_POLLING;
     }
 
     /**
@@ -123,18 +128,4 @@ private int getConcurrencyLevel() {
                 0
         );
     }
-
-    /**
-     * 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
-        );
-    }
 }

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

@@ -2,15 +2,17 @@
 
 import lombok.experimental.UtilityClass;
 
+import java.time.Duration;
+
 @UtilityClass
 class ConcurrentMessageBrokerConstants {
     /**
      * The default amount of time to sleep the thread when there was an error organising the message processing threads.
      */
-    static final int DEFAULT_BACKOFF_TIME_IN_MS = 10_000;
+    static final Duration DEFAULT_BACKOFF_TIME = Duration.ofSeconds(10);
 
     /**
      * 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.
      */
-    static final long DEFAULT_CONCURRENCY_POLLING_IN_MS = 60_000L;
+    static final Duration DEFAULT_CONCURRENCY_POLLING = Duration.ofMinutes(1);
 }

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

@@ -2,8 +2,11 @@
 
 import com.jashmore.documentation.annotations.NotThreadSafe;
 import com.jashmore.documentation.annotations.Nullable;
+import com.jashmore.documentation.annotations.Positive;
 import com.jashmore.documentation.annotations.PositiveOrZero;
 
+import java.time.Duration;
+
 /**
  * Properties for dynamically configuring how the {@link ConcurrentMessageBroker} is able to process messages concurrently.
  *
@@ -25,7 +28,7 @@ public interface ConcurrentMessageBrokerProperties {
      * <p>The {@link ConcurrentMessageBroker} will maintain the rate of concurrency by checking that the current concurrency rate is aligned with this value.
      * If there is currently less messages being processed than this value, more requests for messages will be made to meet this value. If there are
      * more messages being processed or requested than this value, this coordinating thread will block until enough messages have been processed.  If a
-     * permit has not be obtained before the timeout defined by {@link #getConcurrencyPollingRateInMilliseconds()}, it will recalculate this
+     * permit has not be obtained before the timeout defined by {@link #getConcurrencyPollingRate()}, it will recalculate this
      * concurrency rate again and wait for a permit.
      *
      * <p>Note that the concurrency rate does not get applied instantly and there are multiple attributing factors for when the rate of concurrency will
@@ -40,7 +43,7 @@ public interface ConcurrentMessageBrokerProperties {
      *     </li>
      *     <li>
      *         The delay in the concurrency rate change may be due to transitioning from allowing zero threads to a number of threads. In a worst
-     *         case scenario there would be a delay of {@link #getConcurrencyPollingRateInMilliseconds()} before the rate of concurrency is changed.
+     *         case scenario there would be a delay of {@link #getConcurrencyPollingRate()} before the rate of concurrency is changed.
      *     </li>
      * </ul>
      *
@@ -50,7 +53,7 @@ public interface ConcurrentMessageBrokerProperties {
     int getConcurrencyLevel();
 
     /**
-     * The number of milliseconds that the coordinating thread will sleep when the maximum rate of concurrency has been reached before checking the
+     * The duration that the coordinating thread will sleep when the maximum rate of concurrency has been reached before checking the
      * concurrency rate again.
      *
      * <p>The reason that this property is needed is because during the processing of messages, which could be a significantly long period, the rate
@@ -62,25 +65,27 @@ public interface ConcurrentMessageBrokerProperties {
      * that the coordinating thread is awoken and is therefore less CPU intensive. However, decreasing this polling period makes it more responsive to changes
      * to the rate of concurrency.
      *
-     * <p>If this value is null or less than zero, {@link ConcurrentMessageBrokerConstants#DEFAULT_CONCURRENCY_POLLING_IN_MS} will be used instead.
+     * <p>If this duration is null or negative, {@link ConcurrentMessageBrokerConstants#DEFAULT_CONCURRENCY_POLLING} will be used instead. It is not
+     * recommended to have a low value as that will result in this background thread constantly trying to determine fi the concurrency rate can be
+     * changed.
      *
-     * @return the number of milliseconds between polls for the concurrency level
+     * @return the amount of time between polls for the concurrency level
      */
     @Nullable
-    @PositiveOrZero
-    Long getConcurrencyPollingRateInMilliseconds();
+    @Positive
+    Duration getConcurrencyPollingRate();
 
     /**
-     * The number of milliseconds that the coordinating thread should backoff if there was an error trying to request a message.
+     * The duration that the coordinating thread should backoff if there was an error trying to request a message.
      *
      * <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 or negative, {@link ConcurrentMessageBrokerConstants#DEFAULT_BACKOFF_TIME_IN_MS} will be used as the backoff period.
+     * <p>If this value is null or negative, {@link ConcurrentMessageBrokerConstants#DEFAULT_BACKOFF_TIME} will be used as the backoff period.
      *
-     * @return the number of milliseconds to sleep the thread after an error is thrown
+     * @return the amount of time to sleep the thread after an error is thrown
      */
     @Nullable
     @PositiveOrZero
-    Long getErrorBackoffTimeInMilliseconds();
+    Duration getErrorBackoffTime();
 }

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

@@ -8,6 +8,8 @@
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
 
+import java.time.Duration;
+
 /**
  * Implementation that stores the value as non-mutable field values and therefore will return the same value on every call.
  *
@@ -21,20 +23,18 @@
 @ThreadSafe
 public final class StaticConcurrentMessageBrokerProperties implements ConcurrentMessageBrokerProperties {
     private final Integer concurrencyLevel;
-    private final Long preferredConcurrencyPollingRateInMilliseconds;
-    private final Long errorBackoffTimeInMilliseconds;
+    private final Duration preferredConcurrencyPollingRate;
+    private final Duration errorBackoffTime;
 
     public StaticConcurrentMessageBrokerProperties(final Integer concurrencyLevel,
-                                                   final Long preferredConcurrencyPollingRateInMilliseconds,
-                                                   final Long errorBackoffTimeInMilliseconds) {
+                                                   final Duration preferredConcurrencyPollingRate,
+                                                   final Duration errorBackoffTime) {
         Preconditions.checkNotNull(concurrencyLevel, "concurrencyLevel should not be null");
         Preconditions.checkPositiveOrZero(concurrencyLevel, "concurrencyLevel should be greater than or equal to zero");
-        Preconditions.checkArgument(preferredConcurrencyPollingRateInMilliseconds == null || preferredConcurrencyPollingRateInMilliseconds >= 0,
-                "preferredConcurrencyPollingRateInMilliseconds should null or greater than or equal to zero");
 
         this.concurrencyLevel = concurrencyLevel;
-        this.preferredConcurrencyPollingRateInMilliseconds = preferredConcurrencyPollingRateInMilliseconds;
-        this.errorBackoffTimeInMilliseconds = errorBackoffTimeInMilliseconds;
+        this.preferredConcurrencyPollingRate = preferredConcurrencyPollingRate;
+        this.errorBackoffTime = errorBackoffTime;
     }
 
     @PositiveOrZero
@@ -43,16 +43,17 @@ public int getConcurrencyLevel() {
         return concurrencyLevel;
     }
 
+    @Nullable
     @PositiveOrZero
     @Override
-    public Long getConcurrencyPollingRateInMilliseconds() {
-        return preferredConcurrencyPollingRateInMilliseconds;
+    public Duration getConcurrencyPollingRate() {
+        return preferredConcurrencyPollingRate;
     }
 
     @Nullable
     @PositiveOrZero
     @Override
-    public Long getErrorBackoffTimeInMilliseconds() {
-        return errorBackoffTimeInMilliseconds;
+    public Duration getErrorBackoffTime() {
+        return errorBackoffTime;
     }
 }