This is an automated email from the ASF dual-hosted git repository.
aajisaka pushed a commit to branch branch-3.2.3
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/branch-3.2.3 by this push:
new 65783db HDFS-14099. Unknown frame descriptor when decompressing
multiple frames (#3836)
65783db is described below
commit 65783db095bc4370c6a33408436de09f09096e88
Author: Ashutosh Gupta <[email protected]>
AuthorDate: Tue Dec 28 18:14:38 2021 +0530
HDFS-14099. Unknown frame descriptor when decompressing multiple frames
(#3836)
Co-authored-by: xuzq <[email protected]>
Co-authored-by: Ashutosh Gupta <[email protected]>
Signed-off-by: Akira Ajisaka <[email protected]>
(cherry picked from commit caab29ec889a0771191b58714c306439b2415d91)
(cherry picked from commit 6535a183b22dc8915cea9a93ab8c0370ef9bd2f0)
(cherry picked from commit 5aeaa368efb6a3b847013b411b352eb3fcbd3ac0)
---
.../io/compress/zstd/ZStandardDecompressor.java | 13 +++++
.../zstd/TestZStandardCompressorDecompressor.java | 59 ++++++++++++++++++++++
2 files changed, 72 insertions(+)
diff --git
a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardDecompressor.java
b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardDecompressor.java
index bc9d29c..adf2fe6 100644
---
a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardDecompressor.java
+++
b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardDecompressor.java
@@ -113,6 +113,12 @@ public class ZStandardDecompressor implements Decompressor
{
compressedDirectBuf.put(
userBuf, userBufOff, bytesInCompressedBuffer);
+ // Set the finished to false when compressedDirectBuf still
+ // contains some bytes.
+ if (compressedDirectBuf.position() > 0 && finished) {
+ finished = false;
+ }
+
userBufOff += bytesInCompressedBuffer;
userBufferBytesToConsume -= bytesInCompressedBuffer;
}
@@ -186,6 +192,13 @@ public class ZStandardDecompressor implements Decompressor
{
0,
directBufferSize
);
+
+ // Set the finished to false when compressedDirectBuf still
+ // contains some bytes.
+ if (remaining > 0 && finished) {
+ finished = false;
+ }
+
uncompressedDirectBuf.limit(n);
// Get at most 'len' bytes
diff --git
a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/zstd/TestZStandardCompressorDecompressor.java
b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/zstd/TestZStandardCompressorDecompressor.java
index dcfb7e9..4c629be 100644
---
a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/zstd/TestZStandardCompressorDecompressor.java
+++
b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/zstd/TestZStandardCompressorDecompressor.java
@@ -234,6 +234,65 @@ public class TestZStandardCompressorDecompressor {
}
}
+ /**
+ * Verify decompressor logic with some finish operation in compress.
+ */
+ @Test
+ public void testCompressorDecompressorWithFinish() throws Exception {
+ DataOutputStream deflateOut = null;
+ DataInputStream inflateIn = null;
+ int byteSize = 1024 * 100;
+ byte[] bytes = generate(byteSize);
+ int firstLength = 1024 * 30;
+
+ int bufferSize = IO_FILE_BUFFER_SIZE_DEFAULT;
+ try {
+ DataOutputBuffer compressedDataBuffer = new DataOutputBuffer();
+ CompressionOutputStream deflateFilter =
+ new CompressorStream(compressedDataBuffer, new
ZStandardCompressor(),
+ bufferSize);
+
+ deflateOut =
+ new DataOutputStream(new BufferedOutputStream(deflateFilter));
+
+ // Write some data and finish.
+ deflateOut.write(bytes, 0, firstLength);
+ deflateFilter.finish();
+ deflateOut.flush();
+
+ // ResetState then write some data and finish.
+ deflateFilter.resetState();
+ deflateOut.write(bytes, firstLength, firstLength);
+ deflateFilter.finish();
+ deflateOut.flush();
+
+ // ResetState then write some data and finish.
+ deflateFilter.resetState();
+ deflateOut.write(bytes, firstLength * 2, byteSize - firstLength * 2);
+ deflateFilter.finish();
+ deflateOut.flush();
+
+ DataInputBuffer deCompressedDataBuffer = new DataInputBuffer();
+ deCompressedDataBuffer.reset(compressedDataBuffer.getData(), 0,
+ compressedDataBuffer.getLength());
+
+ CompressionInputStream inflateFilter =
+ new DecompressorStream(deCompressedDataBuffer,
+ new ZStandardDecompressor(bufferSize), bufferSize);
+
+ inflateIn = new DataInputStream(new BufferedInputStream(inflateFilter));
+
+ byte[] result = new byte[byteSize];
+ inflateIn.read(result);
+ assertArrayEquals(
+ "original array not equals compress/decompressed array", bytes,
+ result);
+ } finally {
+ IOUtils.closeQuietly(deflateOut);
+ IOUtils.closeQuietly(inflateIn);
+ }
+ }
+
@Test
public void testZStandardCompressDecompressInMultiThreads() throws Exception
{
MultithreadedTestUtil.TestContext ctx =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
