(#765) Restore Hadoop ByteBuffer compression compatibility

src/main/java/org/xerial/snappy/Snappy.java

@@ -132,14 +132,15 @@ public static int compress(byte[] input, int inputOffset, int inputLength, byte[
     /**
      * Compress the content in the given input buffer. After the compression,
      * you can retrieve the compressed data from the output buffer [pos() ...
-     * limit()) (compressed data size = limit() - pos() = remaining())
+     * limit()) (compressed data size = limit() - pos() = remaining()). The output buffer's current limit does
+     * not restrict compression; its position is preserved and its limit is set to the end of the compressed data.
      *
      * @param uncompressed buffer[pos() ... limit()) containing the input data
-     * @param compressed output of the compressed data. Uses range [pos()..].
+     * @param compressed output of the compressed data. Uses range [pos()..capacity()).
      * @return byte size of the compressed data.
      * @throws SnappyError when the input is not a direct buffer
      * @throws IllegalArgumentException when the output buffer has less than {@link #maxCompressedLength(int)} bytes
-     * remaining
+     * available between its position and capacity
      */
     public static int compress(ByteBuffer uncompressed, ByteBuffer compressed)
             throws IOException
@@ -157,7 +158,7 @@ public static int compress(ByteBuffer uncompressed, ByteBuffer compressed)
         int uPos = uncompressed.position();
         int uLen = uncompressed.remaining();
         int cPos = compressed.position();
-        checkOutputSpace(compressed.remaining(), 0, maxCompressedLength(uLen));
+        checkOutputSpace(compressed.capacity(), cPos, maxCompressedLength(uLen));
         int compressedSize = impl.rawCompress(uncompressed, uPos, uLen, compressed,
                 cPos);
 

src/test/java/org/xerial/snappy/SnappyBoundsCheckTest.java

@@ -67,6 +67,71 @@ public void compressByteBufferToMaxCompressedLengthOutput()
         assertArrayEquals(data, result);
     }
 
+    private static void assertCompressionWithEmptyOutputLimit(ByteBuffer compressed, byte[] data)
+            throws Exception
+    {
+        int outputPosition = compressed.position();
+        compressed.limit(outputPosition);
+        int compressedSize = Snappy.compress(directBufferOf(data), compressed);
+        assertEquals(outputPosition, compressed.position());
+        assertEquals(outputPosition + compressedSize, compressed.limit());
+        ByteBuffer uncompressed = ByteBuffer.allocateDirect(data.length);
+        assertEquals(data.length, Snappy.uncompress(compressed, uncompressed));
+        byte[] result = new byte[data.length];
+        uncompressed.get(result);
+        assertArrayEquals(data, result);
+    }
+
+    @Test
+    public void compressByteBufferWithEmptyOutputLimit()
+            throws Exception
+    {
+        // Hadoop's SnappyCompressor sets the output limit to zero before compression.
+        byte[] data = randomBytes(100);
+        assertCompressionWithEmptyOutputLimit(ByteBuffer.allocateDirect(Snappy.maxCompressedLength(data.length)), data);
+    }
+
+    @Test
+    public void compressByteBufferWithEmptyOutputLimitAtOffset()
+            throws Exception
+    {
+        byte[] data = randomBytes(100);
+        ByteBuffer compressed = ByteBuffer.allocateDirect(40 + Snappy.maxCompressedLength(data.length));
+        compressed.position(40);
+        assertCompressionWithEmptyOutputLimit(compressed, data);
+    }
+
+    @Test
+    public void compressByteBufferWithEmptyOutputSliceLimit()
+            throws Exception
+    {
+        byte[] data = randomBytes(100);
+        ByteBuffer parent = ByteBuffer.allocateDirect(4096);
+        parent.position(40);
+        parent.limit(40 + Snappy.maxCompressedLength(data.length));
+        assertCompressionWithEmptyOutputLimit(parent.slice(), data);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void compressByteBufferWithInsufficientCapacityAfterPosition()
+            throws Exception
+    {
+        byte[] data = randomBytes(100);
+        ByteBuffer compressed = ByteBuffer.allocateDirect(Snappy.maxCompressedLength(data.length));
+        compressed.position(1);
+        Snappy.compress(directBufferOf(data), compressed);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void compressByteBufferToUndersizedOutputSlice()
+            throws Exception
+    {
+        ByteBuffer parent = ByteBuffer.allocateDirect(4096);
+        parent.position(40);
+        parent.limit(104);
+        Snappy.compress(directBufferOf(randomBytes(100)), parent.slice());
+    }
+
     @Test(expected = IllegalArgumentException.class)
     public void compressByteArrayToUndersizedOutput()
             throws Exception