[hbase] branch branch-2 updated: HBASE-26324 Reuse compressors and decompressors in WAL CompressionContext (#3728)

[ap...@apache.org]

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

apurtell pushed a commit to branch branch-2
in repository https://gitbox.apache.org/repos/asf/hbase.git


The following commit(s) were added to refs/heads/branch-2 by this push:
     new 5af0220  HBASE-26324 Reuse compressors and decompressors in WAL CompressionContext (#3728)
5af0220 is described below

commit 5af02201d261c95dbc0c2f69ded23d372db1f223
Author: Andrew Purtell <ap...@apache.org>
AuthorDate: Thu Oct 7 09:45:17 2021 -0700

    HBASE-26324 Reuse compressors and decompressors in WAL CompressionContext (#3728)
    
    Signed-off-by: Viraj Jasani <vj...@apache.org>
---
 .../hbase/regionserver/wal/CompressionContext.java   | 20 ++++++++++++++++++--
 1 file changed, 18 insertions(+), 2 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
index 82bad93..bfb7f9a 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
@@ -32,6 +32,8 @@ import org.apache.hadoop.hbase.io.BoundedDelegatingInputStream;
 import org.apache.hadoop.hbase.io.TagCompressionContext;
 import org.apache.hadoop.hbase.io.compress.Compression;
 import org.apache.hadoop.hbase.io.util.Dictionary;
+import org.apache.hadoop.io.compress.Compressor;
+import org.apache.hadoop.io.compress.Decompressor;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -69,6 +71,8 @@ public class CompressionContext {
     static final int IO_BUFFER_SIZE = 4096;
 
     private final Compression.Algorithm algorithm;
+    private Compressor compressor;
+    private Decompressor decompressor;
     private BoundedDelegatingInputStream lowerIn;
     private ByteArrayOutputStream lowerOut;
     private InputStream compressedIn;
@@ -87,7 +91,10 @@ public class CompressionContext {
       if (compressedOut == null) {
         // Create the output streams here the first time around.
         lowerOut = new ByteArrayOutputStream();
-        compressedOut = algorithm.createCompressionStream(lowerOut, algorithm.getCompressor(),
+        if (compressor == null) {
+          compressor = algorithm.getCompressor();
+        }
+        compressedOut = algorithm.createCompressionStream(lowerOut, compressor,
           IO_BUFFER_SIZE);
       } else {
         lowerOut.reset();
@@ -107,7 +114,10 @@ public class CompressionContext {
       // Create the input streams here the first time around.
       if (compressedIn == null) {
         lowerIn = new BoundedDelegatingInputStream(in, inLength);
-        compressedIn = algorithm.createDecompressionStream(lowerIn, algorithm.getDecompressor(),
+        if (decompressor == null) {
+          decompressor = algorithm.getDecompressor();
+        }
+        compressedIn = algorithm.createDecompressionStream(lowerIn, decompressor,
           IO_BUFFER_SIZE);
       } else {
         lowerIn.setDelegate(in, inLength);
@@ -152,6 +162,12 @@ public class CompressionContext {
         }
       }
       lowerIn = null;
+      if (compressor != null) {
+        compressor.reset();
+      }
+      if (decompressor != null) {
+        decompressor.reset();
+      }
     }
 
   }