HADOOP-15822. zstd compressor can fail with a small output buffer. Contributed by Jason Lowe.

(cherry picked from commit 8f97d6f2cd)

dev-support/docker/Dockerfile

@@ -63,6 +63,7 @@ RUN apt-get -q update \
         libsnappy-dev \
         libssl-dev \
         libtool \
+        libzstd1-dev \
         locales \
         make \
         pinentry-curses \

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zstd/ZStandardDecompressor.java

@@ -262,8 +262,8 @@ public class ZStandardDecompressor implements Decompressor {
 
     int originalPosition = dst.position();
     int n = inflateBytesDirect(
-        src, src.position(), src.remaining(), dst, dst.position(),
-        dst.remaining()
+        src, src.position(), src.limit(), dst, dst.position(),
+        dst.limit()
     );
     dst.position(originalPosition + n);
     if (bytesInCompressedBuffer > 0) {

hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/compress/zstd/ZStandardCompressor.c

@@ -195,11 +195,14 @@ JNIEXPORT jint Java_org_apache_hadoop_io_compress_zstd_ZStandardCompressor_defla
     ZSTD_inBuffer input = { uncompressed_bytes, uncompressed_direct_buf_len, uncompressed_direct_buf_off };
     ZSTD_outBuffer output = { compressed_bytes, compressed_direct_buf_len, 0 };
 
-    size_t size = dlsym_ZSTD_compressStream(stream, &output, &input);
+    size_t size;
+    if (uncompressed_direct_buf_len != 0) {
+        size = dlsym_ZSTD_compressStream(stream, &output, &input);
         if (dlsym_ZSTD_isError(size)) {
             THROW(env, "java/lang/InternalError", dlsym_ZSTD_getErrorName(size));
             return (jint) 0;
         }
+    }
     if (finish && input.pos == input.size) {
         // end the stream, flush and  write the frame epilogue
         size = dlsym_ZSTD_endStream(stream, &output);

hadoop-common-project/hadoop-common/src/main/native/src/org/apache/hadoop/io/compress/zstd/ZStandardDecompressor.c

@@ -178,6 +178,7 @@ JNIEXPORT jint JNICALL Java_org_apache_hadoop_io_compress_zstd_ZStandardDecompre
         return (jint) 0;
     }
     uncompressed_bytes = ((char*) uncompressed_bytes) + uncompressed_direct_buf_off;
+    uncompressed_direct_buf_len -= uncompressed_direct_buf_off;
 
     ZSTD_inBuffer input = { compressed_bytes, compressed_direct_buf_len, compressed_direct_buf_off };
     ZSTD_outBuffer output = { uncompressed_bytes, uncompressed_direct_buf_len, 0 };

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/zstd/TestZStandardCompressorDecompressor.java

@@ -414,14 +414,12 @@ public class TestZStandardCompressorDecompressor {
     outBuf.clear();
     while (!decompressor.finished()) {
       decompressor.decompress(inBuf, outBuf);
-      if (outBuf.remaining() == 0) {
       outBuf.flip();
       while (outBuf.remaining() > 0) {
         assertEquals(expected.get(), outBuf.get());
       }
       outBuf.clear();
     }
-    }
     outBuf.flip();
     while (outBuf.remaining() > 0) {
       assertEquals(expected.get(), outBuf.get());