Status

Current state: "Under Discussion"

Discussion thread: https://lists.apache.org/thread/k9ydcdyxdyfgoy73cgsgs8w46yhpxrw2

JIRA: 

Authors: xinyu<xinyuzhou@automq.com>

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

In the cloud era, shared storage architecture is increasingly popular. Almost all cloud providers offer three types of shared storage services: block storage, file storage, and object storage. These services have many advantages over traditional local disks:

...

But running Kafka on shared storage services is challenging because its classic ISR-based storage engine is designed for local disks. For example, although Kafka can be deployed on file storage services like NetApp Files, it can't benefit from the durability and elasticity of shared file storage.

At AutoMQ, our goal is This KIP aims to make Kafka stateless and enable , allowing it to run seamlessly on shared storage systems , and transforming its architecture from shared nothing to shared storage. But this KIP does not plan to merge AutoMQ into Kafka directly, as it's a major decision for the Kafka community and would require significant resources.This KIP aims to introduce a It introduces a unified log layer for deploying Kafka on various shared storage mediums, with offering the following benefits:

• A unified log layer that supports both current replication-based local file storage and shared storage simultaneously.

• Greater flexibility for major users or vendors to deploy Kafka on their own shared storage, as many large companies have their own DFS or HDFS infrastructure.

• Prevent community fragmentation, allowing users or vendors to extend the storage layer without complex modifications to Kafka, fostering greater community collaboration.

...

Essentially, we can use shared storage like S3 or NFS to enhance Kafka's scalability and eliminate cross-AZ traffic. Luckily, the community has made significant efforts to reduce cross-AZ traffic for producers (KIP-1123: Rack-aware partitioning for Kafka Producer) and consumers (KIP-392: Allow consumers to fetch from closest replica, KIP-881: Rack-aware Partition Assignment for Kafka Consumers). With Kafka on shared storage, data replication is unnecessary, eliminating cross-AZ replication traffic. Therefore, this KIP will focus on creating an abstract log layer to facilitate building stateless Kafka on shared storage, enhancing scalability.

...

1. Refactor the log implementations by abstracting the log and log segment, allowing all managers like LogManager and LogCleaner to depend on the abstract log class instead of the specific UnifiedLog implementation.

2. Define a new abstraction to mask the differences between various shared storage services, called ' Stream. ' We will use Stream to implement a SharedLog that extends the AbstractLog. Just as a File API bridges the gap between LocalLog and a hard disk, Stream will bridge the gap between SharedLog and shared storage services like S3.

...

Currently, Kafka's storage engine heavily relies on UnifiedLog, which includes LocalLog and RemoteLog, and some logic even depends on local files. This makes current storage unsuitable for shared storage support.Image Removed

Image Added

This KIP proposes a new inheritance system for log classes:

1. AbstractLog and AbstractLogSegment are basic abstract classes containing common interfaces and logic methods for both bog file logs and shared logs.

2. All managers relying on UnifiedLog should be refactored to depend on AbstractLog.

3. UnifiedLog and LogSegment will implement AbstractLog and AbstractLogSegment. This is the core of the current classic storage engine.

4. A new log implementation, SharedLog and SharedLogSegment, will also depend on AbstractLog and AbstractLogSegment. We recommend using 'Stream' APIs for shared log implementation, which can then be applied to S3 or other storage services. However, 'Stream' is  is optional.

Optional Stream layer

...

A stream is truly append-only data streaming with extremely simple APIs. Similar to how a log is divided into LogSegments, a stream consists of multiple StreamSlices, each being a short section of the stream.

public interface Stream {
    long streamId();

    long streamEpoch();

    long startOffset();

    long confirmOffset();

    long nextOffset();

    CompletableFuture<AppendResult> append(AppendContext context, RecordBatch recordBatch);

    CompletableFuture<FetchResult> fetch(FetchContext context, long startOffset, long endOffset, int maxBytesHint);

    CompletableFuture<Void> trim(long newStartOffset);

    CompletableFuture<Void> close();
    
    CompletableFuture<Void> destroy();
}

...

Each partition now has a shared log with multiple streams, necessitating storage for relationship information to identify the meta stream's Stream ID for a specific partition. This KIP recommends using KRaft to store this data, requiring the addition of three KRaft controller APIs and a KV client implemented by KRaft.

public interface KVClient {
    CompletableFuture<Value> putKVIfAbsent(KeyValue keyValue);

    CompletableFuture<Value> putKV(KeyValue keyValue);

    CompletableFuture<Value> getKV(Key key);

    CompletableFuture<Value> delKV(Key key);
}

GET_KVS(ApiMessageType.GET_KVS, false, true),
PUT_KVS(ApiMessageType.PUT_KVS, false, true),
DELETE_KVS(ApiMessageType.DELETE_KVS, false, true),

...