...

Compression improves network and disk utilization and leads to better cluster throughput at the expense of latency. Several compression algorithms (and compression levels) exist which trade off compression ratio for speed. With the use of accelerators, we can speed up compression (improve latency) without affecting compression ratio too much. 
Kafka currently supports 4 compression algorithms; Gzip, Zstd, Snappy and LZ4. This KIP proposes a framework that supports the acceleration of any of these algorithms in Kafka with any hardware accelerator provider. This KIP does not introduce a new compression algorithm but instead allows for compression acceleration on producers, brokers or consumers. Compression, decompression and recompression can be independently accelerated. This means that the accelerated compression strategy MUST be compatible with existing strategies.
This KIP also introduces an accelerated GzipCompressor that uses the QAT (https://www.intel.com/content/www/us/en/products/docs/accelerator-engines/what-is-intel-qat.html) hardware accelerator

Public Interfaces

Compression Service

...

public interface CompressionService {

    /**
     * The compression type for this compression codec. 
     * @see org.apache.kafka.common.config.TopicConfig#COMPRESSION_TYPE_DOC        
     */
    CompressionTypeString type();

     public default boolean isAvailable() { return false; }/**
     * Throws UnavailableServiceException if service is not available
     */
    public void checkAvailable();

    /**
     * returns an OutputStream that will compress data compatible with the compression type.
     *
     * @param bufferStream The buffer to write the compressed data to
     * @param compressionLevel an optional parameter that sets the compression level if applicable
     */
    OutputStream compressedOutputStream(ByteBufferOutputStreamOutputStream bufferStream, Optional<Integer> compressionLevel);

    /**
     * returns an InputStream that will decompress data compatible with the compression type.
     *
     * @param buffer The {@link ByteBuffer} instance holding the data to decompress.
     * @param decompressionBufferSupplier The supplier of ByteBuffer(s) used for decompression if supported.
     */
    InputStream decompressedInputStream(ByteBuffer buffer, BufferSupplier decompressionBufferSupplier);
}

The following utility classes will be added to the public API for the convenience of service providers

org.apache.kafka.common.utils.ByteBufferOutputStream

org.apache.kafka.common.utils.ByteBufferInputStream

org.apache.kafka.common.utils.BufferSupplier

org.apache.kafka.common.record.CompressionType

BufferProvider decompressionBufferProvider);

	public interface BufferProvider {
        public ByteBuffer get(int capacity);
        public void release(ByteBuffer buffer);
    } 
}

Proposed Changes

This KIP leverages Java’s Service Loader API to offload compression/decompression for a specific codec to a provided alternative. Each existing compression codec represents a service. A service provider is any alternative implementation of the specific service that implements the CompressionService interface. On startup the service loader will attempt to load service providers for each configured service and will replace the default implementation with the provided implementation. 

...

A specific accelerated codec must be compatible with the codec it is accelerating in all directions. Since compression and decompression typically occur in separate processes, the accelerated codec need not be available in all nodes of the cluster for compression/decompression to occur correctly. This KIP does not affect the message format and is purely opportunistic; Compression and decompression will continue to operate normally even if no accelerators are present.

Due to the nature of the service loader API, any untrusted and untested jar file on the application class path could be dynamically loaded by the Kafka process. This introduces compatibility as well as security concerns

Test Plan

New unit tests and integration tests will be included as part of this feature. Tests must prove 

• Compatibility between an existing codec and a corresponding accelerated codec.
• The default behavior with this feature disabled is unchanged
• Correct error handling in failure scenarios
• Multiple compression service types can be correctly used and managed simultaneously
• Validate compression options (e.g compression levels) are honored were applicable 

Rejected Alternatives

See (KIP-984: Add pluggable compression interface to Kafka)