* Fix typos in docs and Javadocs for SI SNS

* Simplify `try..catch` logic in the `SnsMessageHandler` and add missed `currentThread().interrupt()`
* Add `.fifo` logic to the `SnsAsyncTopicArnResolver`

Author: artembilan

docs/src/main/asciidoc/sns.adoc

@@ -331,7 +331,7 @@ public static class MyConfiguration {
 
     @Bean
     public HttpRequestHandler sqsMessageDrivenChannelAdapter(PollableChannel inputChannel) {
-    	SnsInboundChannelAdapter adapter = new SnsInboundChannelAdapter(amazonSns(), "/mySampleTopic");
+    	SnsInboundChannelAdapter adapter = new SnsInboundChannelAdapter(this.amazonSns, "/mySampleTopic");
     	adapter.setRequestChannel(inputChannel);
     	adapter.setHandleNotificationStatus(true);
     	return adapter;
@@ -360,8 +360,8 @@ The Java Config looks like:
 [source,java]
 ----
 @Bean
-public MessageHandler snsMessageHandler() {
-    SnsMessageHandler handler = new SnsMessageHandler(amazonSns());
+public MessageHandler snsMessageHandler(SnsAsyncClient amazonSns) {
+    SnsMessageHandler handler = new SnsMessageHandler(amazonSns);
     handler.setTopicArn("arn:aws:sns:eu-west:123456789012:test");
     String bodyExpression = "T(SnsBodyBuilder).withDefault(payload).forProtocols(payload.substring(0, 140), 'sms')";
     handler.setBodyExpression(spelExpressionParser.parseExpression(bodyExpression));

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/SnsAsyncTopicArnResolver.java

@@ -15,9 +15,11 @@
  */
 package io.awspring.cloud.sns.core;
 
+import java.util.Map;
 import org.springframework.util.Assert;
 import software.amazon.awssdk.arns.Arn;
 import software.amazon.awssdk.services.sns.SnsAsyncClient;
+import software.amazon.awssdk.services.sns.model.CreateTopicRequest;
 
 /**
  * A {@link TopicArnResolver} implementation to determine topic ARN by name against an {@link SnsAsyncClient}.
@@ -27,6 +29,7 @@
  * @since 4.0
  */
 public class SnsAsyncTopicArnResolver implements TopicArnResolver {
+
 	private final SnsAsyncClient snsClient;
 
 	public SnsAsyncTopicArnResolver(SnsAsyncClient snsClient) {
@@ -36,7 +39,7 @@ public SnsAsyncTopicArnResolver(SnsAsyncClient snsClient) {
 
 	/**
 	 * Resolve topic ARN by topic name. If topicName is already an ARN, it returns {@link Arn}. If topicName is just a
-	 * string with a topic name, it attempts to create a topic or if the topic already exists, just returns its ARN.
+	 * string with a topic name, it attempts to create a topic, or if the topic already exists, just returns its ARN.
 	 */
 	@Override
 	public Arn resolveTopicArn(String topicName) {
@@ -45,8 +48,15 @@ public Arn resolveTopicArn(String topicName) {
 			return Arn.fromString(topicName);
 		}
 		else {
+			CreateTopicRequest.Builder builder = CreateTopicRequest.builder().name(topicName);
+
+			// fix for https://github.com/awspring/spring-cloud-aws/issues/707
+			if (topicName.endsWith(".fifo")) {
+				builder.attributes(Map.of("FifoTopic", "true"));
+			}
+
 			// if the topic exists, createTopic returns a successful response with the topic arn
-			return Arn.fromString(this.snsClient.createTopic(request -> request.name(topicName)).join().topicArn());
+			return Arn.fromString(this.snsClient.createTopic(builder.build()).join().topicArn());
 		}
 	}
 

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/integration/SnsHeaderMapper.java

@@ -40,7 +40,7 @@
  *
  * @author Artem Bilan
  *
- * @since 2.0
+ * @since 4.0
  */
 public class SnsHeaderMapper implements HeaderMapper<Map<String, MessageAttributeValue>> {
 
@@ -55,7 +55,7 @@ public class SnsHeaderMapper implements HeaderMapper<Map<String, MessageAttribut
 	 * Spring Integration Message's headers. The values can also contain simple wildcard patterns (e.g. "foo*" or
 	 * "*foo") to be matched. Also supports negated ('!') patterns. First match wins (positive or negative). To match
 	 * the names starting with {@code !} symbol, you have to escape it prepending with the {@code \} symbol in the
-	 * pattern definition. Defaults to map all ({@code *}) if the type is supported by SQS. The
+	 * pattern definition. Defaults to map all ({@code *}) if the type is supported by SNS. The
 	 * {@link MessageHeaders#ID}, {@link MessageHeaders#TIMESTAMP}, {@link NativeMessageHeaderAccessor#NATIVE_HEADERS},
 	 * {@link SnsHeaders#MESSAGE_ID_HEADER and {@link SnsHeaders#TOPIC_HEADER} are ignored by default.
 	 * @param outboundHeaderNames The inbound header names.
@@ -101,22 +101,23 @@ else if (messageHeaderValue instanceof byte[] bytes) {
 		}
 	}
 
-	private MessageAttributeValue getBinaryMessageAttribute(ByteBuffer messageHeaderValue) {
+	private static MessageAttributeValue getBinaryMessageAttribute(ByteBuffer messageHeaderValue) {
 		return buildMessageAttribute("Binary", messageHeaderValue);
 	}
 
-	private MessageAttributeValue getStringMessageAttribute(String messageHeaderValue) {
+	private static MessageAttributeValue getStringMessageAttribute(String messageHeaderValue) {
 		return buildMessageAttribute("String", messageHeaderValue);
 	}
 
-	private MessageAttributeValue getNumberMessageAttribute(Object messageHeaderValue) {
-		Assert.isTrue(NumberUtils.STANDARD_NUMBER_TYPES.contains(messageHeaderValue.getClass()),
+	private static MessageAttributeValue getNumberMessageAttribute(Object messageHeaderValue) {
+		Class<?> messageHeaderValueClass = messageHeaderValue.getClass();
+		Assert.isTrue(NumberUtils.STANDARD_NUMBER_TYPES.contains(messageHeaderValueClass),
 				"Only standard number types are accepted as message header.");
 
-		return buildMessageAttribute("Number." + messageHeaderValue.getClass().getName(), messageHeaderValue);
+		return buildMessageAttribute("Number." + messageHeaderValueClass.getName(), messageHeaderValue);
 	}
 
-	private MessageAttributeValue buildMessageAttribute(String dataType, Object value) {
+	private static MessageAttributeValue buildMessageAttribute(String dataType, Object value) {
 		MessageAttributeValue.Builder messageAttributeValue = MessageAttributeValue.builder().dataType(dataType);
 		if (value instanceof ByteBuffer byteBuffer) {
 			messageAttributeValue.binaryValue(SdkBytes.fromByteBuffer(byteBuffer));

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/integration/SnsInboundChannelAdapter.java

@@ -94,8 +94,8 @@ public SnsInboundChannelAdapter(SnsClient amazonSns, String... path) {
 	}
 
 	/**
-	 * The flag indicating if the adapter should send {@code SubscriptionConfirmation/UnsubscribeConfirmation}
-	 * message to the `output-channel` or not.
+	 * The flag indicating if the adapter should send {@code SubscriptionConfirmation/UnsubscribeConfirmation} message
+	 * to the `output-channel` or not.
 	 * @param handleNotificationStatus the flag to set. Default is {@code false}.
 	 */
 	public void setHandleNotificationStatus(boolean handleNotificationStatus) {

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/integration/SnsMessageHandler.java

@@ -57,14 +57,14 @@
  * from there and the {@code messageStructure} of the {@link PublishRequest} is set to {@code json}. For the convenience
  * the package {@code org.springframework.integration.aws.support} is imported to the {@link #evaluationContext} to
  * allow bypassing it for the {@link SnsBodyBuilder} from the {@link #bodyExpression} definition. For example:
- * 
+ *
  * <pre class="code">
  * {@code
  * String bodyExpression = "SnsBodyBuilder.withDefault(payload).forProtocols(payload.substring(0, 140), 'sms')";
  * snsMessageHandler.setBodyExpression(spelExpressionParser.parseExpression(bodyExpression));
  * }
  * </pre>
- * 
+ *
  * </li>
  * <li>Otherwise the {@code payload} (or the {@link #bodyExpression} evaluation result) is converted to the
  * {@link String} using {@link #getConversionService()}.</li>
@@ -245,24 +245,23 @@ protected void handleMessageInternal(Message<?> message) {
 		}
 
 		Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class);
-		if (sendTimeout == null || sendTimeout < 0) {
-			try {
+		try {
+			if (sendTimeout == null || sendTimeout < 0) {
 				resultFuture.get();
 			}
-			catch (InterruptedException | ExecutionException ex) {
-				throw new IllegalStateException(ex);
-			}
-		}
-		else {
-			try {
+			else {
 				resultFuture.get(sendTimeout, TimeUnit.MILLISECONDS);
 			}
-			catch (TimeoutException te) {
-				throw new MessageTimeoutException(message, "Timeout waiting for response from AmazonKinesis", te);
-			}
-			catch (InterruptedException | ExecutionException ex) {
-				throw new IllegalStateException(ex);
-			}
+		}
+		catch (TimeoutException te) {
+			throw new MessageTimeoutException(message, "Timeout waiting for response from AmazonKinesis", te);
+		}
+		catch (InterruptedException ex) {
+			Thread.currentThread().interrupt();
+			throw new IllegalStateException(ex);
+		}
+		catch (ExecutionException ex) {
+			throw new IllegalStateException(ex);
 		}
 	}
 

spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/integration/SnsMessageHandlerTests.java

@@ -24,7 +24,6 @@
 import io.awspring.cloud.sns.core.SnsHeaders;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -43,6 +42,7 @@
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import software.amazon.awssdk.services.sns.SnsAsyncClient;
+import software.amazon.awssdk.services.sns.model.CreateTopicRequest;
 import software.amazon.awssdk.services.sns.model.CreateTopicResponse;
 import software.amazon.awssdk.services.sns.model.MessageAttributeValue;
 import software.amazon.awssdk.services.sns.model.PublishRequest;
@@ -110,13 +110,12 @@ void snsMessageHandler() {
 	public static class ContextConfiguration {
 
 		@Bean
-		@SuppressWarnings("unchecked")
 		public SnsAsyncClient amazonSNS() {
 			SnsAsyncClient mock = mock();
 
 			willAnswer(invocation -> CompletableFuture.completedFuture(
 					CreateTopicResponse.builder().topicArn("arn:aws:sns:eu-west-1:111111111111:topic.fifo").build()))
-					.given(mock).createTopic(any(Consumer.class));
+					.given(mock).createTopic(any(CreateTopicRequest.class));
 
 			willAnswer(
 					invocation -> CompletableFuture.completedFuture(PublishResponse.builder().messageId("111").build()))