...

• Diskless Topic: A topic which does not append directly to or serve from block storage devices, and does not utilize direct replication.
• Classic Topic: A topic which appends directly to and serves data only from block storage devices and utilizes direct replication, the present Kafka status quo.
• Tiered Topic: A classic topic using KIP-405 Tiered Storage to move inactive segments off of block storage, but still appends to and serves data from block storage devices.
• Diskless-Ready Cluster: An Apache Kafka cluster which is configured to host Diskless Topics in addition to Classic and Tiered topics.
• Broker Disk: A local block device available for a broker to store log directories on. May be direct-attached or network-attached.
• Object Storage: A shared, durable, concurrent, and eventually consistent storage supporting arbitrary sized byte values and a minimal set of atomic operations: put, delete, list, and ranged get.
• Object Key: A unique reference to an object within Object Storage.
• Broker/Client Rack: A hint provided to Apache Kafka brokers and clients about the underlying infrastructure topology. Data transfers within a rack are assumed to have lower costs and lower latency than transfers between racks. Failures are assumed to be less correlated across racks, so durable information is persisted across multiple racks. When Apache Kafka is deployed in a hyperscaler, this is typically synonymous with “Region”, “Availability Zone”, or “Zone”.
• Batch: A container for Kafka records and a unit of record representation in the network protocol and storage format in both status quo Kafka classic and diskless topics.
• Shared Log WAL Segment Object: An object containing a shared log segment batches for one or more diskless topic-partitions on the object storage. Contains record batches similar to classic Kafka topics.
• Batch Coordinate: A reference to a record batch within a shared log segment objectWAL Segment, at some byte range.
• Diskless Batch Coordinator: Component serving as source of truth about batch coordinates and log segment objectsWAL Segments. Establishes a global order of batches within a diskless topic, and serves coordinates to enable retrieval of batches.
• Hyperscaler/Cloud: Widely available cloud computing vendors offering compute, storage, and networking solutions.
• Object Compaction: Distinct from log compaction. A background asynchronous process which reads from and writes to multiple shared log segment objects to manage already-written objects.

Proposed Changes

Overview

Proposed Changes

Overview

The KIP proposes The KIP proposes a new kind of topic in Kafka called diskless topic, which have the following properties and immediate effects:

1. Data in diskless topics is durably stored solely in object storage , and not in at all times. 
   1. Local segments on broker disks
   . 
2. Classic concepts of log directories, segment rolling, object writing and deletion, message flushing, tiered storage, etc. and the related APIs do not apply. Effectively, brokers will not create any files related to diskless topics.
3. 1. serve as caches and not sources of truth
   2. Remote storage may have higher latency than local disks, increasing the latency of
Remote storage may have higher latency than local disks, increasing the latency of
4. 1. Kafka requests and end-to-end data latency.
5. Kafka delegates replication of diskless topics to object storage, and does not perform replication itself.
   1. Classic concepts of
replicas,
6. 1. replication, ISR,
partition assignment and the related APIs
7. 1. and unclean leader elections do not apply.
   2. All operators can use efficient types of storage backends, such as ones with erasure coding.
   3. Hyperscaler operators can avoid most inter-zone data replication costs.
8. All brokers are capable of interacting with all diskless topics, and no broker is uniquely considered the leader of a partition.
   1. Classic replica leader election
is not applicable to
9. 1. for diskless topics does not affect produce request functionality, and all requests may use any broker as if it were the leader.
   2. Clusters are able to perform fine-grained client balancing across the cluster independently of topic/partition hot spots.
   3. Any broker may build a replica of any set of diskless partitions by contacting the batch coordinator, without any other replicas being live.
   4. Hyperscaler operators can avoid most inter-zone data ingress/egress costs.

This KIP allows a non-fundamental functional limitation between diskless and classic topics: Diskless topics will not immediately support compaction.

The preliminary research suggests that these limitations are not fundamental and can be resolved. They are out of scope of this KIP, and are expected to be addressed in follow-up KIPs. See the Further Work section for more details.

Similar to Tiered Topics, Diskless Topics will not immediately support compaction. In the future, once Compaction for Tiered Topics is designed and implemented, Diskless topics will immediately support compaction.

In other functional In other functional aspects, diskless topics are indistinguishable from classic topics. This includes durability guarantees, ordering guarantees, transactional and non-transactional producer API, consumer API, consumer groups, share groups, data retention, and authorization mechanisms. Streaming applications can be moved from classic to diskless topics without changing client versions, application code, or experiencing correctness problems.

Because a single cluster can host diskless, classic, and tiered topics together, application developers are able to trade off latency and cost on a per-topic basis, while not incurring additional operational overhead of multiple clusters.

Data Flow

Diskless Topics differ from Classic Topics primarily in how appends are invoked on each replica.

Image Added

Rather than appending to the leader immediately and waiting for fetches from replicas to initiate appends on all replicas, Diskless topics write data to Object Storage. This data is then retrieved by replicas, and finally appended to local segments.

In order to satisfy the requirement In order to satisfy the requirement of multiple brokers writing to a single partition with a total order, it is necessary to decouple the handling of batch data and batch metadata. This permits low-cost replication of batch data with fewer consistency guarantees, while processing batch metadata and the state of the partition with strong consistency guarantees.

...

For diskless topics, the Batch Coordinator is now responsible for offset/timestamp assignment (2), and the fetch broker each replica is responsible for injecting the offsets into the batch data (3). These components provide the effect that fetched data sent to the consumer has the same structure as data appended to a classic segment.

Image RemovedImage Added

In diskless topics, we separate storage I/O from batch ordering. The data being persisted to object storage has no implicit order, and ordering is delegated to the batch coordinates. Coordinates passed from each broker to the batch coordinator are locally ordered, and have an order only relative to other coordinates passed from that same broker. The batch coordinator performs the append and assigns a global order to the batch coordinates in each partition.

Data Format and

...

WAL Segment Objects

The main unit of storage in Kafka is a record batch, and is the smallest unit of data which may be produced or consumed. Kafka groups these batches together into contiguous files to more efficiently manipulate them in the backing storage, while still preserving access to the individual batches. This remains true in diskless topics.

Batches for classic topics are grouped into Segments, which each contain batches from only a single partition, and batches in log segments always have offsets and timestamps applied. In contrast, data from multiple diskless topics and partitions are mixed in single objects, called shared log segmentsWAL Segments. This is necessary to keep the object storage write operation costs reasonable.

The batches within a shared log segment WAL Segment are ordered by their layout in the object, but do not have any inherent order relative to other shared log segmentsWAL Segments. Their logical offsets and append timestamps may be unset.

By locating data from multiple partitions together, shared log segments WAL Segments prevent the number of topic-partitions from influencing object read and write behavior, and reduce the total number of individual reads and writes to the backing storage.

...

Batches are grouped by partition into contiguous parts of the object. This improves the locality of later operations in the same way as segments do for classic topics. Shared log segment WAL Segment objects are given globally unique (e.g. UUID) names and uploaded without coordination or conflicts.

The shared log segment WAL Segment objects are immutable, similar to inactive segments in classic and tiered topics. Once an object is uploaded, its content is associated with its name for the whole lifetime of the object. This also aligns with the typical interface of object storage where random writes to an object are not supported.

...

The concrete details of the asynchronous garbage collection process is out of scope of this KIP and will be described in a separate KIP.

Consume path

Data is consumed from diskless topics in the following way:

1. The consumer sends a Fetch request to the broker.
2. The broker queries the Batch Coordinator for the relevant batch coordinates.
3. The broker gets the data either from the object storage and/or from the cache.
4. The broker injects the computed offsets and timestamps into the batches.
5. The broker constructs and sends the Fetch response to the Consumer.

Image Removed

Depending on the structure of the shared log segment, a single Fetch request could be satisfied by one GET request, or may need multiple requests. In order to improve the locality of requests, Diskless shared log segments will be periodically compacted in order to increase sequential spans of batches. The compaction is discussed in detail in KIP-1165: Object Compaction for Diskless.

Caching in local segments

Once data is persisted in WAL segments and the batch coordinator, the next step is to populate a cache with the data in its finalized order as determined by the batch coordinator. Each broker will be assigned as a replica for a subset of partitions in the cluster, using the existing KRaft metadata mechanisms. The broker will issue a read operation for each object containing at least one replicated partition. Once the read completes, the broker will inject record offsets as determined by the batch coordinator, and append the finalized batch to a local log segment on-disk. This will build the typical indexes used to optimize fetch queries and interactions with segments.

Distinct from Classic topics, these local segments are not required to be durable, as durable storage of data is delegatedrestricted to the batch coordinator and WAL segments.

Consume path

Data is consumed from Diskless topics from the local segments on-disk. This is essentially the same as consuming from a follower for classic topics. It also permits use of the kernel page cache, reducing memory pressure from the consume path to be allocated for the produce path buffers, and uses indexes built during the cache building step. This decouples the Fetch request rate from the batch coordinator operation rate.

Moving to Tiered Storage

Image Added

In order to control the number of many small WAL files, the batch coordinator memory size, and the rebuild time of local segments after a replica change, data is moved to Tiered Storage using the existing KIP-405 interfaces. Existing implementations of tiered storage RSM and RLMM can be used as-is, with no knowledge of the presence or absence of the Diskless produce path in front.

In order to use the existing interfaces, one broker per partition is chosen as the leader. This leader is permitted to copy segments to tiered storage. Replicas are responsible for serving fetch requests from tiered segments, similar to classic topics.

Effectively, Diskless becomes a front-end for Local Segments, and Tiered Storage becomes the back-end, providing the same benefits for optimizing storage usage for both Diskless and Classic topicsObject storage implementations are expected to implement caching to reduce the quantity of overall requests to the underlying storage. These caches should be partitioned per-rack, to maintain good locality for requests. Caching is discussed in detail in a follow-up KIP.

Batch deletion

There are various reasons a batch may require deletion:

• the local size and/or time retention settings of the topic;
• the partition truncation by the user;
• deletion of a topic.

...