Status

Current state: [Under Discussion]

Discussion thread: here []

JIRA: here []

Motivation

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

The 'CompressionService' interface encapsulates the basic functions that an accelerated codec must provide. A service provider will supply a specific implementation of this service for a specific 'compressionType'.

public interface CompressionService {

    /**
     * The compression type for this compression codec
     */
    CompressionType type();

    public default boolean isAvailable() { return false; }

    /**
     * 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(ByteBufferOutputStream 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

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. 

Compression will only be offloaded to a compression provider only if it is specified in the corresponding configuration file of the producer, server or consumer. If no service provider is configured explicitly then the base implementation will be used. A new configuration is added for each existing codec, 'compression.{codec}.provider' whose value is a class name of the service provider. A ConfigException will be thrown if the provided class is not found on the class path. 


 

Compatibility, Deprecation, and Migration Plan

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.

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

Rejected Alternatives

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