Potential memory leak when consuming for over a Month

[FransHeuvelmans]

java-pubsublite-kafka/src/main/java/com/google/cloud/pubsublite/kafka/SingleSubscriptionConsumerImpl.java

Line 137 in 4a575ea

stopSleepingSignals.add(wakeupTriggered);

Hey java-pubsublite-kafka-team,

I noticed a potential memory leak in a heap dump of a long-running deployment of pubsublite-kafka). It occurred in a program which uses spring-kafka in combination with pubsublite-kafka to consume a stream of (Avro encoded) data. It might be that we simply do not use the API correctly or have another issue. We use Maven and it does not indicate any conflicting dependencies for pubsublite-kafka.

Problem: After running and consuming data for over a month, the wakeupTriggered Future is reported at a size of over 6 GiB in the heap dump (after analyzing with Eclipse MAT) for each Partition. The problem seems to be that ApiFuturesExtensions.whenFirstDone keeps adding Listeners but they are never GC-ed if the wakeupTrigger Future is never triggered. The analysis also shows that assignmentChanged has grown to a couple of MiBs because of Listeners, but it is nowhere near as big. I think it is being triggered every once in a while.

I’ve made a crude fix in a fork but it kind of breaks the wakeUp functionality so it is probably not a good solution. The big memory leak was gone after this fix (for us it works).

Sadly I can’t share the code of the program and I haven’t invested time in a minimal bug recreation. I understand if this is too little information to go on but I thought I'd report it.

Version info:
pubsublite-kafka: v1.0.0
Spring-kafka: 2.9.0
libraries-bom: 26.0.0
Java version: openjdk 11.0.16 2022-07-19
GC: Garbage-First (G1) Garbage Collector