Skip to content

Commit 3eedc09

Browse files
sunhellysunhelly
authored
HBASE-27672 Read RPC threads may BLOCKED at the Configuration.get when using java compression (#5075)
Signed-off-by: Bryan Beaudreault <bbeaudreault@apache.org>
1 parent 59fdaa2 commit 3eedc09

File tree

9 files changed

+62
-28
lines changed

9 files changed

+62
-28
lines changed

‎hbase-compression/hbase-compression-aircompressor/src/main/java/org/apache/hadoop/hbase/io/compress/aircompressor/Lz4Codec.java

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,11 @@ public class Lz4Codec implements Configurable, CompressionCodec {
4646
public static final String LZ4_BUFFER_SIZE_KEY = "hbase.io.compress.lz4.buffersize";
4747

4848
private Configuration conf;
49+
private int bufferSize;
4950

5051
public Lz4Codec() {
5152
conf = new Configuration();
53+
bufferSize = getBufferSize(conf);
5254
}
5355

5456
@Override
@@ -59,6 +61,7 @@ public Configuration getConf() {
5961
@Override
6062
public void setConf(Configuration conf) {
6163
this.conf = conf;
64+
this.bufferSize = getBufferSize(conf);
6265
}
6366

6467
@Override
@@ -79,7 +82,7 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti
7982
@Override
8083
public CompressionInputStream createInputStream(InputStream in, Decompressor d)
8184
throws IOException {
82-
return new BlockDecompressorStream(in, d, getBufferSize(conf));
85+
return new BlockDecompressorStream(in, d, bufferSize);
8386
}
8487

8588
@Override
@@ -90,7 +93,6 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc
9093
@Override
9194
public CompressionOutputStream createOutputStream(OutputStream out, Compressor c)
9295
throws IOException {
93-
int bufferSize = getBufferSize(conf);
9496
return new BlockCompressorStream(out, c, bufferSize,
9597
CompressionUtil.compressionOverhead(bufferSize));
9698
}

‎hbase-compression/hbase-compression-aircompressor/src/main/java/org/apache/hadoop/hbase/io/compress/aircompressor/LzoCodec.java

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,11 @@ public class LzoCodec implements Configurable, CompressionCodec {
4646
public static final String LZO_BUFFER_SIZE_KEY = "hbase.io.compress.lzo.buffersize";
4747

4848
private Configuration conf;
49+
private int bufferSize;
4950

5051
public LzoCodec() {
5152
conf = new Configuration();
53+
bufferSize = getBufferSize(conf);
5254
}
5355

5456
@Override
@@ -59,6 +61,7 @@ public Configuration getConf() {
5961
@Override
6062
public void setConf(Configuration conf) {
6163
this.conf = conf;
64+
this.bufferSize = getBufferSize(conf);
6265
}
6366

6467
@Override
@@ -79,7 +82,7 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti
7982
@Override
8083
public CompressionInputStream createInputStream(InputStream in, Decompressor d)
8184
throws IOException {
82-
return new BlockDecompressorStream(in, d, getBufferSize(conf));
85+
return new BlockDecompressorStream(in, d, bufferSize);
8386
}
8487

8588
@Override
@@ -90,7 +93,6 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc
9093
@Override
9194
public CompressionOutputStream createOutputStream(OutputStream out, Compressor c)
9295
throws IOException {
93-
int bufferSize = getBufferSize(conf);
9496
return new BlockCompressorStream(out, c, bufferSize,
9597
CompressionUtil.compressionOverhead(bufferSize));
9698
}

‎hbase-compression/hbase-compression-aircompressor/src/main/java/org/apache/hadoop/hbase/io/compress/aircompressor/SnappyCodec.java

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,11 @@ public class SnappyCodec implements Configurable, CompressionCodec {
4646
public static final String SNAPPY_BUFFER_SIZE_KEY = "hbase.io.compress.snappy.buffersize";
4747

4848
private Configuration conf;
49+
private int bufferSize;
4950

5051
public SnappyCodec() {
5152
conf = new Configuration();
53+
bufferSize = getBufferSize(conf);
5254
}
5355

5456
@Override
@@ -59,6 +61,7 @@ public Configuration getConf() {
5961
@Override
6062
public void setConf(Configuration conf) {
6163
this.conf = conf;
64+
this.bufferSize = getBufferSize(conf);
6265
}
6366

6467
@Override
@@ -79,7 +82,7 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti
7982
@Override
8083
public CompressionInputStream createInputStream(InputStream in, Decompressor d)
8184
throws IOException {
82-
return new BlockDecompressorStream(in, d, getBufferSize(conf));
85+
return new BlockDecompressorStream(in, d, bufferSize);
8386
}
8487

8588
@Override
@@ -90,7 +93,6 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc
9093
@Override
9194
public CompressionOutputStream createOutputStream(OutputStream out, Compressor c)
9295
throws IOException {
93-
int bufferSize = getBufferSize(conf);
9496
return new BlockCompressorStream(out, c, bufferSize,
9597
CompressionUtil.compressionOverhead(bufferSize));
9698
}

‎hbase-compression/hbase-compression-aircompressor/src/main/java/org/apache/hadoop/hbase/io/compress/aircompressor/ZstdCodec.java

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,14 +54,17 @@ public class ZstdCodec implements Configurable, CompressionCodec {
5454
public static final int ZSTD_BUFFER_SIZE_DEFAULT = 256 * 1024;
5555

5656
private Configuration conf;
57+
private int bufferSize;
5758

5859
public ZstdCodec() {
5960
conf = new Configuration();
61+
bufferSize = getBufferSize(conf);
6062
}
6163

6264
@Override
6365
public void setConf(Configuration conf) {
6466
this.conf = conf;
67+
this.bufferSize = getBufferSize(conf);
6568
}
6669

6770
@Override
@@ -87,7 +90,7 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti
8790
@Override
8891
public CompressionInputStream createInputStream(InputStream in, Decompressor d)
8992
throws IOException {
90-
return new BlockDecompressorStream(in, d, getBufferSize(conf));
93+
return new BlockDecompressorStream(in, d, bufferSize);
9194
}
9295

9396
@Override
@@ -98,7 +101,6 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc
98101
@Override
99102
public CompressionOutputStream createOutputStream(OutputStream out, Compressor c)
100103
throws IOException {
101-
int bufferSize = getBufferSize(conf);
102104
return new BlockCompressorStream(out, c, bufferSize,
103105
CompressionUtil.compressionOverhead(bufferSize));
104106
}

‎hbase-compression/hbase-compression-brotli/src/main/java/org/apache/hadoop/hbase/io/compress/brotli/BrotliCodec.java

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -47,9 +47,15 @@ public class BrotliCodec implements Configurable, CompressionCodec {
4747
public static final int BROTLI_BUFFERSIZE_DEFAULT = 256 * 1024;
4848

4949
private Configuration conf;
50+
private int bufferSize;
51+
private int level;
52+
private int window;
5053

5154
public BrotliCodec() {
5255
conf = new Configuration();
56+
bufferSize = getBufferSize(conf);
57+
level = getLevel(conf);
58+
window = getWindow(conf);
5359
}
5460

5561
@Override
@@ -60,16 +66,19 @@ public Configuration getConf() {
6066
@Override
6167
public void setConf(Configuration conf) {
6268
this.conf = conf;
69+
this.bufferSize = getBufferSize(conf);
70+
this.level = getLevel(conf);
71+
this.window = getWindow(conf);
6372
}
6473

6574
@Override
6675
public Compressor createCompressor() {
67-
return new BrotliCompressor(getLevel(conf), getWindow(conf), getBufferSize(conf));
76+
return new BrotliCompressor(level, window, bufferSize);
6877
}
6978

7079
@Override
7180
public Decompressor createDecompressor() {
72-
return new BrotliDecompressor(getBufferSize(conf));
81+
return new BrotliDecompressor(bufferSize);
7382
}
7483

7584
@Override
@@ -80,7 +89,7 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti
8089
@Override
8190
public CompressionInputStream createInputStream(InputStream in, Decompressor d)
8291
throws IOException {
83-
return new BlockDecompressorStream(in, d, getBufferSize(conf));
92+
return new BlockDecompressorStream(in, d, bufferSize);
8493
}
8594

8695
@Override
@@ -91,7 +100,6 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc
91100
@Override
92101
public CompressionOutputStream createOutputStream(OutputStream out, Compressor c)
93102
throws IOException {
94-
int bufferSize = getBufferSize(conf);
95103
return new BlockCompressorStream(out, c, bufferSize,
96104
CompressionUtil.compressionOverhead(bufferSize));
97105
}

‎hbase-compression/hbase-compression-lz4/src/main/java/org/apache/hadoop/hbase/io/compress/lz4/Lz4Codec.java

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -44,9 +44,11 @@ public class Lz4Codec implements Configurable, CompressionCodec {
4444
public static final String LZ4_BUFFER_SIZE_KEY = "hbase.io.compress.lz4.buffersize";
4545

4646
private Configuration conf;
47+
private int bufferSize;
4748

4849
public Lz4Codec() {
4950
conf = new Configuration();
51+
this.bufferSize = getBufferSize(conf);
5052
}
5153

5254
@Override
@@ -57,16 +59,17 @@ public Configuration getConf() {
5759
@Override
5860
public void setConf(Configuration conf) {
5961
this.conf = conf;
62+
this.bufferSize = getBufferSize(conf);
6063
}
6164

6265
@Override
6366
public Compressor createCompressor() {
64-
return new Lz4Compressor(getBufferSize(conf));
67+
return new Lz4Compressor(bufferSize);
6568
}
6669

6770
@Override
6871
public Decompressor createDecompressor() {
69-
return new Lz4Decompressor(getBufferSize(conf));
72+
return new Lz4Decompressor(bufferSize);
7073
}
7174

7275
@Override
@@ -77,7 +80,7 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti
7780
@Override
7881
public CompressionInputStream createInputStream(InputStream in, Decompressor d)
7982
throws IOException {
80-
return new BlockDecompressorStream(in, d, getBufferSize(conf));
83+
return new BlockDecompressorStream(in, d, bufferSize);
8184
}
8285

8386
@Override
@@ -88,7 +91,6 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc
8891
@Override
8992
public CompressionOutputStream createOutputStream(OutputStream out, Compressor c)
9093
throws IOException {
91-
int bufferSize = getBufferSize(conf);
9294
return new BlockCompressorStream(out, c, bufferSize,
9395
CompressionUtil.compressionOverhead(bufferSize));
9496
}

‎hbase-compression/hbase-compression-snappy/src/main/java/org/apache/hadoop/hbase/io/compress/xerial/SnappyCodec.java

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -44,9 +44,11 @@ public class SnappyCodec implements Configurable, CompressionCodec {
4444
public static final String SNAPPY_BUFFER_SIZE_KEY = "hbase.io.compress.snappy.buffersize";
4545

4646
private Configuration conf;
47+
private int bufferSize;
4748

4849
public SnappyCodec() {
4950
conf = new Configuration();
51+
bufferSize = getBufferSize(conf);
5052
}
5153

5254
@Override
@@ -57,16 +59,17 @@ public Configuration getConf() {
5759
@Override
5860
public void setConf(Configuration conf) {
5961
this.conf = conf;
62+
this.bufferSize = getBufferSize(conf);
6063
}
6164

6265
@Override
6366
public Compressor createCompressor() {
64-
return new SnappyCompressor(getBufferSize(conf));
67+
return new SnappyCompressor(bufferSize);
6568
}
6669

6770
@Override
6871
public Decompressor createDecompressor() {
69-
return new SnappyDecompressor(getBufferSize(conf));
72+
return new SnappyDecompressor(bufferSize);
7073
}
7174

7275
@Override
@@ -77,7 +80,7 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti
7780
@Override
7881
public CompressionInputStream createInputStream(InputStream in, Decompressor d)
7982
throws IOException {
80-
return new BlockDecompressorStream(in, d, getBufferSize(conf));
83+
return new BlockDecompressorStream(in, d, bufferSize);
8184
}
8285

8386
@Override
@@ -88,7 +91,6 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc
8891
@Override
8992
public CompressionOutputStream createOutputStream(OutputStream out, Compressor c)
9093
throws IOException {
91-
int bufferSize = getBufferSize(conf);
9294
return new BlockCompressorStream(out, c, bufferSize,
9395
Snappy.maxCompressedLength(bufferSize) - bufferSize); // overhead only
9496
}

‎hbase-compression/hbase-compression-xz/src/main/java/org/apache/hadoop/hbase/io/compress/xz/LzmaCodec.java

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -44,9 +44,13 @@ public class LzmaCodec implements Configurable, CompressionCodec {
4444
public static final int LZMA_BUFFERSIZE_DEFAULT = 256 * 1024;
4545

4646
private Configuration conf;
47+
private int bufferSize;
48+
private int level;
4749

4850
public LzmaCodec() {
4951
conf = new Configuration();
52+
bufferSize = getBufferSize(conf);
53+
level = getLevel(conf);
5054
}
5155

5256
@Override
@@ -57,16 +61,18 @@ public Configuration getConf() {
5761
@Override
5862
public void setConf(Configuration conf) {
5963
this.conf = conf;
64+
this.bufferSize = getBufferSize(conf);
65+
this.level = getLevel(conf);
6066
}
6167

6268
@Override
6369
public Compressor createCompressor() {
64-
return new LzmaCompressor(getLevel(conf), getBufferSize(conf));
70+
return new LzmaCompressor(level, bufferSize);
6571
}
6672

6773
@Override
6874
public Decompressor createDecompressor() {
69-
return new LzmaDecompressor(getBufferSize(conf));
75+
return new LzmaDecompressor(bufferSize);
7076
}
7177

7278
@Override
@@ -77,7 +83,7 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti
7783
@Override
7884
public CompressionInputStream createInputStream(InputStream in, Decompressor d)
7985
throws IOException {
80-
return new BlockDecompressorStream(in, d, getBufferSize(conf));
86+
return new BlockDecompressorStream(in, d, bufferSize);
8187
}
8288

8389
@Override
@@ -88,7 +94,6 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc
8894
@Override
8995
public CompressionOutputStream createOutputStream(OutputStream out, Compressor c)
9096
throws IOException {
91-
int bufferSize = getBufferSize(conf);
9297
return new BlockCompressorStream(out, c, bufferSize,
9398
CompressionUtil.compressionOverhead(bufferSize));
9499
}

0 commit comments

Comments
 (0)