Suspected memory leak in KafkaThrottledLatestProcessedCommit

[tilakchowdary]

I have observed that our heap slowly increasing with KafkaThrottledLatestProcessedCommit

Here is the heap map Any inputs are greatly appreciated

[cescoffier]

We would need to see the application. There is a poison pill mechanism that makes sure we do not accumulate too much.

So make sure your messages are acknowledged.

[ozangunalp]

Yes, probably your messages are not acknowledged properly, so not cleared from tracked messages.

[tilakchowdary]

@cescoffier @ozangunalp Thanks for the suggestion. As I understand if it was due to messages are not being acknowledged properly it would result in TooManyMessagesWithoutAckException. However, In our case we don't see this issue and we see uninterrupted records consumption. From heap analysis only the processedOffsets field was leaking. I suspect concurrency usage of KafkaThrottledLatestProcessedCommit. Methods processed(long offset) and clearLesserSequentiallyProcessedOffsetsAndReturnLargestOffset() - Calling these concurrently by multiple threads resulting inconsistent view of processedOffsets.

[cescoffier]

Aren't these methods getting called from event loops ?

[tilakchowdary]

@cescoffier Thanks for clarification. Yes they are. Now I am not sure where is the issue. Will try to come up with a test case.

[ozangunalp]

If you've specific logs linked to this issue it can be handy. Otherwise, I don't see why this might happen.

[cescoffier]

@ozangunalp maybe if concurrency > 2, so multiple event loops. But still, should not lead to a leak.

[ozangunalp]

I've thought of that, but with concurrency > 2 sub-channels have their own copy of the commit handler and their own context acting on tracked records.

[cescoffier]

Yeah, that's what I just saw.

[tilakchowdary]

I think issue is that with IncomingKafkaRecordBatch only latestOffsetRecords of that batch are stored in OffsetStore.receivedOffsets and subsequently removed from processedOffsets. But processedOffsets will have all Offsets not just the latestOffsetRecords.

[ozangunalp]

I'll verify batch consuming for leaks. Are you acking individual messages and not the batch as a whole?

[tilakchowdary]

@ozangunalp Yes acking individual messages