diff --git a/docs/src/main/asciidoc/sqs.adoc b/docs/src/main/asciidoc/sqs.adoc index d9d91c011..451a061e4 100644 --- a/docs/src/main/asciidoc/sqs.adoc +++ b/docs/src/main/asciidoc/sqs.adoc @@ -255,6 +255,7 @@ See <> 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 listener for the missing queue is skipped at startup with a warning log and subsequent polls for it return no messages, so application startup is not blocked. |`queueAttributeNames` |Collection diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolver.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolver.java index 5bf87d805..a97ecd667 100644 --- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolver.java +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolver.java @@ -86,16 +86,22 @@ public CompletableFuture resolveQueueAttributes() { } private CompletableFuture 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 resolveQueueUrl() { @@ -120,10 +126,10 @@ private CompletableFuture doResolveQueueUrl() { } private CompletableFuture 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 createQueue() { diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolvingException.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolvingException.java index b45a4a587..c09ea4358 100644 --- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolvingException.java +++ b/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,36 @@ */ 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. + */ + 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. + */ + public boolean isQueueIgnored() { + return this.queueIgnored; } } diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/QueueNotFoundStrategy.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/QueueNotFoundStrategy.java index 8e99d8d9a..b830ae155 100644 --- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/QueueNotFoundStrategy.java +++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/QueueNotFoundStrategy.java @@ -32,6 +32,15 @@ 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 listener for a queue that is not found, log a warning, and allow application startup to + * proceed. Subsequent polls for that listener return no messages. 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 } diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/source/AbstractPollingMessageSource.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/source/AbstractPollingMessageSource.java index 91b714120..c3c948f5d 100644 --- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/source/AbstractPollingMessageSource.java +++ b/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(); diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/QueueAttributesResolverIntegrationTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/QueueAttributesResolverIntegrationTests.java index 41f43fbad..1511607a2 100644 --- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/QueueAttributesResolverIntegrationTests.java +++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/QueueAttributesResolverIntegrationTests.java @@ -113,6 +113,26 @@ void shouldNotCreateQueue() { .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";