Fix duplicate error logging when processing exception thrown in SQS listener.

This commit fixes this issue by delegating exception processing to
AbstractMethodMessageHandler only when there is a configured handler.
If there is none, exception is logged.

Fixes gh-394
Closes gh-465

Co-authored-by: Maciej Walkowiak <[email protected]>

Author: sayembd

spring-cloud-aws-messaging/src/main/java/org/springframework/cloud/aws/messaging/listener/QueueMessageHandler.java

@@ -52,6 +52,7 @@
 import org.springframework.messaging.handler.invocation.AbstractMethodMessageHandler;
 import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
 import org.springframework.messaging.handler.invocation.HandlerMethodReturnValueHandler;
+import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.comparator.ComparableComparator;
 import org.springframework.validation.Errors;
@@ -233,7 +234,15 @@ protected void handleNoMatch(Set<MappingInformation> ts, String lookupDestinatio
 	@Override
 	protected void processHandlerMethodException(HandlerMethod handlerMethod,
 			Exception ex, Message<?> message) {
-		super.processHandlerMethodException(handlerMethod, ex, message);
+		InvocableHandlerMethod exceptionHandlerMethod = getExceptionHandlerMethod(
+				handlerMethod, ex);
+		if (exceptionHandlerMethod != null) {
+			super.processHandlerMethodException(handlerMethod, ex, message);
+		}
+		else {
+			this.logger.error("An exception occurred while invoking the handler method",
+					ex);
+		}
 		throw new MessagingException(
 				"An exception occurred while invoking the handler method", ex);
 	}

spring-cloud-aws-messaging/src/main/java/org/springframework/cloud/aws/messaging/listener/SimpleMessageListenerContainer.java

@@ -418,7 +418,7 @@ public void run() {
 				applyDeletionPolicyOnSuccess(receiptHandle);
 			}
 			catch (MessagingException messagingException) {
-				applyDeletionPolicyOnError(receiptHandle, messagingException);
+				applyDeletionPolicyOnError(receiptHandle);
 			}
 		}
 
@@ -430,17 +430,12 @@ private void applyDeletionPolicyOnSuccess(String receiptHandle) {
 			}
 		}
 
-		private void applyDeletionPolicyOnError(String receiptHandle,
-				MessagingException messagingException) {
+		private void applyDeletionPolicyOnError(String receiptHandle) {
 			if (this.deletionPolicy == SqsMessageDeletionPolicy.ALWAYS
 					|| (this.deletionPolicy == SqsMessageDeletionPolicy.NO_REDRIVE
 							&& !this.hasRedrivePolicy)) {
 				deleteMessage(receiptHandle);
 			}
-			else if (this.deletionPolicy == SqsMessageDeletionPolicy.ON_SUCCESS) {
-				getLogger().error("Exception encountered while processing message.",
-						messagingException);
-			}
 		}
 
 		private void deleteMessage(String receiptHandle) {

spring-cloud-aws-messaging/src/test/java/org/springframework/cloud/aws/messaging/listener/QueueMessageHandlerTest.java

@@ -70,6 +70,7 @@
 import org.springframework.messaging.support.MessageBuilder;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -566,6 +567,40 @@ public void getMappingForMethod_methodWithExpressionProducingMultipleQueueNames_
 				.containsAll(Arrays.asList("queueOne", "queueTwo"))).isTrue();
 	}
 
+	@Test
+	public void processHandlerMethodException_invocableHandlerMethodNotAvailable_errorMustBeLogged() {
+		// Arrange
+		StaticApplicationContext applicationContext = new StaticApplicationContext();
+		applicationContext.registerSingleton("sqsListenerWithoutMessageExceptionHandler",
+				SqsListenerWithoutMessageExceptionHandler.class);
+		applicationContext.registerBeanDefinition("queueMessageHandler",
+				getQueueMessageHandlerBeanDefinition());
+		applicationContext.refresh();
+		MessageHandler messageHandler = applicationContext.getBean(MessageHandler.class);
+
+		LoggerContext logContext = (LoggerContext) LoggerFactory.getILoggerFactory();
+		ListAppender<ILoggingEvent> appender = new ListAppender<>();
+		appender.start();
+		Logger log = logContext.getLogger(QueueMessageHandler.class);
+		log.setLevel(Level.ERROR);
+		log.addAppender(appender);
+		appender.setContext(log.getLoggerContext());
+
+		// Act
+		try {
+			messageHandler.handleMessage(MessageBuilder.withPayload("testContent")
+					.setHeader(QueueMessageHandler.LOGICAL_RESOURCE_ID, "receive")
+					.build());
+			fail();
+		}
+		catch (MessagingException e) {
+			// ignore
+		}
+
+		// Assert
+		assertThat(appender.list).hasSize(1);
+	}
+
 	@SuppressWarnings("UnusedDeclaration")
 	private static class IncomingMessageHandler {
 
@@ -604,6 +639,15 @@ private String getLastReceivedMessage() {
 
 	}
 
+	private static class SqsListenerWithoutMessageExceptionHandler {
+
+		@SqsListener("receive")
+		public String receive(String value) {
+			throw new RuntimeException("test exception");
+		}
+
+	}
+
 	private static class IncomingMessageHandlerWithMultipleQueueNames {
 
 		private String lastReceivedMessage;