Merge pull request #120 from JaidenAshmore/issue/119_handles_sdk_interrupted_exceptions_in_message_retrievers

refs 119: handles Interruptions when retrieving messages as they don't send InterruptedExceptions

Author: JaidenAshmore

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

@@ -14,6 +14,8 @@
 import com.jashmore.sqs.util.retriever.RetrieverUtils;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import lombok.extern.slf4j.Slf4j;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.core.exception.SdkInterruptedException;
 import software.amazon.awssdk.services.sqs.SqsAsyncClient;
 import software.amazon.awssdk.services.sqs.model.Message;
 import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
@@ -139,11 +141,25 @@ public void run() {
                     for (final Message message : response.messages()) {
                         messagesDownloaded.put(message);
                     }
-                } catch (InterruptedException interruptedException) {
+                } catch (final InterruptedException interruptedException) {
                     log.debug("Thread interrupted while placing messages on internal queue");
                     break;
                 }
             } catch (final ExecutionException | RuntimeException exception) {
+                // Supposedly the SqsAsyncClient can get interrupted and this will remove the interrupted status from the thread and then wrap it
+                // in it's own version of the interrupted exception...If this happens when the retriever is being shut down it will keep on processing
+                // because it does not realise it is being shut down, therefore we have to check for this and quit if necessary
+                if (exception instanceof ExecutionException) {
+                    final Throwable executionExceptionCause = exception.getCause();
+                    if (executionExceptionCause instanceof SdkClientException) {
+                        if (executionExceptionCause.getCause() instanceof SdkInterruptedException) {
+                            log.debug("Thread interrupted while receiving messages");
+                            break;
+                        }
+                    }
+                }
+
+                log.error("Exception thrown when retrieving messages", exception);
                 try {
                     final long errorBackoffTimeInMilliseconds = getErrorBackoffTimeInMilliseconds();
                     log.error("Error thrown while organising threads to process messages. Backing off for {}ms", errorBackoffTimeInMilliseconds, exception);

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

@@ -11,6 +11,8 @@
 import com.jashmore.sqs.util.properties.PropertyUtils;
 import com.jashmore.sqs.util.retriever.RetrieverUtils;
 import lombok.extern.slf4j.Slf4j;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.core.exception.SdkInterruptedException;
 import software.amazon.awssdk.services.sqs.SqsAsyncClient;
 import software.amazon.awssdk.services.sqs.model.Message;
 import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
@@ -134,6 +136,19 @@ public void run() {
                     break;
                 }
             } catch (final ExecutionException | RuntimeException exception) {
+                // Supposedly the SqsAsyncClient can get interrupted and this will remove the interrupted status from the thread and then wrap it
+                // in it's own version of the interrupted exception...If this happens when the retriever is being shut down it will keep on processing
+                // because it does not realise it is being shut down, therefore we have to check for this and quit if necessary
+                if (exception instanceof ExecutionException) {
+                    final Throwable executionExceptionCause = exception.getCause();
+                    if (executionExceptionCause instanceof SdkClientException) {
+                        if (executionExceptionCause.getCause() instanceof SdkInterruptedException) {
+                            log.debug("Thread interrupted receiving messages");
+                            break;
+                        }
+                    }
+                }
+
                 log.error("Exception thrown when retrieving messages", exception);
 
                 try {

java-dynamic-sqs-listener-core/src/test/java/com/jashmore/sqs/retriever/batching/BatchingMessageRetrieverTest.java

@@ -2,6 +2,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -19,6 +20,8 @@
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.core.exception.SdkInterruptedException;
 import software.amazon.awssdk.services.sqs.SqsAsyncClient;
 import software.amazon.awssdk.services.sqs.model.Message;
 import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
@@ -559,6 +562,25 @@ public void whenEnoughThreadsRequestMessagesTheRetrievalOfMessagesWillTrigger()
         assertThat(actualMessage).isEqualTo(message);
     }
 
+    @Test
+    public void whenSqsAsyncClientThrowsSdkInterruptedExceptionTheBackgroundThreadShouldStop() throws Exception {
+        // arrange
+        final int threadsRequestingMessages = DEFAULT_PROPERTIES.getNumberOfThreadsWaitingTrigger();
+        final BatchingMessageRetriever backgroundThread = new BatchingMessageRetriever(QUEUE_PROPERTIES, sqsAsyncClient,
+                DEFAULT_PROPERTIES, new AtomicInteger(threadsRequestingMessages), new LinkedBlockingQueue<>(), new Object());
+
+        doThrow(new ExecutionException(SdkClientException.builder().cause(new SdkInterruptedException()).build())).when(responseThrowingException).get();
+        when(sqsAsyncClient.receiveMessage(any(ReceiveMessageRequest.class)))
+                .thenReturn(responseThrowingException);
+
+        // act
+        backgroundThread.run();
+
+        // assert
+        final ArgumentCaptor<ReceiveMessageRequest> receiveMessageRequestArgumentCaptor = ArgumentCaptor.forClass(ReceiveMessageRequest.class);
+        verify(sqsAsyncClient).receiveMessage(receiveMessageRequestArgumentCaptor.capture());
+    }
+
     private void responseThrowingException(final CompletableFuture<ReceiveMessageResponse> mockResponse,
                                            final Throwable throwable) {
         try {

java-dynamic-sqs-listener-core/src/test/java/com/jashmore/sqs/retriever/prefetch/PrefetchingMessageRetrieverTest.java

@@ -20,6 +20,8 @@
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.core.exception.SdkInterruptedException;
 import software.amazon.awssdk.services.sqs.SqsAsyncClient;
 import software.amazon.awssdk.services.sqs.model.Message;
 import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
@@ -631,6 +633,26 @@ public void cancellingRetrieverDuringBackoffWillNotRequestMoreMessages() throws
         verify(sqsAsyncClient, times(1)).receiveMessage(any(ReceiveMessageRequest.class));
     }
 
+    @Test
+    public void whenSqsAsyncClientThrowsSdkInterruptedExceptionTheBackgroundThreadShouldStop() throws Exception {
+        // arrange
+        final LinkedBlockingQueue<Message> internalMessageQueue = new LinkedBlockingQueue<>();
+        internalMessageQueue.add(Message.builder().build());
+        doThrow(new ExecutionException(SdkClientException.builder().cause(new SdkInterruptedException()).build())).when(responseThrowingException).get();
+        when(sqsAsyncClient.receiveMessage(any(ReceiveMessageRequest.class)))
+                .thenReturn(responseThrowingException);
+        final PrefetchingMessageRetriever backgroundMessagePrefetcher
+                = new PrefetchingMessageRetriever(sqsAsyncClient, QUEUE_PROPERTIES, DEFAULT_PREFETCHING_PROPERTIES,
+                internalMessageQueue, 5);
+
+        // act
+        backgroundMessagePrefetcher.run();
+
+        // assert
+        final ArgumentCaptor<ReceiveMessageRequest> receiveMessageRequestArgumentCaptor = ArgumentCaptor.forClass(ReceiveMessageRequest.class);
+        verify(sqsAsyncClient).receiveMessage(receiveMessageRequestArgumentCaptor.capture());
+    }
+
     private void responseThrowingException(final CompletableFuture<ReceiveMessageResponse> mockResponse,
                                            final Throwable throwable) {
         try {