From 45434c93e865386de8c5f4036e795a3362557d56 Mon Sep 17 00:00:00 2001 From: Stephen Jung Date: Tue, 10 Nov 2020 11:38:22 -0800 Subject: [PATCH] HADOOP-17096. Fix ZStandardCompressor input buffer offset (#2104). Contributed by Stephen Jung (Stripe). --- .../apache/hadoop/io/compress/zstd/ZStandardCompressor.java | 4 ++-- .../org/apache/hadoop/io/compress/zstd/ZStandardCompressor.c | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.java index 960083519d..8b1b6db086 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.java @@ -159,7 +159,7 @@ public boolean needsInput() { } // have we consumed all input - if (keepUncompressedBuf && uncompressedDirectBufLen > 0) { + if (keepUncompressedBuf && uncompressedDirectBufLen - uncompressedDirectBufOff > 0) { return false; } @@ -223,7 +223,7 @@ public int compress(byte[] b, int off, int len) throws IOException { compressedDirectBuf.limit(n); // Check if we have consumed all input buffer - if (uncompressedDirectBufLen <= 0) { + if (uncompressedDirectBufLen - uncompressedDirectBufOff <= 0) { // consumed all input buffer keepUncompressedBuf = false; uncompressedDirectBuf.clear(); diff --git a/hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.c b/hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.c index 41eb9e2c85..6581f292b4 100644 --- a/hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.c +++ b/hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.c @@ -219,13 +219,13 @@ JNIEXPORT jint Java_org_apache_hadoop_io_compress_zstd_ZStandardCompressor_defla return (jint) 0; } - bytes_read += input.pos; + bytes_read += input.pos - uncompressed_direct_buf_off; bytes_written += output.pos; (*env)->SetLongField(env, this, ZStandardCompressor_bytesRead, bytes_read); (*env)->SetLongField(env, this, ZStandardCompressor_bytesWritten, bytes_written); (*env)->SetIntField(env, this, ZStandardCompressor_uncompressedDirectBufOff, input.pos); - (*env)->SetIntField(env, this, ZStandardCompressor_uncompressedDirectBufLen, input.size - input.pos); + (*env)->SetIntField(env, this, ZStandardCompressor_uncompressedDirectBufLen, input.size); return (jint) output.pos; }