GH-1143 Add IGNORE QueueNotFoundStrategy option for SQS

docs/src/main/asciidoc/sqs.adoc

@@ -255,6 +255,7 @@ See <<template-message-conversion>> for more information.
 #CREATE
 |Set the strategy to use in case a queue is not found.
 With `QueueNotFoundStrategy#FAIL`, an exception is thrown in case a queue is not found.
+With `QueueNotFoundStrategy#IGNORE`, the message source for the missing queue is skipped at startup with a warning log and no polling thread is created for that queue, so application startup is not blocked. The queue is resolved again on container restart and the listener starts normally if the queue has been created.
 
 |`queueAttributeNames`
 |Collection<AttributeNames>

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolver.java

@@ -86,16 +86,22 @@ public CompletableFuture<QueueAttributes> resolveQueueAttributes() {
 	}
 
 	private CompletableFuture<QueueAttributes> wrapException(Throwable t) {
+		Throwable unwrapped = t instanceof CompletionException ? t.getCause() : t;
+
+		if (unwrapped instanceof QueueDoesNotExistException && QueueNotFoundStrategy.IGNORE.equals(this.queueNotFoundStrategy)) {
+			return CompletableFutures.failedFuture(new QueueAttributesResolvingException(
+					"Queue not found: " + this.queueName, unwrapped, true));
+		}
+
 		String message = "Error resolving attributes for queue "
 			+ this.queueName + " with strategy " + this.queueNotFoundStrategy + " and queueAttributesNames " + this.queueAttributeNames;
 
-		if (t.getCause() instanceof SqsException) {
+		if (unwrapped instanceof SqsException) {
 			message += "\n This might be due to connectivity issues or incorrect configuration. " +
 				"Please verify your AWS credentials, network settings, and queue configuration.";
 		}
 
-		return CompletableFutures.failedFuture(new QueueAttributesResolvingException(message,
-			t instanceof CompletionException ? t.getCause() : t));
+		return CompletableFutures.failedFuture(new QueueAttributesResolvingException(message, unwrapped));
 	}
 
 	private CompletableFuture<String> resolveQueueUrl() {
@@ -120,10 +126,10 @@ private CompletableFuture<String> doResolveQueueUrl() {
 	}
 
 	private CompletableFuture<String> handleException(Throwable t) {
-		return t.getCause() instanceof QueueDoesNotExistException
-			&& QueueNotFoundStrategy.CREATE.equals(this.queueNotFoundStrategy)
-				? createQueue()
-				: CompletableFutures.failedFuture(t);
+		if (t.getCause() instanceof QueueDoesNotExistException && QueueNotFoundStrategy.CREATE.equals(this.queueNotFoundStrategy)) {
+			return createQueue();
+		}
+		return CompletableFutures.failedFuture(t);
 	}
 
 	private CompletableFuture<String> createQueue() {

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolvingException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2022 the original author or authors.
+ * Copyright 2013-2026 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,12 +24,38 @@
  */
 public class QueueAttributesResolvingException extends RuntimeException {
 
+	private final boolean queueIgnored;
+
 	/**
 	 * Create an instance with the message and throwable cause.
 	 * @param message the error message.
 	 * @param cause the cause.
 	 */
 	public QueueAttributesResolvingException(String message, Throwable cause) {
+		this(message, cause, false);
+	}
+
+	/**
+	 * Create an instance with the message, cause, and a flag indicating that the resolver treated the missing queue as
+	 * ignored per {@link io.awspring.cloud.sqs.listener.QueueNotFoundStrategy#IGNORE}.
+	 * @param message the error message.
+	 * @param cause the cause.
+	 * @param queueIgnored whether the resolver signalled that the queue should be ignored.
+	 * @since 4.2
+	 */
+	public QueueAttributesResolvingException(String message, Throwable cause, boolean queueIgnored) {
 		super(message, cause);
+		this.queueIgnored = queueIgnored;
+	}
+
+	/**
+	 * Whether the resolver signalled that the missing queue should be ignored under
+	 * {@link io.awspring.cloud.sqs.listener.QueueNotFoundStrategy#IGNORE}, so the listener can skip startup rather than
+	 * fail the application context.
+	 * @return {@code true} if the queue should be ignored.
+	 * @since 4.2
+	 */
+	public boolean isQueueIgnored() {
+		return this.queueIgnored;
 	}
 }

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/QueueNotFoundStrategy.java

@@ -32,6 +32,16 @@ public enum QueueNotFoundStrategy {
 	 * Create queues that are not found at startup. Mind that in production environments the application might not have
 	 * permissions to create the queue and throw an exception.
 	 */
-	CREATE
+	CREATE,
+
+	/**
+	 * Skip starting the message source for a queue that is not found, log a warning, and allow application startup to
+	 * proceed. Since the source is not started, no polling thread is created for that queue. The queue will be resolved
+	 * again on the next container restart, which allows the listener to start normally if the queue has been created.
+	 * Useful when a queue may legitimately be absent in some deployments (for example, optional feature queues) and
+	 * neither {@link #CREATE} nor {@link #FAIL} fits. This mirrors the default behavior of Spring Cloud AWS 2.x's
+	 * {@code spring-cloud-starter-aws-messaging}, which silently ignored missing queues at startup.
+	 */
+	IGNORE
 
 }

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/BatchingAcknowledgementProcessor.java

@@ -80,9 +80,9 @@ public class BatchingAcknowledgementProcessor<T> extends AbstractOrderingAcknowl
 	private BlockingQueue<Message<T>> acks;
 
 	/**
-	 * Number of messages that have been received for acknowledgement but are not in {@link #acks} nor in the buffer yet.
-	 * Incremented before a message is offered to the queue and decremented after it has been added to the buffer, so a
-	 * message is always accounted for while the polling thread is holding it in between the two.
+	 * Number of messages that have been received for acknowledgement but are not in {@link #acks} nor in the buffer
+	 * yet. Incremented before a message is offered to the queue and decremented after it has been added to the buffer,
+	 * so a message is always accounted for while the polling thread is holding it in between the two.
 	 */
 	private final AtomicInteger unbufferedAcks = new AtomicInteger();
 

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/source/AbstractPollingMessageSource.java

@@ -16,6 +16,7 @@
 package io.awspring.cloud.sqs.listener.source;
 
 import io.awspring.cloud.sqs.ConfigUtils;
+import io.awspring.cloud.sqs.QueueAttributesResolvingException;
 import io.awspring.cloud.sqs.listener.ContainerOptions;
 import io.awspring.cloud.sqs.listener.IdentifiableContainerComponent;
 import io.awspring.cloud.sqs.listener.MessageProcessingContext;
@@ -185,7 +186,17 @@ public void start() {
 							eap -> eap.setAcknowledgementResultCallback(this.acknowledgementResultCallback))
 					.acceptIfInstance(this.acknowledgmentProcessor, TaskExecutorAware.class,
 							ea -> ea.setTaskExecutor(this.taskExecutor));
-			doStart();
+			try {
+				doStart();
+			}
+			catch (CompletionException ce) {
+				if (ce.getCause() instanceof QueueAttributesResolvingException qare && qare.isQueueIgnored()) {
+					logger.warn("Skipping start for queue {}: {}", this.pollingEndpointName, qare.getMessage());
+					this.running = false;
+					return;
+				}
+				throw ce;
+			}
 			setupAcknowledgementForConversion(this.acknowledgmentProcessor.getAcknowledgementCallback());
 			this.acknowledgmentProcessor.start();
 			startPollingThread();

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/QueueAttributesResolverIntegrationTests.java

@@ -108,11 +108,32 @@ void shouldNotCreateQueue() {
 		assertThatThrownBy(() -> resolver.resolveQueueAttributes().join())
 			.isInstanceOf(CompletionException.class)
 			.extracting(Throwable::getCause)
-			.isInstanceOf(QueueAttributesResolvingException.class)
+			.isInstanceOfSatisfying(QueueAttributesResolvingException.class,
+				qare -> assertThat(qare.isQueueIgnored()).isFalse())
 			.extracting(Throwable::getCause)
 			.isInstanceOf(QueueDoesNotExistException.class);
 	}
 
+	@Test
+	void shouldIgnoreQueueWhenStrategyIsIgnore() {
+		String queueName = "testQueueName-" + UUID.randomUUID();
+		SqsAsyncClient client = createAsyncClient();
+		QueueAttributesResolver resolver = QueueAttributesResolver
+			.builder()
+			.queueAttributeNames(Collections.emptyList())
+			.sqsAsyncClient(client)
+			.queueName(queueName)
+			.queueNotFoundStrategy(QueueNotFoundStrategy.IGNORE)
+			.build();
+		assertThatThrownBy(() -> resolver.resolveQueueAttributes().join())
+			.isInstanceOf(CompletionException.class)
+			.cause()
+			.isInstanceOfSatisfying(QueueAttributesResolvingException.class,
+				qare -> assertThat(qare.isQueueIgnored()).isTrue())
+			.cause()
+			.isInstanceOf(QueueDoesNotExistException.class);
+	}
+
 	@Test
 	void shouldGetQueueAttributes() {
 		String queueName = "should-get-queue-attributes";

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsQueueNotFoundStrategyIntegrationTests.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright 2013-2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.awspring.cloud.sqs.integration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import io.awspring.cloud.sqs.QueueAttributesResolvingException;
+import io.awspring.cloud.sqs.listener.QueueNotFoundStrategy;
+import io.awspring.cloud.sqs.listener.SqsMessageListenerContainer;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
+import java.time.Duration;
+import java.util.UUID;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
+import software.amazon.awssdk.services.sqs.model.QueueDoesNotExistException;
+
+/**
+ * Integration tests for {@link QueueNotFoundStrategy}.
+ *
+ * @author Bill Kim
+ */
+class SqsQueueNotFoundStrategyIntegrationTests extends BaseSqsIntegrationTest {
+
+	@Test
+	void shouldStartAndStopWithoutPollingWhenIgnoredQueueIsMissing() throws Exception {
+		SqsAsyncClient client = createAsyncClient();
+		SqsTemplate template = SqsTemplate.newTemplate(client);
+		String missingQueueName = uniqueQueueName();
+		CountDownLatch messagesReceived = new CountDownLatch(1);
+		SqsMessageListenerContainer<String> container = createContainer(client, QueueNotFoundStrategy.IGNORE,
+				messagesReceived, missingQueueName);
+
+		assertDoesNotThrow(container::start);
+		assertThat(container.isRunning()).isTrue();
+
+		createQueue(client, missingQueueName).join();
+		template.send(missingQueueName, "should-not-be-consumed-before-restart");
+
+		assertThat(messagesReceived.await(2, TimeUnit.SECONDS)).isFalse();
+
+		assertDoesNotThrow(() -> container.stop());
+		assertThat(container.isRunning()).isFalse();
+	}
+
+	@Test
+	void shouldPollAfterIgnoredQueueIsCreatedAndContainerIsRestarted() throws Exception {
+		SqsAsyncClient client = createAsyncClient();
+		SqsTemplate template = SqsTemplate.newTemplate(client);
+		String missingQueueName = uniqueQueueName();
+		CountDownLatch messagesReceived = new CountDownLatch(1);
+		SqsMessageListenerContainer<String> container = createContainer(client, QueueNotFoundStrategy.IGNORE,
+				messagesReceived, missingQueueName);
+
+		container.start();
+		createQueue(client, missingQueueName).join();
+		template.send(missingQueueName, "should-be-consumed-after-restart");
+
+		assertThat(messagesReceived.await(2, TimeUnit.SECONDS)).isFalse();
+
+		container.stop();
+		container.start();
+
+		assertThat(messagesReceived.await(10, TimeUnit.SECONDS)).isTrue();
+
+		container.stop();
+	}
+
+	@Test
+	void shouldConsumeFromExistingQueueWhenAnotherQueueIsIgnored() throws Exception {
+		SqsAsyncClient client = createAsyncClient();
+		SqsTemplate template = SqsTemplate.newTemplate(client);
+		String existingQueueName = uniqueQueueName();
+		String missingQueueName = uniqueQueueName();
+		CountDownLatch messagesReceived = new CountDownLatch(1);
+		SqsMessageListenerContainer<String> container = createContainer(client, QueueNotFoundStrategy.IGNORE,
+				messagesReceived, missingQueueName, existingQueueName);
+
+		createQueue(client, existingQueueName).join();
+		container.start();
+		template.send(existingQueueName, "should-be-consumed-from-existing-queue");
+
+		assertThat(messagesReceived.await(10, TimeUnit.SECONDS)).isTrue();
+
+		container.stop();
+	}
+
+	@Test
+	void shouldThrowWhenQueueIsMissingAndStrategyIsFail() {
+		SqsAsyncClient client = createAsyncClient();
+		CountDownLatch messagesReceived = new CountDownLatch(1);
+		SqsMessageListenerContainer<String> container = createContainer(client, QueueNotFoundStrategy.FAIL,
+				messagesReceived, uniqueQueueName());
+
+		assertThatThrownBy(container::start).isInstanceOf(CompletionException.class).cause()
+				.isInstanceOfSatisfying(QueueAttributesResolvingException.class,
+						qare -> assertThat(qare.isQueueIgnored()).isFalse())
+				.cause().isInstanceOf(QueueDoesNotExistException.class);
+
+		container.stop();
+	}
+
+	private SqsMessageListenerContainer<String> createContainer(SqsAsyncClient client,
+			QueueNotFoundStrategy queueNotFoundStrategy, CountDownLatch messagesReceived, String... queueNames) {
+		return SqsMessageListenerContainer.<String> builder().sqsAsyncClient(client).queueNames(queueNames)
+				.configure(options -> options.queueNotFoundStrategy(queueNotFoundStrategy)
+						.pollTimeout(Duration.ofMillis(200)).maxMessagesPerPoll(1).maxConcurrentMessages(1))
+				.messageListener(message -> messagesReceived.countDown()).build();
+	}
+
+	private String uniqueQueueName() {
+		return "queue-not-found-strategy-" + UUID.randomUUID();
+	}
+
+}