[pulsar] branch master updated: [fix][offload] Fix memory leak while Offloading ledgers (#18500)

[eolivelli]

This is an automated email from the ASF dual-hosted git repository.

eolivelli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 6ff7d459697 [fix][offload] Fix memory leak while Offloading ledgers 
(#18500)
6ff7d459697 is described below

commit 6ff7d459697c2496de29ef077eb0f574632ebe6d
Author: Enrico Olivelli <eolive...@apache.org>
AuthorDate: Wed Nov 16 13:44:43 2022 +0100

    [fix][offload] Fix memory leak while Offloading ledgers (#18500)
---
 .../jcloud/impl/BlockAwareSegmentInputStreamImpl.java     |  2 +-
 .../jcloud/impl/BlockAwareSegmentInputStreamTest.java     | 15 +++++++++++++++
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git 
a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlockAwareSegmentInputStreamImpl.java
 
b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlockAwareSegmentInputStreamImpl.java
index 5f778cab51f..d07fbdb9247 100644
--- 
a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlockAwareSegmentInputStreamImpl.java
+++ 
b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlockAwareSegmentInputStreamImpl.java
@@ -278,7 +278,7 @@ public class BlockAwareSegmentInputStreamImpl extends 
BlockAwareSegmentInputStre
         // And through debug, writeBlobStore.uploadMultipartPart in the 
offload method also will trigger
         // the close method.
         // So we add the close variable to avoid release paddingBuf twice.
-        if (!close.compareAndSet(false, true)) {
+        if (close.compareAndSet(false, true)) {
             super.close();
             dataBlockHeaderStream.close();
             if (!entriesByteBuf.isEmpty()) {
diff --git 
a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlockAwareSegmentInputStreamTest.java
 
b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlockAwareSegmentInputStreamTest.java
index 47989e03605..5ca4d6da20b 100644
--- 
a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlockAwareSegmentInputStreamTest.java
+++ 
b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlockAwareSegmentInputStreamTest.java
@@ -30,6 +30,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Iterator;
@@ -798,4 +799,18 @@ public class BlockAwareSegmentInputStreamTest {
 
         inputStream.close();
     }
+
+    @Test
+    public void testCloseReleaseResources() throws Exception {
+        ReadHandle readHandle = new MockReadHandle(1, 10, 10);
+
+        BlockAwareSegmentInputStreamImpl inputStream = new 
BlockAwareSegmentInputStreamImpl(readHandle, 0, 1024);
+        inputStream.read();
+        Field field = 
BlockAwareSegmentInputStreamImpl.class.getDeclaredField("paddingBuf");
+        field.setAccessible(true);
+        ByteBuf paddingBuf = (ByteBuf) field.get(inputStream);
+        assertEquals(1, paddingBuf.refCnt());
+        inputStream.close();
+        assertEquals(0, paddingBuf.refCnt());
+    }
 }