SQSListener not consuming messages from queue

Viewed 946

I cannot see the messages in the SQS queue being consumed by the @SqsListener

import org.springframework.cloud.aws.messaging.listener.annotation.SqsListener; //others

@Component
public class Consumer{
  private static final Logger logger = LoggerFactory.getLogger(Consumer.class);

  @SqsListener(value = "TEST-MY-QUEUE")
  public void receiveMessage(String stringJson)  {
    System.out.println("***Consuming message: " + stringJson);
    logger.info("Consuming message: " + stringJson);
  }

}

My configuration (Here I print the client queues, and I can deffo spot the queue I want to consume - TEST-MY-QUEUE . It prints the URL correctly in the region. I am also able to see the region loaded correctly (same as queue) in regionProvider

@Configuration

public class AwsConfiguration { 

  @Bean
  @Primary
  AmazonSQSAsync sqsClient() {
    AmazonSQSAsync amazonSQSAsync = AmazonSQSAsyncClientBuilder.defaultClient();
    System.out.println("Client queues = " + amazonSQSAsync.listQueues()); //The queue I want to consume is here
    return amazonSQSAsync;
  }

  @Bean
  AwsRegionProvider regionProvider() {
    DefaultAwsRegionProviderChain defaultAwsRegionProviderChain = new DefaultAwsRegionProviderChain();
    System.out.println("Region = " + defaultAwsRegionProviderChain.getRegion());
    return defaultAwsRegionProviderChain;
  }

@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer(AmazonSQSAsync amazonSQSAsync, QueueMessageHandler queueMessageHandler) {
    
    SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer();
    simpleMessageListenerContainer.setAmazonSqs(amazonSQSAsync);
    simpleMessageListenerContainer.setMessageHandler(queueMessageHandler);
    simpleMessageListenerContainer.setMaxNumberOfMessages(10);
    simpleMessageListenerContainer.setTaskExecutor(threadPoolTaskExecutor());
    return simpleMessageListenerContainer;
  }

  @Bean
  public QueueMessageHandler queueMessageHandler(AmazonSQSAsync amazonSQSAsync) {
    QueueMessageHandlerFactory queueMessageHandlerFactory = new QueueMessageHandlerFactory();
    queueMessageHandlerFactory.setAmazonSqs(amazonSQSAsync);
    QueueMessageHandler queueMessageHandler = queueMessageHandlerFactory.createQueueMessageHandler();
    return queueMessageHandler;
  }

  public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(10);
    executor.initialize();
    return executor;
  }

And pom.xml (Java 11, spring boot, spring cloud aws)


        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.5.6</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-aws-core</artifactId>
            <version>2.2.6.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-aws-autoconfigure</artifactId>
            <version>2.2.6.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-bootstrap</artifactId>
            <version>3.0.3</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-aws</artifactId>
            <version>2.2.6.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-aws-messaging</artifactId>
            <version>2.2.6.RELEASE</version>
        </dependency>

I noticed very similar issues in the questions here and I changed my dependencies in pom.xml to be spring-cloud-starter-aws-messaging but didnt fix for me. I double checked the names (queue, annotation) all seems fine When I run my app, starts fine but I dont see any logs or exception. Not one message consumed. What am I missing?

Thank you

2 Answers

You are using a third party API. To use invoke Amazon Simple Queue Service (SQS) from a Java project, use the Official AWS SDK for Java V2. If you are not aware how to use this SDK, see this DEV Guide:

Developer guide - AWS SDK for Java 2.x

For AWS SQS specific information, see:

Working with Amazon Simple Queue Service

This has links to AWS Github where you will find POM dependencies, code, etc.

At the end it was an issue with the config (using the credentials)

In application.yml

credentials:
  useDefaultAwsCredentialsChain: true #Will use credentials in /.aws 

And then in the AWSConfig class where you create the AmazonSQSAsync, just make it use that config

public AmazonSQSAsync amazonSQSAsync() {
  DefaultAWSCredentialsProviderChain defaultAWSCredentialsProviderChain = new DefaultAWSCredentialsProviderChain();
  return AmazonSQSAsyncClientBuilder.standard().withRegion(region)
            .withCredentials(defaultAWSCredentialsProviderChain)
            .build();
Related