AbstractPollingMessageSource doesn't clear pollingFutures on poll errors

[dzmitry-dulko]

Component:
"SQS"

Description
I found out that if poll in AbstractPollingMessageSource in package io.awspring.cloud.sqs.listener.source;
failed for some reason it will keep adding CompletableFutures in pollingFutures and not clearing it as supposed to

private <F> CompletableFuture<F> managePollingFuture(CompletableFuture<F> pollingFuture) {
		this.pollingFutures.add(pollingFuture);
		pollingFuture.thenRun(() -> this.pollingFutures.remove(pollingFuture));
		return pollingFuture;
	}

in theRun(...) description

Returns a new CompletionStage that, when this stage completes normally, executes the given action.

And it leads to uncleared resources which lead me to OOM. Is that suppose to be?

Sample
In my case it was 400 and 403 (expired credentials probably) responses from AWS

Please advise on this

[tomazfernandes]

Hey @dzmitry-dulko, thanks for bringing this up.

You're correct, if the polling future fails the thenRun callback won't be executed and the future won't be cleaned.

We should use whenComplete instead, so it's executed whether there's an error or not.

Would you like to contribute a PR with the fix?

Ideally we'd have a test reproducing the error scenario and making sure it's cleaned after.

[dzmitry-dulko]

Would you like to contribute a PR with the fix?

I do

Ideally we'd have a test reproducing the error scenario and making sure it's cleaned after.

You may try using fake credentials. I personally used fake secret key to our AWS account and was checking pollingFutures in debug, and yes it was not cleaning resources

spring:
    aws:
      credentials:
        secret-key: "some fake key"