module distributedlog-common/distributedlog-protocol: refactor ByteBu…

…f release usage (#3693) Co-authored-by: lushiji <[email protected]> (cherry picked from commit a71c7a7)
apache · Dec 8, 2022 · 8e352d4 · 8e352d4
1 parent bf18fc5
commit 8e352d4
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 15 deletions.
diff --git a/...istributedlog/common/src/test/java/org/apache/distributedlog/io/TestCompressionCodec.java b/...istributedlog/common/src/test/java/org/apache/distributedlog/io/TestCompressionCodec.java
@@ -24,6 +24,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.buffer.Unpooled;
+import io.netty.util.ReferenceCountUtil;
 import java.nio.ByteBuffer;
 import org.junit.Test;
 
@@ -70,9 +71,9 @@ private void testCompressionCodec(CompressionCodec codec) throws Exception {
         decompressedBuf.readBytes(decompressedData);
         assertArrayEquals("The decompressed bytes should be same as the original bytes",
                 data, decompressedData);
-        buf.release();
-        compressedBuf.release();
-        decompressedBuf.release();
+        ReferenceCountUtil.safeRelease(buf);
+        ReferenceCountUtil.safeRelease(compressedBuf);
+        ReferenceCountUtil.safeRelease(decompressedBuf);
     }
 
     private void testCompressionCodec2(CompressionCodec codec) throws Exception {
@@ -93,9 +94,9 @@ private void testCompressionCodec2(CompressionCodec codec) throws Exception {
         byte[] decompressedData = new byte[decompressedBuf.readableBytes()];
         decompressedBuf.slice().readBytes(decompressedData);
 
-        buffer.release();
-        compressedBuf.release();
-        decompressedBuf.release();
+        ReferenceCountUtil.safeRelease(buffer);
+        ReferenceCountUtil.safeRelease(compressedBuf);
+        ReferenceCountUtil.safeRelease(decompressedBuf);
     }
 
 }
diff --git a/...ributedlog/protocol/src/main/java/org/apache/distributedlog/EnvelopedRecordSetReader.java b/...ributedlog/protocol/src/main/java/org/apache/distributedlog/EnvelopedRecordSetReader.java
@@ -22,6 +22,7 @@
 import static org.apache.distributedlog.LogRecordSet.VERSION;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
 import java.io.IOException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.distributedlog.io.CompressionCodec;
@@ -80,10 +81,10 @@ class EnvelopedRecordSetReader implements LogRecordSet.Reader {
             CompressionCodec codec = CompressionUtils.getCompressionCodec(Type.of(codecCode));
             this.reader = codec.decompress(compressedBuf, decompressedDataLen);
         } finally {
-            compressedBuf.release();
+            ReferenceCountUtil.safeRelease(compressedBuf);
         }
         if (numRecords == 0) {
-            this.reader.release();
+            ReferenceCountUtil.safeRelease(this.reader);
         }
     }
 
@@ -110,7 +111,7 @@ public LogRecordWithDLSN nextRecord() throws IOException {
 
         // release the record set buffer when exhausting the reader
         if (0 == numRecords) {
-            this.reader.release();
+            ReferenceCountUtil.safeRelease(this.reader);
         }
 
         return record;
@@ -120,7 +121,7 @@ public LogRecordWithDLSN nextRecord() throws IOException {
     public void release() {
         if (0 != numRecords) {
             numRecords = 0;
-            reader.release();
+            ReferenceCountUtil.safeRelease(reader);
         }
     }
 }
diff --git a/...ributedlog/protocol/src/main/java/org/apache/distributedlog/EnvelopedRecordSetWriter.java b/...ributedlog/protocol/src/main/java/org/apache/distributedlog/EnvelopedRecordSetWriter.java
@@ -157,14 +157,14 @@ ByteBuf createBuffer() {
     @Override
     public synchronized void completeTransmit(long lssn, long entryId, long startSlotId) {
         satisfyPromises(lssn, entryId, startSlotId);
-        buffer.release();
-        ReferenceCountUtil.release(recordSetBuffer);
+        ReferenceCountUtil.safeRelease(buffer);
+        ReferenceCountUtil.safeRelease(recordSetBuffer);
     }
 
     @Override
     public synchronized void abortTransmit(Throwable reason) {
         cancelPromises(reason);
-        buffer.release();
-        ReferenceCountUtil.release(recordSetBuffer);
+        ReferenceCountUtil.safeRelease(buffer);
+        ReferenceCountUtil.safeRelease(recordSetBuffer);
     }
 }
diff --git a/stream/distributedlog/protocol/src/main/java/org/apache/distributedlog/LogRecord.java b/stream/distributedlog/protocol/src/main/java/org/apache/distributedlog/LogRecord.java
@@ -21,6 +21,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufInputStream;
 import io.netty.buffer.Unpooled;
+import io.netty.util.ReferenceCountUtil;
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
@@ -225,7 +226,7 @@ public ByteBuf getPayloadBuf() {
 
     void setPayloadBuf(ByteBuf payload, boolean copyData) {
         if (null != this.payload) {
-            this.payload.release();
+            ReferenceCountUtil.safeRelease(this.payload);
         }
         if (copyData) {
             this.payload = Unpooled.copiedBuffer(payload);