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

(cherry picked from commit 8f97d6f2cdfccefba5457ae3d561e9ce0109da3f)
(cherry picked from commit 63396beab4178f3855be27c869bfb78336c71166)

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,10 +195,13 @@ 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);
-    if (dlsym_ZSTD_isError(size)) {
-        THROW(env, "java/lang/InternalError", dlsym_ZSTD_getErrorName(size));
-        return (jint) 0;
+    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

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,13 +414,11 @@ 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());
       }
+      outBuf.clear();
     }
     outBuf.flip();
     while (outBuf.remaining() > 0) {