reactor.rabbitmq.Utils.cache could cause memory leak

robotmrv commented 5 years ago

reactor.rabbitmq.Utils.cache() and methods which use it could cause memory leak.
For details please see https://github.com/reactor/reactor-core/issues/1916

Expected Behavior

Utils#cache() should not submit invalidation task for "infinite" time on success, but just cache result.

Actual Behavior

Utils#cache() submits invalidation task with delay Duration.ofMillis(Long.MAX_VALUE) i.e. "infinite" time on success, which could cause memory leak

Steps to Reproduce

make a lot of cache(myMono).subscribe() and invalidation tasks will never be garbage collected.

Possible Solution

see related reactor-core issue https://github.com/reactor/reactor-core/issues/1916
or if you do not see issue here just warn in javadocs that Utils#cache() or Sender.rpcClient() could cause memory leaks if they are invoked many times.

Your Environment

Reactor version(s) used: reactor-rabbitmq:1.3.0.RELEASE
Other relevant libraries versions (eg. netty, ...):
JVM version (javar -version):
OS and version (eg uname -a):

simonbasle commented 5 years ago

Again, I don't think that "make a lot of cache(MyMono).subscribe() is a relevant pattern, and this can probably only be fixed in core, so this doesn't feel too bad. I'm opening a PR on the core issue though, as a general low-criticality enhancement.

acogoluegnes commented 5 years ago

This PR in Reactor Core fixes the memory leak problem for Duration.ofMillis(Long.MAX_VALUE), that is for Utils#cache.

robotmrv commented 5 years ago

@simonbasle
I am happy with this solution

I'm opening a PR on the core issue though, as a general low-criticality enhancement.

but unfortunately

I don't think that "make a lot of cache(MyMono).subscribe() is a relevant pattern

this is my pattern, because we have dynamic routing key in RPC so execution is something like

Web Endpoint (it could be many requests) ->
do some stuff ->
create new RabbitMQ RPC client and call it ->
response Ok

So Utils.cache() caches channel and submits to Schedulers.parallel() invalidation task which will never be evicted, and will hold channel and all other stuff from MonoCacheTime till application shutdown

acogoluegnes commented 5 years ago

@robotmrv The PR mentioned above will not schedule any invalidation task as soon as Duration.ofMillis(Long.MAX_VALUE) is used, which is the case in Utils#cache. So no task should leak. Does that address your concern?

robotmrv commented 5 years ago

@acogoluegnes I've just pointed out that there is a case where current behavior could lead to significant leak and it should be fixed.
I am ok and happy with solution from the PR

reactor / reactor-rabbitmq