@Override
  protected int compressNative(ByteBuffer sourceBuffer, int sourceBufferOffset, int sourceDataSize,
      ByteBuffer compressedBuffer, int compressedBufferOffset, int compressedDataSize) throws CompressionException {

    // Zstd requires both buffers type be the same (either both direct memory or both heap), no mixed.
    if (sourceBuffer.isDirect() != compressedBuffer.isDirect()) {
      throw new IllegalArgumentException(
          "Cannot compress due to mismatch sourceBuffer type and compressedBuffer buffer type.");
    }

    // Invoke compress byte array or direct buffer based on buffer type.
    long compressedSize;
    if (sourceBuffer.isDirect()) {
      compressedSize =
          Zstd.compressDirectByteBuffer(compressedBuffer, compressedBufferOffset, compressedDataSize, sourceBuffer,
              sourceBufferOffset, sourceDataSize, getCompressionLevel());
    } else {
      compressedSize =
          Zstd.compressByteArray(compressedBuffer.array(), compressedBuffer.arrayOffset() + compressedBufferOffset,
              compressedDataSize, sourceBuffer.array(), sourceBuffer.arrayOffset() + sourceBufferOffset, sourceDataSize,
              getCompressionLevel());
    }

    // Check for error.
    if (Zstd.isError(compressedSize)) {
      throw new CompressionException(String.format("Zstd compression failed with error code: %d, name: %s. "
              + "sourceBuffer.limit=%d, sourceBufferOffset=%d, sourceDataSize=%d, "
              + "compressedBuffer.capacity=%d, compressedBufferOffset=%d, compressedDataSize=%d, compressionLevel=%d",
          compressedSize, Zstd.getErrorName(compressedSize), sourceBuffer.limit(), sourceBufferOffset, sourceDataSize,
          compressedBuffer.capacity(), compressedBufferOffset, compressedDataSize, getCompressionLevel()));
    }

    return (int) compressedSize;
  }
Invoke the Zstd compression algorithm given the source data, the compressed destination buffer and its offset.
The compression output is stored in the compressed buffer at the specific offset.
All parameters have been verified before calling so implementation can skip verification.

@param sourceBuffer The source uncompressed data.  It has already been null and empty verified.
@param sourceBufferOffset Offset in sourceData to start reading.
@param sourceDataSize Number of bytes in sourceData to compress.
@param compressedBuffer  The buffer to hold the compressed data.
@param compressedBufferOffset The offset in compressedBuffer where the compression output should write.
@param compressedDataSize Size of the buffer where compression can write to.  This size should be same as
                            estimateMaxCompressedDataSize().
@return The size of the compressed data in bytes.
ambry-commons/src/main/java/com/github/ambry/compression/ZstdCompression.java