[fix][broker] Fix direct memory leak in RawReaderImpl

[lhotari]

Fixes #18927

Motivation

There's a direct memory leak in RawReaderImpl, please see #18927 for details.

Modifications

Override methods org.apache.pulsar.client.impl.ConsumerBase#failPendingReceive and org.apache.pulsar.client.impl.ConsumerBase#clearIncomingMessages in org.apache.pulsar.client.impl.RawReaderImpl.RawConsumerImpl so that cleanup happens properly. Don't create RawMessageImpl instances in messageReceived method if consumer is already closing or closed.
The changes should prevent future direct memory leaks caused by seeks or when the consumer is already closed.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: lhotari#113

[congbobo184]

closeAsync and messageReceived don't in the same thread run and have no locks. so the message also can be added in incomingRawMessages after it has been closed

step :

1. thread-1 messageReceived check the state is ready.
   https://github.com/apache/pulsar/pull/18928/files#diff-2d41ea4027219e13a0a942edc54352bb0a997060112705796d1d811213d585f3R245
2. thread-2 closeAsync() the RawReaderImpl state change to close and reset the incomingRawMessages
   https://github.com/apache/pulsar/pull/18928/files#diff-2d41ea4027219e13a0a942edc54352bb0a997060112705796d1d811213d585f3R245
3. thread-1 messageReceived continue to run the add operation
   https://github.com/apache/pulsar/pull/18928/files#diff-2d41ea4027219e13a0a942edc54352bb0a997060112705796d1d811213d585f3R252

[lhotari]

Yes, that's true. Pulsar contains a lot of such race conditions. I wonder if it's worth fixing that one, since it would add complexity to the code. There are multiple of such race conditions in ConsumerBase's closeAsync method. The fix would have to be done there. The problem is more severe in ConsumerBase since it can impact the memory limit controller. I'll create a separate issue to track the issue.

[lhotari]

I filed #18938

[lhotari]

I also added #18939

[poorbarcode]

Should we release headersAndPayload here?

[lhotari]

Perhaps. I think this is a problem that should be covered while fixing #18938 . It's not specific to RawReaderImpl's consumer.

[eolivelli]

+1

I think that the best approach is to fix all the findings in separate patches.

I would commit this patch and then let @lhotari move forward with further fixes

[codecov-commenter]

Codecov Report

❌ Patch coverage is 82.75862% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 46.79%. Comparing base (050b310) to head (5d35ed3).
⚠️ Report is 3124 commits behind head on master.

Files with missing lines	Patch %	Lines
...a/org/apache/pulsar/client/impl/RawReaderImpl.java	82.75%	3 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18928      +/-   ##
============================================
- Coverage     47.35%   46.79%   -0.56%     
- Complexity     9384    10533    +1149     
============================================
  Files           623      706      +83     
  Lines         59104    69021    +9917     
  Branches       6146     7394    +1248     
============================================
+ Hits          27987    32300    +4313     
- Misses        28100    33118    +5018     
- Partials       3017     3603     +586     

Flag	Coverage Δ
unittests	46.79% <82.75%> (-0.56%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...a/org/apache/pulsar/client/impl/RawReaderImpl.java	81.73% <82.75%> (-2.18%)	⬇️

... and 146 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.