Memory leak AsyncRabbitTemplate about spring-amqp HOT 8 OPEN

@tetrade ,

thank you for the report!

I'd be happy to fix the problem, but I don't understand how to read that object tree.
Yes, I do see RabbitMessageFuture and its TimeoutTask reference. But it is not clear for me what is requestMessage and why that RabbitMessageFuture is a part of it?

Any chances to determine what you are doing in your code with this AsyncRabbitTemplate?

from spring-amqp.

@artembilan
requestMessage - name of field in RabbitFuture. I guess that TimeoutTask in some reason didn't deleted from memory and still have reference from Scheldued that created here org.springframework.amqp.rabbit.AsyncRabbitTemplate#timeoutTask

from spring-amqp.

@artembilan

In short, I just use Rabbit to send a photo to a third-party service for processing and wait for a response via Direct Reply-to. When all the Future (many photos have been analyzed) are completed, I consider the project processed and upload everything to the user.

I initially thought that I was mistaken, because I have an internal ConcurrentMap in my code that takes into account the Id of the photo and stores it as the value of RabbitFuture so that when re-requesting it, I understand where to wait for a response.

However, given my implementation, there is hardly a leak there. And the profiler shows that there is a leak in scheduled tasks in Java

from spring-amqp.

As you see the logic there is like this:

	@Nullable
	private ScheduledFuture<?> timeoutTask(RabbitFuture<?> future) {
		if (this.receiveTimeout > 0) {
			this.lock.lock();
			try {
				if (!this.running) {
					this.pending.remove(future.getCorrelationId());
					throw new IllegalStateException("'AsyncRabbitTemplate' must be started.");
				}
				return this.taskScheduler.schedule(
						new TimeoutTask(future, this.pending, this.directReplyToContainer),
						Instant.now().plusMillis(this.receiveTimeout));
			}
			finally {
				this.lock.unlock();
			}
		}
		return null;
	}

So, such a timeout task is scheduled to be executed after that this.receiveTimeout (30 seconds by default).
Let's see if the problem with that ThreadPoolTaskScheduler with the default as private volatile int poolSize = 1;!
This way, it feels like "too many messages" would cause such a pool to be exhausted and therefore we see some kind of memory leak for some period of time until the queue of those TimeoutTask is cleared.

You can set an external ThreadPoolTaskScheduler with a bigger thread pool to mitigate workload.

Again: this is just an assumption since I don't see how else those TimeoutTask instances could be held in the memory if scheduling is done properly.

from spring-amqp.