Spring-cloud-aws: FIFO support in v1.2.2.RC1 gives an error when sending message to FIFO queue

[jeffczaplewski]

Newbie logging an issue so apologies in advance.

we're trying to send a message to a FIFO queue using the v.1.2.2.RC1 release and whenever we send a message to the queue we get an error that the MessageGroupId is not present. When digging in a little bit, it seems that in the SqsClientConfiguration class, an AmazonSQSBufferedAsyncClient is returned and used when sending the message to the FIFO queue (if no other bean defined?). Currently there is no support in AWS for sending messages to a FIFO queue using an AmazonSQSBufferedAsyncClient.

So is the correct way to send to a FIFO queue then with v1.2.2.RC1 to define your own bean like this:
@Bean(destroyMethod = "shutdown") public AmazonSQSAsyncClient amazonSQS() throws Exception { AmazonWebserviceClientFactoryBean<AmazonSQSAsyncClient> clientFactoryBean = new AmazonWebserviceClientFactoryBean<>(AmazonSQSAsyncClient.class, this.awsCredentialsProvider, this.regionProvider); clientFactoryBean.afterPropertiesSet(); return clientFactoryBean.getObject(); }

The sample method below should throw an error (i copied it from a controller test method i implemented in our app to make sure it was using the same AWS SDK as spring cloud aws messaging)

`
BasicAWSCredentials awsCreds = new BasicAWSCredentials("", "");

    AmazonSQSAsync sqs = AmazonSQSAsyncClientBuilder.standard().withCredentials(new AWSStaticCredentialsProvider(awsCreds))
            .withRegion(Regions.US_EAST_1).build();

    AmazonSQSBufferedAsyncClient asyncClient = new AmazonSQSBufferedAsyncClient(sqs);

    ObjectMapper mapper = new ObjectMapper();
    String value = mapper.writeValueAsString(transactionDTO);

    SendMessageRequest sendMessageRequest = new SendMessageRequest("https://sqs.us-east-1.amazonaws.com/stuff/stuff.fifo", value);

    sendMessageRequest.setMessageGroupId("theMessageGroup");

    Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();

    MessageAttributeValue contentType = new MessageAttributeValue().withDataType("String").withStringValue(MediaType.APPLICATION_JSON_VALUE);
    MessageAttributeValue id = new MessageAttributeValue().withDataType("String").withStringValue(UUID.randomUUID().toString());
    MessageAttributeValue timestamp = new MessageAttributeValue().withDataType("String").withStringValue(new Date().getTime()+"");

    messageAttributes.put("contentType", contentType);
    messageAttributes.put("id", id);
    messageAttributes.put("timestamp", timestamp);

    sendMessageRequest.setMessageAttributes(messageAttributes);

    SendMessageResult result = asyncClient.sendMessage(sendMessageRequest);

    result.getMessageId();

`

If you change the above and remove the AmazonSQSBufferedAsyncClient asyncClient = new AmazonSQSBufferedAsyncClient(sqs); and just use the AmazonSQSAsync object to send the message, it all works fine.

if you trace the org.apache.http.wire output, you'll see that the serialized version using the buffered async client is missing the message group id

output without the buffered async client
{ "Action": [ "SendMessage" ], "Version": [ "2012-11-05" ], "MessageBody": [ "{"removed":"removed"}" ], "MessageAttribute.1.Name": [ "id" ], "MessageAttribute.1.Value.StringValue": [ "1510118384513" ], "MessageAttribute.1.Value.DataType": [ "String" ], "MessageAttribute.2.Name": [ "contentType" ], "MessageAttribute.2.Value.StringValue": [ "application/json" ], "MessageAttribute.2.Value.DataType": [ "String" ], "MessageAttribute.3.Name": [ "timestamp" ], "MessageAttribute.3.Value.StringValue": [ "1510118384513" ], "MessageAttribute.3.Value.DataType": [ "String" ], "MessageGroupId": [ "theMessageGroup" ] }

output with the buffered async client

{ "Action": [ "SendMessageBatch" ], "Version": [ "2012-11-05" ], "SendMessageBatchRequestEntry.1.Id": [ "0" ], "SendMessageBatchRequestEntry.1.MessageBody": [ "{}" ], "SendMessageBatchRequestEntry.1.MessageAttribute.1.Name": [ "id" ], "SendMessageBatchRequestEntry.1.MessageAttribute.1.Value.StringValue": [ "01c1c23a-ab8b-f730-ec5b-5dec12e2e500" ], "SendMessageBatchRequestEntry.1.MessageAttribute.1.Value.DataType": [ "String" ], "SendMessageBatchRequestEntry.1.MessageAttribute.2.Name": [ "contentType" ], "SendMessageBatchRequestEntry.1.MessageAttribute.2.Value.StringValue": [ "application/json;charset=UTF-8" ], "SendMessageBatchRequestEntry.1.MessageAttribute.2.Value.DataType": [ "String" ], "SendMessageBatchRequestEntry.1.MessageAttribute.3.Name": [ "timestamp" ], "SendMessageBatchRequestEntry.1.MessageAttribute.3.Value.StringValue": [ "1510117521473" ], "SendMessageBatchRequestEntry.1.MessageAttribute.3.Value.DataType": [ "Number.java.lang.Long" ] }

just looking for the right way to do this. essentially you're losing support for batch requests if you need to send messages to a fifo queue from the same service....or you need to do everything using the AWS SDK?

thanks in advance.

[jeffczaplewski]

maybe this just needs better documentation.

[schleifer-john]

I ran into the same issue. The core problem is that the AmazonSQSBufferedAsyncClient is loaded by default in the SqsClientConfiguration Spring Boot AutoConfig and FIFO queues don't support the AmazonSQSBufferedAsyncClient.

AWS Documentation:
http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/FIFO-queues.html

The Amazon SQS Buffered Asynchronous Client doesn't currently support FIFO queues.

AWS SDK GitHub Issue: https://github.com/aws/aws-sdk-java/issues/1316

SqsClientConfiguration:
https://github.com/spring-cloud/spring-cloud-aws/blob/5fde56ffaf1ff11a80ef99aa122142f2fc758387/spring-cloud-aws-messaging/src/main/java/org/springframework/cloud/aws/messaging/config/annotation/SqsClientConfiguration.java

The Error Message:

Caused by: org.springframework.messaging.MessageDeliveryException: The request must contain the parameter MessageGroupId. (Service: AmazonSQS; Status Code: 400; Error Code: MissingParameter; Request ID: 3ab60692-c2c8-5bff-8566-5439bd76fa94); nested exception is com.amazonaws.services.sqs.model.AmazonSQSException: The request must contain the parameter MessageGroupId. (Service: AmazonSQS; Status Code: 400; Error Code: MissingParameter; Request ID: 3ab60692-c2c8-5bff-8566-5439bd76fa94)
    at org.springframework.cloud.aws.messaging.core.QueueMessageChannel.sendInternal(QueueMessageChannel.java:68)
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:117)
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:104)
    at org.springframework.cloud.aws.messaging.core.support.AbstractMessageChannelMessagingSendingTemplate.doSend(AbstractMessageChannelMessagingSendingTemplate.java:49)
    at org.springframework.cloud.aws.messaging.core.support.AbstractMessageChannelMessagingSendingTemplate.doSend(AbstractMessageChannelMessagingSendingTemplate.java:35)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:143)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:123)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:118)
    at org.springframework.cloud.aws.messaging.core.support.AbstractMessageChannelMessagingSendingTemplate.convertAndSend(AbstractMessageChannelMessagingSendingTemplate.java:61)

Solution:
To fix this, I am overriding the SqsClientConfiguration Spring Boot AutoConfig by defining my own and returning AmazonSQSAsync instead of AmazonSQSBufferedAsyncClient:

@Configuration
public class FifoSqsClientConfiguration {

    @Autowired(required = false)
    private AWSCredentialsProvider awsCredentialsProvider;

    @Autowired(required = false)
    private RegionProvider regionProvider;

    /**
     * Creates the AmazonSQSAsync Bean. Overriding the default SQS Client Bean
     * config because AmazonSQSBufferedAsyncClient is not currently supported by FIFO queues.
     * 
     * @return the AmazonSQSAsync Bean
     */
    @Lazy
    @Bean(destroyMethod = "shutdown")
    public AmazonSQSAsync amazonSQS() throws Exception {

        AmazonSQSAsyncClientBuilder builder = AmazonSQSAsyncClientBuilder.standard();

        if (this.awsCredentialsProvider != null) {
            builder.withCredentials(this.awsCredentialsProvider);
        }

        if (this.regionProvider != null) {
            builder.withRegion(this.regionProvider.getRegion().getName());
        } else {
            builder.withRegion(Regions.DEFAULT_REGION);
        }

        return builder.build();
    }

    @Bean
    public QueueMessagingTemplate queueMessagingTemplate(AmazonSQSAsync amazonSqsClient, ResourceIdResolver resourceIdResolver) {
        return new QueueMessagingTemplate(amazonSqsClient, resourceIdResolver);
    }
}

Then to send messages, you need to add a Message Group Id and Deduplication Id as part of the message headers:

    @Autowired
    private QueueMessagingTemplate queueMessagingTemplate;

    public void send(Object object) {
        Map<String, Object> headers = new HashMap<>();
        headers.put(SqsMessageHeaders.SQS_GROUP_ID_HEADER, "my-application");
        headers.put(SqsMessageHeaders.SQS_DEDUPLICATION_ID_HEADER, UUID.randomUUID().toString());
        queueMessagingTemplate.convertAndSend("my-queue.fifo", object, headers);
    }

Not sure what the best-practices are here for these ids, but I am assuming the Message Group Id could be your application id and deduplication id could be a randomly generated id.

For listeners, I didn't need to do anything special other than use the correct FIFO-supported SQS Client.

I'm not sure if the AWS SDK plans to support AmazonSQSBufferedAsyncClient for FIFO queues in the future. If they do, then the custom configuration wouldn't be necessary. Alternatively, it would be nice if this project provided a configurable properties flag to enable FIFO queue mode. This way the auto config could decide the correct SQS client configuration to load.

[desiderati]

I've just tested with Spring Cloud AWS 2.2.0.RELEASE and you just have to pass the message-group-id and message-deduplication-id headers. There is no need to reconfigure amazonSQS client.

[tmnuwan12]

@maciejwalkowiak Is this enhancement still required..?

[maciejwalkowiak]

Since the underlying bug in AWS SDK has been fixed I think there is nothing to do on Spring Cloud AWS side. If you find some ideas for improvements please open new issue.