[fix][broker] Fix memory leak during topic compaction

[coderzc]

Motivation

Found the following error log in CompactionTest.testCompactionDuplicate:

2023-11-30T12:48:53,002 - ERROR - [pulsar-client-io-692-4:ResourceLeakDetector@337] - LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
	io.netty.buffer.AdvancedLeakAwareByteBuf.readByte(AdvancedLeakAwareByteBuf.java:401)
	org.apache.pulsar.common.api.proto.LightProtoCodec.readVarInt(LightProtoCodec.java:140)
	org.apache.pulsar.common.api.proto.MessageMetadata.parseFrom(MessageMetadata.java:1315)
	org.apache.pulsar.common.protocol.Commands.parseMessageMetadata(Commands.java:459)
	org.apache.pulsar.common.protocol.Commands.parseMessageMetadata(Commands.java:446)
	org.apache.pulsar.client.impl.RawReaderImpl$RawConsumerImpl.tryCompletePending(RawReaderImpl.java:148)
	org.apache.pulsar.client.impl.RawReaderImpl$RawConsumerImpl.receiveRawAsync(RawReaderImpl.java:197)
	org.apache.pulsar.client.impl.RawReaderImpl.readNextAsync(RawReaderImpl.java:85)
	org.apache.pulsar.compaction.TwoPhaseCompactor.phaseTwoLoop(TwoPhaseCompactor.java:249)
	org.apache.pulsar.compaction.TwoPhaseCompactor.lambda$phaseTwoLoop$16(TwoPhaseCompactor.java:256)
	java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718)
	java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)
	java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
	java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Thread.java:833)

Fix memory leak after RawReader reconnects when topic compaction, this issue introduced from #21081.

Modifications

Close RawMessage if messageId <= lastCompactedMessageId.

Documentation

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

Matching PR in forked repository

PR in forked repository:

[codecov-commenter]

Codecov Report

Merging #21647 (b0656d2) into master (81a9a52) will increase coverage by 48.23%.
Report is 4 commits behind head on master.
The diff coverage is 100.00%.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #21647       +/-   ##
=============================================
+ Coverage     24.71%   72.95%   +48.23%     
- Complexity      199     3748     +3549     
=============================================
  Files          1713     1892      +179     
  Lines        131084   140681     +9597     
  Branches      14320    15495     +1175     
=============================================
+ Hits          32403   102634    +70231     
+ Misses        93583    29921    -63662     
- Partials       5098     8126     +3028     

Flag	Coverage Δ
inttests	24.18% <0.00%> (?)
systests	24.67% <0.00%> (-0.05%)	⬇️
unittests	72.26% <100.00%> (?)

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

Files	Coverage Δ
...ar/broker/authorization/AuthorizationProvider.java	18.51% <ø> (+16.66%)	⬆️
...rg/apache/pulsar/compaction/TwoPhaseCompactor.java	73.94% <100.00%> (+33.44%)	⬆️
.../java/org/apache/pulsar/client/impl/ClientCnx.java	70.45% <ø> (+32.78%)	⬆️
...a/org/apache/pulsar/client/impl/Murmur3Hash32.java	80.00% <ø> (ø)

... and 1550 files with indirect coverage changes

[poorbarcode]

Good work, could you add a test for this case?

[coderzc]

Good work, could you add a test for this case?

If a memory leak occurs here, the above log will appear in CompactionTest.testCompactionDuplicate