SQS Source has unbounded parallelism

[aserrallerios]

Every time downstream pulls when the buffer is not full, a new async request is spawned using AmazonSQSAsyncClient, thus ignoring the longPollingDuration or the maxBufferSize configuration. This makes it unusable and dangerous.

I understand that throughput of this Source is achieved with parallelism of requests of AmazonSQSAsyncClient. This client can be instantiated with a custom thread pool but if the downstream is too fast, this will effectively starve other usages of the same client instance. For example:

implicit val sqsClient : AmazonSQSAsyncClient = ???

alpakkaSqsSource(queueName1) ~> } ~> fastFlow ~> deleteSqsMessage
alpakkaSqsSource(queueName2) ~> }

The sqsSources will exhaust all the resources of the sqsClient. This can be easily fixed using a different sqsClient, but IMHO this should be properly advertised on the documentation because it's clearly not how you'd use this api:

alpakkaSqsSource(queueName1)(sqsClient1) ~> } ~> fastFlow ~> deleteSqsMessage(sqsClient3)
alpakkaSqsSource(queueName2)(sqsClient2) ~> }

Now, we've avoided starvation BUT if we check our SQS metrics on AWS console we'll find astronomic polling rates (which are billed to your AWS account). This happens because the Source polls whenever its buffer it's not full on every downstream pull, as I previously said.

One would expect that the Source honors the longPollingDuration and the maxBufferSize configuration in order to control its internal logic and throughput, but it fails under the most trivial scenarios:

Scenario 1:

Max buffer: 100
Current buffer: 85
Batch size: 10
Downstream pulls: 2

Outcome: 2 requests are performed, when only 1 of the responses would fit in the local buffer

Scenario 2:

Max buffer: 100
Current buffer: 0
Batch size: 10
Downstream pulls: 2

Outcome: 2 requests are performed, even when the application has no clue about the current messages on the server, making longPollingDuration parameter useless

---

Solution?
Source may keep local state to track if there are pending async requests and the latest server response.
I think that would be enough, and it'd make the Source behave as expected.

[dpfeiffer]

@aserrallerios What do you think of configurable parallel async request count? This would allow to keep throughput high if needed but would also allow to keep costs low. I'll try to open a PR by the end of the week.

[aserrallerios]

There are some problems with that.

• You'd need to configure AmazonSQSAsyncClient and alpakka Source with almost the same parameters. It feels inelegant.
• You achieve the same thing configuring the AmazonSQSAsyncClient thread pool in the first place.
• Even though parallelism would be explicitly bounded, it wouldn't still be an optimal usage of resources.
• IMO parallelism shouldn't be higher than buffer max size / batch max size, it doesn't make any sense to me.

What would you think of a constructor that doesn't take a AmazonSQSAsyncClient as parameter, and instantiates it with the recommended parameters (for the thread pool and for the http client)? That would fix one of the problems: the parallelism would be bounded with a single "parallelism" parameter or even calculated by buffer max size / batch max size parameters, and users of the alpakka api would not starve their other AmazonSQSAsyncClients usages.

Then, optimizing the usage of the queue would be a different thing, but we'd be pretty close if the max parallelism is calculated and not a user setting. The only thing remaining would to keep track of the remote queue: if it's empty (the last request returned 0 results, for example), just limit parallelism at 1.

[dpfeiffer]

In the first place I decided to hand in the AmazonSQSAsyncClient so it could be used in other places as well, so you don't have to spin up two thread pools when you simply want to achieve something like this:

SqsSource(...)
  .via(foo)
  .via(deleteMessageFromQueue)
  .runWith(...)

I agree that configuring everything correctly will get really hard if we add a parallelism parameter and allow to pass in a AmazonSQSAsyncClient. Maybe we could provide both constructors, one that gets a custom client and one that doesn't?

I wouldn't calculate the parallelism with buffer max size / batch max size because it makes it intransparent and event harder to configure when we provide 2 different constructors. A separate parallelism setting (with a check that it doesn't exceed the max buffer size if all requests are finished) would allow more transparent configuration.

I am not sure if we can infer a limited parallelism from an empty response. As I know this can happen also if the queue is not empty (maybe we need a more sophisticated way of doing that)

[aserrallerios]

An empty response means that remote queue may be empty, and you can leverage on the long polling until it has messages again. Another option is to use the AWS API to directly query the queue, but the result would be the same, you cannot assume that the queue is empty even when the latest request returned that the queue size is 0. What we got here is a best effort approach, and I agree that is not so important if the async requests are bounded.

A derived parallelism setting would make it intransparent but reasonable, you'll always comply with the max buffer setting. The check that you propose sounds like "ignore max buffer size" check to me.

[dpfeiffer]

@aserrallerios I'll try out a few options on the weekend.