Merge pull request #100 from JaidenAshmore/issue/97_remove_throwable_catching

refs #97: Remove usages of catch Throwable

Author: JaidenAshmore

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/argument/DelegatingArgumentResolverService.java

@@ -23,12 +23,13 @@ public Object resolveArgument(final QueueProperties queueProperties, final Metho
                 .map(resolver -> {
                     try {
                         return resolver.resolveArgumentForParameter(queueProperties, methodParameter, message);
-                    } catch (final Throwable throwable) {
-                        if (ArgumentResolutionException.class.isAssignableFrom(throwable.getClass())) {
-                            throw throwable;
+                    } catch (final RuntimeException runtimeException) {
+                        // Make sure to wrap any unintended exceptions with the expected exception for errors
+                        if (!ArgumentResolutionException.class.isAssignableFrom(runtimeException.getClass())) {
+                            throw new ArgumentResolutionException("Error obtaining an argument value for parameter", runtimeException);
                         }
 
-                        throw new ArgumentResolutionException("Error obtaining an argument value for parameter", throwable);
+                        throw runtimeException;
                     }
                 })
                 .findFirst()

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

@@ -90,8 +90,8 @@ public void run() {
         try {
             shutdownConcurrentThreads(concurrentThreadsExecutor, messageProcessingThreadsExecutor);
             log.debug("Message controller shut down");
-        } catch (final Throwable throwable) {
-            log.error("Exception thrown while waiting for broker to shutdown", throwable);
+        } catch (final RuntimeException runtimeException) {
+            log.error("Exception thrown while waiting for broker to shutdown", runtimeException);
         }
     }
 

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/processor/DefaultMessageProcessor.java

@@ -10,6 +10,7 @@
 import lombok.AllArgsConstructor;
 import software.amazon.awssdk.services.sqs.model.Message;
 
+import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Parameter;
 import java.util.Arrays;
@@ -39,8 +40,8 @@ public void processMessage(final Message message) throws MessageProcessingExcept
         final Object result;
         try {
             result = messageConsumerMethod.invoke(messageConsumerBean, arguments);
-        } catch (final Throwable throwable) {
-            throw new MessageProcessingException("Error processing message", throwable);
+        } catch (final InvocationTargetException | IllegalAccessException | RuntimeException exception) {
+            throw new MessageProcessingException("Error processing message", exception);
         }
 
         if (hasAcknowledgeParameter()) {

java-dynamic-sqs-listener-core/src/main/java/com/jashmore/sqs/processor/retryable/RetryableMessageProcessor.java

@@ -35,18 +35,18 @@ public void processMessage(final Message message) throws MessageProcessingExcept
     private void processMessageWithRetries(final Message message, final int retryAttempts) throws MessageProcessingException {
         try {
             delegateMessageProcessor.processMessage(message);
-        } catch (final Throwable throwable) {
-            if (retryAttempts > 0) {
-                log.error("Error processing message. Trying " + retryAttempts + " more times", throwable);
-                try {
-                    backoff();
-                } catch (final InterruptedException exception) {
-                    throw new MessageProcessingException("Interrupted while processing message", exception);
-                }
-                processMessageWithRetries(message, retryAttempts - 1);
-            } else {
-                throw throwable;
+        } catch (final RuntimeException runtimeException) {
+            if (retryAttempts <= 0) {
+                throw runtimeException;
             }
+
+            log.error("Error processing message. Trying " + retryAttempts + " more times", runtimeException);
+            try {
+                backoff();
+            } catch (final InterruptedException exception) {
+                throw new MessageProcessingException("Interrupted while processing message", exception);
+            }
+            processMessageWithRetries(message, retryAttempts - 1);
         }
     }
 

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

@@ -20,6 +20,7 @@
 import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
 
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -132,6 +133,7 @@ public void run() {
                     log.debug("Thread interrupted while obtaining messages from SQS");
                     break;
                 }
+
                 try {
                     for (final Message message : response.messages()) {
                         messagesDownloaded.put(message);
@@ -140,11 +142,10 @@ public void run() {
                     log.debug("Thread interrupted while placing messages on internal queue");
                     break;
                 }
-            } catch (final Throwable throwable) {
-                log.error("Error thrown trying to obtain messages", throwable);
+            } catch (final ExecutionException | RuntimeException exception) {
                 try {
                     final long errorBackoffTimeInMilliseconds = getErrorBackoffTimeInMilliseconds();
-                    log.error("Error thrown while organising threads to process messages. Backing off for {}ms", errorBackoffTimeInMilliseconds, throwable);
+                    log.error("Error thrown while organising threads to process messages. Backing off for {}ms", errorBackoffTimeInMilliseconds, exception);
                     backoff(errorBackoffTimeInMilliseconds);
                 } catch (final InterruptedException interruptedException) {
                     log.debug("Thread interrupted during backoff period");

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

@@ -18,6 +18,7 @@
 
 import java.util.concurrent.BlockingDeque;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.SynchronousQueue;
 
@@ -131,8 +132,8 @@ public void run() {
                     log.debug("Thread interrupted while placing messages onto queue. Exiting...");
                     break;
                 }
-            } catch (final Throwable throwable) {
-                log.error("Exception thrown when retrieving messages", throwable);
+            } catch (final ExecutionException | RuntimeException exception) {
+                log.error("Exception thrown when retrieving messages", exception);
 
                 try {
                     Thread.sleep(getBackoffTimeInMs());