From 409a9001f70507a34c911f7027897a9d069b7179 Mon Sep 17 00:00:00 2001
From: Hang Chen <chenhang@apache.org>
Date: Thu, 26 Aug 2021 17:50:30 +0800
Subject: [PATCH] revert PR 11594 to avoid copy data to direct buffer afer
 aircompressor upgrade to 0.20 (#11792)

### Motivation
Due to aircompressor 0.19 can't work with heap buffer on JDK1.8, so #11594 use copy data to direct buffer to avoid NoSuchMethodError exception. Now aircompressor released 0.20 and #11790 has upgrade the aircompressor version to 0.20 to fix this issue, we can revert #11594 to avoid copy data to direct buffer to improve performance.

### Modification
1. revert Fix java.lang.NoSuchMethodError: java.nio.ByteBuffer.position(I)Ljava/nio/ByteBuffer when enabling topic metadata compression #11594, but keep the tests.
---
 .../mledger/impl/MetaStoreImpl.java           | 22 +++----------------
 1 file changed, 3 insertions(+), 19 deletions(-)

diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
index 75ced800f836fd..5ad62b228bce08 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
@@ -317,7 +317,6 @@ public byte[] compressLedgerInfo(ManagedLedgerInfo managedLedgerInfo) {
             return managedLedgerInfo.toByteArray();
         }
         ByteBuf metadataByteBuf = null;
-        ByteBuf uncompressedByteBuf = null;
         ByteBuf encodeByteBuf = null;
         try {
             MLDataFormats.ManagedLedgerInfoMetadata mlInfoMetadata = MLDataFormats.ManagedLedgerInfoMetadata
@@ -330,13 +329,9 @@ public byte[] compressLedgerInfo(ManagedLedgerInfo managedLedgerInfo) {
             metadataByteBuf.writeShort(MAGIC_MANAGED_LEDGER_INFO_METADATA);
             metadataByteBuf.writeInt(mlInfoMetadata.getSerializedSize());
             metadataByteBuf.writeBytes(mlInfoMetadata.toByteArray());
-            byte[] byteArray = managedLedgerInfo.toByteArray();
-            // The reason for copy the data to a direct buffer here is to ensure the metadata compression feature can
-            // work on JDK1.8, for more details to see: https://github.com/apache/pulsar/issues/11593
-            uncompressedByteBuf = Unpooled.directBuffer(byteArray.length);
-            uncompressedByteBuf.writeBytes(byteArray);
+
             encodeByteBuf = getCompressionCodec(compressionType)
-                    .encode(uncompressedByteBuf);
+                    .encode(Unpooled.wrappedBuffer(managedLedgerInfo.toByteArray()));
             CompositeByteBuf compositeByteBuf = PulsarByteBufAllocator.DEFAULT.compositeBuffer();
             compositeByteBuf.addComponent(true, metadataByteBuf);
             compositeByteBuf.addComponent(true, encodeByteBuf);
@@ -347,9 +342,6 @@ public byte[] compressLedgerInfo(ManagedLedgerInfo managedLedgerInfo) {
             if (metadataByteBuf != null) {
                 metadataByteBuf.release();
             }
-            if (uncompressedByteBuf != null) {
-                uncompressedByteBuf.release();
-            }
             if (encodeByteBuf != null) {
                 encodeByteBuf.release();
             }
@@ -360,7 +352,6 @@ public ManagedLedgerInfo parseManagedLedgerInfo(byte[] data) throws InvalidProto
         ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
         if (byteBuf.readableBytes() > 0 && byteBuf.readShort() == MAGIC_MANAGED_LEDGER_INFO_METADATA) {
             ByteBuf decodeByteBuf = null;
-            ByteBuf compressedByteBuf = null;
             try {
                 int metadataSize = byteBuf.readInt();
                 byte[] metadataBytes = new byte[metadataSize];
@@ -369,12 +360,8 @@ public ManagedLedgerInfo parseManagedLedgerInfo(byte[] data) throws InvalidProto
                         MLDataFormats.ManagedLedgerInfoMetadata.parseFrom(metadataBytes);
 
                 long unpressedSize = metadata.getUncompressedSize();
-                // The reason for copy the data to a direct buffer here is to ensure the metadata compression feature
-                // can work on JDK1.8, for more details to see: https://github.com/apache/pulsar/issues/11593
-                compressedByteBuf = Unpooled.directBuffer(byteBuf.readableBytes());
-                compressedByteBuf.writeBytes(byteBuf);
                 decodeByteBuf = getCompressionCodec(metadata.getCompressionType())
-                        .decode(compressedByteBuf, (int) unpressedSize);
+                        .decode(byteBuf, (int) unpressedSize);
                 byte[] decodeBytes;
                 // couldn't decode data by ZLIB compression byteBuf array() directly
                 if (decodeByteBuf.hasArray() && !CompressionType.ZLIB.equals(metadata.getCompressionType())) {
@@ -392,9 +379,6 @@ public ManagedLedgerInfo parseManagedLedgerInfo(byte[] data) throws InvalidProto
                 if (decodeByteBuf != null) {
                     decodeByteBuf.release();
                 }
-                if (compressedByteBuf != null) {
-                    compressedByteBuf.release();
-                }
                 byteBuf.release();
             }
         } else {