public static void compressAndDecompressTest(Compression compression, String testMessage,
      int bufferLeftPadSize, int bufferRightPadSize) throws CompressionException {

    byte[] testMessageBinary = testMessage.getBytes(StandardCharsets.UTF_8);
    int testMessageSize = testMessageBinary.length;
    int compressedBufferSize = compression.getCompressBufferSize(testMessageSize);
    int paddedCompressedBufferSize = bufferLeftPadSize + compressedBufferSize + bufferRightPadSize;

    // Apply compression to testMessage using Heap and Direct memory.
    // compressedBuffer consists of bytes[bufferLeftPadSize] + bytes[actualCompressedSize] + bytes[bufferRightPadSize]
    ByteBuffer sourceBufferHeap = ByteBuffer.wrap(testMessageBinary);
    ByteBuffer sourceBufferDirect = ByteBuffer.allocateDirect(testMessageBinary.length);
    sourceBufferDirect.put(testMessageBinary);
    ByteBuffer compressedBufferHeap = ByteBuffer.allocate(paddedCompressedBufferSize);
    ByteBuffer compressedBufferDirect = ByteBuffer.allocateDirect(paddedCompressedBufferSize);

    // Compress Heap source to Heap output.
    sourceBufferHeap.position(0);
    compressedBufferHeap.position(bufferLeftPadSize);
    int actualCompressedSize = compression.compress(sourceBufferHeap, compressedBufferHeap);
    Assert.assertTrue(actualCompressedSize > 0);
    Assert.assertEquals(0, sourceBufferHeap.remaining());
    Assert.assertEquals(testMessageSize, sourceBufferHeap.position());
    Assert.assertEquals(actualCompressedSize, compressedBufferHeap.position() - bufferLeftPadSize);

    int nextActualCompressedSize;
    if (!compression.requireMatchingBufferType()) {
      // Compress Heap source to Direct output.
      sourceBufferHeap.position(0);
      compressedBufferDirect.position(bufferLeftPadSize);
      nextActualCompressedSize = compression.compress(sourceBufferHeap, compressedBufferDirect);
      Assert.assertEquals(actualCompressedSize, nextActualCompressedSize);
      Assert.assertTrue(nextActualCompressedSize > 0);
      Assert.assertEquals(0, sourceBufferHeap.remaining());
      Assert.assertEquals(testMessageSize, sourceBufferHeap.position());
      Assert.assertEquals(nextActualCompressedSize, compressedBufferDirect.position() - bufferLeftPadSize);

      // Compress Direct source to Heap output.
      sourceBufferDirect.position(0);
      compressedBufferHeap.position(bufferLeftPadSize);
      nextActualCompressedSize = compression.compress(sourceBufferDirect, compressedBufferHeap);
      Assert.assertEquals(actualCompressedSize, nextActualCompressedSize);
      Assert.assertTrue(nextActualCompressedSize > 0);
      Assert.assertEquals(0, sourceBufferDirect.remaining());
      Assert.assertEquals(testMessageSize, sourceBufferDirect.position());
      Assert.assertEquals(actualCompressedSize, compressedBufferHeap.position() - bufferLeftPadSize);
    }

    // Compress Direct source to Direct output.
    sourceBufferDirect.position(0);
    compressedBufferDirect.position(bufferLeftPadSize);
    nextActualCompressedSize = compression.compress(sourceBufferDirect, compressedBufferDirect);
    Assert.assertEquals(actualCompressedSize, nextActualCompressedSize);
    Assert.assertTrue(nextActualCompressedSize > 0);
    Assert.assertEquals(0, sourceBufferDirect.remaining());
    Assert.assertEquals(testMessageSize, sourceBufferDirect.position());
    Assert.assertEquals(actualCompressedSize, compressedBufferDirect.position() - bufferLeftPadSize);

    // Apply decompression.
    // compressedBuffer has bytes[bufferLeftPadSize] + bytes[actualCompressedSize] + bytes[bufferRightPadSize]
    // decompressedBuffer contains bytes[bufferLeftPadSize] + bytes[sourceBufferSize] + bytes[bufferRightPadSize]
    compressedBufferHeap.limit(bufferLeftPadSize + actualCompressedSize);
    compressedBufferDirect.limit(bufferLeftPadSize + actualCompressedSize);
    int paddedDecompressedBufferSize = bufferLeftPadSize + testMessageSize + bufferRightPadSize;
    ByteBuffer decompressedBufferHeap = ByteBuffer.allocate(paddedDecompressedBufferSize);
    ByteBuffer decompressedBufferDirect = ByteBuffer.allocateDirect(paddedDecompressedBufferSize);
    byte[] decompressedMessage = new byte[testMessageSize];

    // Decompress Heap to Heap.
    compressedBufferHeap.position(bufferLeftPadSize);
    decompressedBufferHeap.position(bufferLeftPadSize);
    int decompressedSize = compression.decompress(compressedBufferHeap, decompressedBufferHeap);
    Assert.assertEquals(testMessageSize, decompressedSize);
    Assert.assertEquals(testMessageSize, decompressedBufferHeap.position() - bufferLeftPadSize);
    Assert.assertEquals(0, compressedBufferHeap.remaining());
    Assert.assertEquals(testMessage, new String(decompressedBufferHeap.array(), bufferLeftPadSize, testMessageSize, StandardCharsets.UTF_8));

    if (!compression.requireMatchingBufferType()) {
      // Decompress Heap to Direct.
      compressedBufferHeap.position(bufferLeftPadSize);
      decompressedBufferDirect.position(bufferLeftPadSize);
      decompressedSize = compression.decompress(compressedBufferHeap, decompressedBufferDirect);
      Assert.assertEquals(testMessageSize, decompressedSize);
      Assert.assertEquals(testMessageSize, decompressedBufferDirect.position() - bufferLeftPadSize);
      Assert.assertEquals(0, compressedBufferHeap.remaining());
      decompressedBufferDirect.position(bufferLeftPadSize);
      decompressedBufferDirect.get(decompressedMessage);
      Assert.assertEquals(testMessage, new String(decompressedMessage, StandardCharsets.UTF_8));

      // Decompress Direct to Heap.
      compressedBufferDirect.position(bufferLeftPadSize);
      decompressedBufferHeap.position(bufferLeftPadSize);
      decompressedSize = compression.decompress(compressedBufferDirect, decompressedBufferHeap);
      Assert.assertEquals(testMessageSize, decompressedSize);
      Assert.assertEquals(testMessageSize, decompressedBufferHeap.position() - bufferLeftPadSize);
      Assert.assertEquals(0, compressedBufferDirect.remaining());
      Assert.assertEquals(testMessage,
          new String(decompressedBufferHeap.array(), bufferLeftPadSize, testMessageSize, StandardCharsets.UTF_8));
    }

    // Decompress Direct to Direct.
    compressedBufferDirect.position(bufferLeftPadSize);
    decompressedBufferDirect.position(bufferLeftPadSize);
    decompressedSize = compression.decompress(compressedBufferDirect, decompressedBufferDirect);
    Assert.assertEquals(testMessageSize, decompressedSize);
    Assert.assertEquals(testMessageSize, decompressedBufferDirect.position() - bufferLeftPadSize);
    Assert.assertEquals(0, compressedBufferDirect.remaining());
    decompressedBufferDirect.position(bufferLeftPadSize);
    decompressedBufferDirect.get(decompressedMessage);
    Assert.assertEquals(testMessage, new String(decompressedMessage, StandardCharsets.UTF_8));
  }
Test compression using compress() and decompress() APIs that uses an internal structure.
It tests a combination of heap buffers and direct memory as input and output.
ambry-commons/src/test/java/com/github/ambry/compression/LZ4CompressionTest.java