...

Data in diskless topics is durably stored in object storage at all times.
1. Local segments on broker disks serve as caches and not sources of truth
2. Remote storage may have higher latency than local disks, increasing the latency of Kafka requests and end-to-end data latency.
Kafka delegates replication of diskless topics to object storage, and does not perform replication itself.
1. Replicas placement is still used to control client traffic and cache placement
2. Any broker may build a replica of any set of diskless partitions by contacting the diskless coordinator, lowering load on other brokers and eliminating unclean leader elections.
3. All operators can use efficient types of storage backends, such as ones with erasure coding.
4. Hyperscaler operators can avoid most inter-zone data replication costs.
All brokers are capable of interacting with all diskless topics, and produce requests do not need to be handled by the partition leader.
1. Produce requests are preferentially served by replicas of the partition, and do not need to be directed to the partition leader.
2. Partition leaders are still elected to upload to manage the ISR state, upload to tiered storage, and handle share fetches.
3. Clusters are able to perform fine-grained client balancing across the cluster independently of topic/partition hot spots.
4. Hyperscaler operators can avoid most inter-zone data ingress/egress costs.

...

New replicas are added to the partition's replica set, and begin building local segments from object storage.
The controller waits for the new replicas to become in-sync.
Old replicas are removed from the partition's replica set.

...

Diskless will rely on the existing follower fetching mechanism (broker returning PreferredReadReplica) in order to allow consumers to read from replicas in their racks and avoid inter-rack network traffic costs. In contrast to classic topics, non-ISR replicas will not be excluded. Instead, they will be deprioritezed and used only if no in-sync replicas are available in the client rack.

...

When the replica is not in-sync, i.e. there's a long gap between the log end in the Diskless Coordinator and the local log.
When the requested offset is older than the earliest offset available locally.

A direct read from remote storage involves:

The broker determines a fetch for a partition cannot be served from local segments
The broker queries the Diskless Coordinator for the relevant batch coordinates.
The broker gets the data either from the object storage.
The broker injects the computed offsets and timestamps into the batches in-memory.
The broker constructs and sends the Fetch response to the Consumer.

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.

Image Modified

Queues

In contrast to normal Fetch requests, ShareFetch requests can only be served by the share-partition leader, which is co-located with the partition leader and manages internal state, such as record state.

...

The concrete details of this feature are out of scope of this KIP and will be described in a separate 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.

Since objects are immutable, logical batch deletion doesn’t change the object content. For each object, it’s necessary to track its effectively used size. This size decreases by the batch size when a batch is deleted from the object. When the used size becomes 0, the object could be actually deleted.

Image Removed

Object deletion will be implemented as an asynchronous operation. First, the object is marked for deletion by the diskless coordinator and a broker performs the actual deletion from the object storage. This has two advantages:

The Diskless Coordinator doesn’t need to be concerned about the object storage, to have credentials and the code to access it.
The deletion could have a grace period. The grace period is a useful way to allow potentially in-progress Fetch requests to successfully finish.

To enforce the time and size retention settings of topics, a background process will periodically check metadata of diskless topics and logically delete affected batches.

It’s quite possible that for compliance reasons, a particular batch has a deadline for physical deletion. Despite the batch being logically deleted, its data are located in an object that may be kept alive by other batches, potentially forever. Therefore, batches with physical deletion deadlines must be either moved to isolated files during merging, or additional merge passes will be necessary to physically delete the data. This problem is addressed in KIP-1165.

Public Interfaces

Plugin Interfaces: Storage Backend

This interface abstracts the access methods to the object storage.

Code Block

language	java

public record ByteRange(long offset, long size) {}

public interface ObjectStorage extends Configurable {
    /**
     * Uploads an object to object storage.
     * <p>An exception must be thrown in case the number of bytes streamed from
     * {@code inputStream} is different from {@code length}.
     * @param key                      key of the object to upload.
     * @param inputStream              data of the object that will be uploaded.
     * @param length                   length of the data that will be uploaded.
     * @throws StorageBackendException if there are errors during the upload.
     */
    void upload(ObjectKey key, InputStream inputStream, long length)
        throws StorageBackendException;


    /**
     * Fetch a range of an object from object storage.
     * <p>This result should be cached for later access
     * @param key                      key of the object to fetch.
     * @param range                    range of bytes within the object.
     * @throws StorageBackendException if there are errors during the fetch.
     */
    InputStream fetch(ObjectKey key, ByteRange range)
        throws StorageBackendException;

    /**
     * Idempotently delete multiple objects.
     * <p>If an object doesn't exist, the operation should still succeed	
     * to be idempotent.
     * @param keys                     keys of objects to delete.
     * @throws StorageBackendException if there are other errors which prevent deletion.
     */
    void delete(Set<ObjectKey> keys)
        throws StorageBackendException;
}

Apache Kafka will not provide a production-grade implementation of this interface, because this would require Kafka depending on third party storage drivers. The list of potential storage is big and it’s not possible to provide implementations for all of them. Implementations will be provided by 3rd party developers in the Kafka ecosystem.

Configurations

Broker Configurations

diskless.system.enable: enables diskless support on a broker.
diskless.storage.class.name: the object storage class name for diskless topics.
diskless.storage.class.path: the object storage class path for diskless topics.
diskless.append.commit.interval.ms: defines how long to wait before closing a shared log segment object to be uploaded/committed.
diskless.append.buffer.max.bytes: defines the maximum size of a shared log segment object before closing it for further upload/commit.

Topic Configurations

diskless.enable: Sets a topic as a diskless topic. Diskless topics can only be defined at creation time (updating a classic topic to a diskless topic is out of scope.)

Client Configurations

Producer:

client.rack: Similar to Consumer's usage of client rack on the follower fetching feature, it defines the producer rack to align with brokers on the same rack.

Broker API

Diskless Coordinator

A new set of APIs will be required to allow brokers to contact the diskless coordinator.

These new requests are out of scope for this KIP, and will be fully defined in KIP-1164: Diskless Coordinator

Client Metadata

We propose to create the version 14 of the Metadata API with the following definition:

MetadataRequest:

Code Block

language	js

{
 "apiKey": 3,
 "type": "request",
 "listeners": ["broker"],
 "name": "MetadataRequest",
 "validVersions": "0-14",
 "flexibleVersions": "9+",
 "fields": [
   // In version 0, an empty array indicates "request metadata for all topics."  In version 1 and
   // higher, an empty array indicates "request metadata for no topics," and a null array is used to
   // indicate "request metadata for all topics."
   //
   // Version 2 and 3 are the same as version 1.
   //
   // Version 4 adds AllowAutoTopicCreation.
   //
   // Starting in version 8, authorized operations can be requested for cluster and topic resource.
   //
   // Version 9 is the first flexible version.
   //
   // Version 10 adds topicId and allows name field to be null. However, this functionality was not implemented on the server.
   // Versions 10 and 11 should not use the topicId field or set topic name to null.
   //
   // Version 11 deprecates IncludeClusterAuthorizedOperations field. This is now exposed
   // by the DescribeCluster API (KIP-700).
   // Version 12 supports topic Id.
   // Version 13 supports top-level error code in the response.
   // Version 14 supports Diskless and allows to send client.rack
   { "name": "Topics", "type": "[]MetadataRequestTopic", "versions": "0+", "nullableVersions": "1+",
     "about": "The topics to fetch metadata for.", "fields": [
     { "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true, "about": "The topic id." },
     { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName", "nullableVersions": "10+",
       "about": "The topic name." }
   ]},
   { "name": "AllowAutoTopicCreation", "type": "bool", "versions": "4+", "default": "true", "ignorable": false,
     "about": "If this is true, the broker may auto-create topics that we requested which do not already exist, if it is configured to do so." },
   { "name": "IncludeClusterAuthorizedOperations", "type": "bool", "versions": "8-10",
     "about": "Whether to include cluster authorized operations." },
   { "name": "IncludeTopicAuthorizedOperations", "type": "bool", "versions": "8+",
     "about": "Whether to include topic authorized operations." },
   { "name": "RackId", "type":  "string", "versions": "14+", "default": "", "ignorable": true,
     "about": "Rack ID of the client making this request."}
 ]
}

Here, RackId is the new field.

MetadataResponse:

Cluster Metadata

Any broker is able to serve Produce and Fetch requests for any Diskless topics. However, in many cases arbitrary selection of the request receiver by the client will not be optimal from the performance point of view. We need a way to let the clients choose brokers to serve their Produce and Fetch requests optimally. Only the partition leader can serve ShareFetch requests. Taking into account these requirements and also ones that other KIPs from the Diskless initiative (e.g. KIP-1164) may have, we propose the following.

We introduce the version 14 of MetadataRequest and MetadataResponse, which will allow the clients to send brokers their rack.id and receive back more information about Diskless topics, such as whether the topic is Diskless (IsDiskless) and what is/are recommended brokers to produce to a particular Diskless partition (PreferredProduceBrokers). The definition of these is provided below in the Public Interfaces section.

Newer clients will use the new API version 14 to send and receive this information. No behavior changes expected regarding classic topics. For Diskless topics, newer clients will do the following:

Send Produce requests to recommended produce brokers (in order and subject to availability).
Send Fetch requests as for classic topics: either to the leader or to the closest replica based on the rack.
Send ShareFetch requests as for classic topics, to the leader.

Older clients will use previous API versions 0-13 and the broker will not be able to tell them whether the topic is Diskless and what are the recommended brokers for producing. To allow Kafka users with older clients and also with third-party clients (which may have their own schedule of adding support for new features) to benefit from Diskless, we will allow them to add “,diskless_rack_id=<rack_id>” to their client ID. The Client ID is always transferred to the broker and the broker will take this information into account. It will respond with the matching version of the MetadataResponse, however, it will modify the true partition metadata for Diskless partitions. If the partition has replicas in the matching rack, the LeaderId will be replaced with one of them. Thus, this client will be able to produce and consume data within the rack and avoid inter-rack network traffic costs. This, however, will prevent ShareFetch requests from being served, because only the real leader can serve them. The way out of this is to specify diskless_rack_id for producers and normal consumers and don’t specify for share consumers. The latter must be separate instances in the code.

If an older client doesn’t provide diskless_rack_id to the broker, it will receive the true metadata for Diskless partitions and will be able to use the full spectrum of Kafka API with them, albeit not necessarily optimally from the point of view of inter-rack traffic reduction.

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

Since objects are immutable, the coordinator metadata will allow logical batch deletion, without changing the object content. To enforce the time and size retention settings of topics, a background process within the diskless coordinator will periodically check metadata of diskless topics and logically delete affected batches. When topic retention settings cause segments to be written to Tiered Storage, the batches contained in those segments are also logically deleted from the diskless coordinator.

For each object, it’s necessary to track its effectively used size. This size decreases by the batch size when a batch is deleted from the object. When the used size becomes 0, the object could be actually deleted.

Image Added

Object deletion will be implemented as an asynchronous operation. First, the object is marked for deletion by the diskless coordinator and a broker performs the actual deletion from the object storage. This has two advantages:

The Diskless Coordinator doesn’t need to be concerned about the object storage, to have credentials and the code to access it.
The deletion could have a grace period. The grace period is a useful way to allow potentially in-progress Fetch requests to successfully finish.
A single object may contain live batches for other coordinators, and the node performing the deletion must contact other coordinators in order to prove the object is safe to delete.
Nodes can periodically reconcile the list of objects in the storage to the list of WAL Segments in the diskless coordinators, and delete orphaned objects that were not properly committed at some earlier time.

It’s quite possible that for compliance reasons, a particular batch has a deadline for physical deletion. Despite the batch being logically deleted, its data is located in an object that may be kept alive by other batches. Therefore it is necessary to regularly move batches from WAL Segments to Tiered Storage segments. In practical terms, it means that the lifetime of a WAL segment is slightly more than the longest configured roll time, and the earliest physical deletion guarantee is after the longest roll.

For example: Topic A rolls a segment and uploads to tiered storage once per hour, and Topic B has a 15 minute retention time. WAL Segments may contain data from both Topic A and B together. 15 minutes after a batch is produced to Topic B, it is logically deleted and no longer visible to consumers. 1 hour after a batch is produced to Topic A, it is rolled and copied to Tiered Storage. Then both batches from topic A and B can be physically deleted and the space reclaimed.

Public Interfaces

Plugin Interfaces: Storage Backend

This interface abstracts the access methods to the object storage.

Code Block

language	java

public record ByteRange(long offset, long size) {}

public interface ObjectStorage extends Configurable {
    /**
     * Uploads an object to object storage.
     * <p>An exception must be thrown in case the number of bytes streamed from
     * {@code inputStream} is different from {@code length}.
     * @param key                      key of the object to upload.
     * @param inputStream              data of the object that will be uploaded.
     * @param length                   length of the data that will be uploaded.
     * @throws StorageBackendException if there are errors during the upload.
     */
    void upload(ObjectKey key, InputStream inputStream, long length)
        throws StorageBackendException;


    /**
     * Fetch a range of an object from object storage.
     * <p>This result should be cached for later access
     * @param key                      key of the object to fetch.
     * @param range                    range of bytes within the object.
     * @throws StorageBackendException if there are errors during the fetch.
     */
    InputStream fetch(ObjectKey key, ByteRange range)
        throws StorageBackendException;

    /**
     * Idempotently delete multiple objects.
     * <p>If an object doesn't exist, the operation should still succeed	
     * to be idempotent.
     * @param keys                     keys of objects to delete.
     * @throws StorageBackendException if there are other errors which prevent deletion.
     */
    void delete(Set<ObjectKey> keys)
        throws StorageBackendException;
}

Apache Kafka will not provide a production-grade implementation of this interface, because this would require Kafka depending on third party storage drivers. The list of potential storage is big and it’s not possible to provide implementations for all of them. Implementations will be provided by 3rd party developers in the Kafka ecosystem.

Configurations

Broker Configurations

diskless.system.enable: enables diskless support on a broker.
diskless.storage.class.name: the object storage class name for diskless topics.
diskless.storage.class.path: the object storage class path for diskless topics.
diskless.append.commit.interval.ms: defines how long to wait before closing a shared log segment object to be uploaded/committed.
diskless.append.buffer.max.bytes: defines the maximum size of a shared log segment object before closing it for further upload/commit.

Topic Configurations

diskless.enable: Sets a topic as a diskless topic. Diskless topics can only be defined at creation time (updating a classic topic to a diskless topic is out of scope.)

Client Configurations

Producer:

client.rack: Similar to Consumer's usage of client rack on the follower fetching feature, it defines the producer rack to align with brokers on the same rack.

Broker API

Diskless Coordinator

A new set of APIs will be required to allow brokers to contact the diskless coordinator.

These new requests are out of scope for this KIP, and will be fully defined in KIP-1164: Diskless Coordinator

Client Metadata

We propose to create the version 14 of the Metadata API with the following definition:

MetadataRequest:

Code Block

language	js

{
 "apiKey": 3,
 "type": "request",
 "listeners": ["broker"],
 "name": "MetadataRequest",
 "validVersions": "0-14",
 "flexibleVersions": "9+",
 "fields": [
   // In version 0, an empty array indicates "request metadata for all topics."  In version 1 and
   // higher, an empty array indicates "request metadata for no topics," and a null array is used to
   // indicate "request metadata for all topics."
   //
   // Version 2 and 3 are the same as version 1.
   //
   // Version 4 adds AllowAutoTopicCreation.
   //
   // Starting in version 8, authorized operations can be requested for cluster and topic resource.
   //
   // Version 9 is the first flexible version.
   //
   // Version 10 adds topicId and allows name field to be null. However, this functionality was not implemented on the server.
   // Versions 10 and 11 should not use the topicId field or set topic name to null.
   //
   // Version 11 deprecates IncludeClusterAuthorizedOperations field. This is now exposed
   // by the DescribeCluster API (KIP-700).
   // Version 12 supports topic Id.
   // Version 13 supports top-level error code in the response.
   // Version 14 supports Diskless and allows to send client.rack
   { "name": "Topics", "type": "[]MetadataRequestTopic

Code Block

language	js

{
 "apiKey": 3,
 "type": "response",
 "name": "MetadataResponse",
 // Version 1 adds fields for the rack of each broker, the controller id, and whether or not the topic is internal.
 //
 // Version 2 adds the cluster ID field.
 //
 // Version 3 adds the throttle time.
 //
 // Version 4 is the same as version 3.
 //
 // Version 5 adds a per-partition offline_replicas field. This field specifies
 // the list of replicas that are offline.
 //
 // Starting in version 6, on quota violation, brokers send out responses before throttling.
 //
 // Version 7 adds the leader epoch to the partition metadata.
 //
 // Starting in version 8, brokers can send authorized operations for topic and cluster.
 //
 // Version 9 is the first flexible version.
 //
 // Version 10 adds topicId.
 //
 // Version 11 deprecates ClusterAuthorizedOperations. This is now exposed
 // by the DescribeCluster API (KIP-700).
 // Version 12 supports topicId.
 // Version 13 supports top-level error code in the response.
 // Version 14 supports Diskless.
 "validVersions": "0-13",
 "flexibleVersions": "9+",
 "fields": [
   { "name": "ThrottleTimeMs", "type": "int32", "versions": "3+", "ignorable": true,
     "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
   { "name": "Brokers", "type": "[]MetadataResponseBroker", "versions": "0+",
     "about": "A list of brokers present in the cluster.", "fields": [
     { "name": "NodeId", "type": "int32", "versions": "0+", "mapKey": true, "entityType": "brokerId",
       "about": "The broker ID." },
     { "name": "Host", "type": "string", "versions": "0+",
 "nullableVersions": "1+",
     "about": "The broker hostnametopics to fetch metadata for." },, "fields": [
     { "name": "PortTopicId", "type": "int32uuid", "versions": "010+",
      "ignorable": true, "about": "The brokertopic portid." },
     { "name": "RackName", "type": "string", "versions": "10+", "nullableVersionsentityType": "1+topicName", "ignorablenullableVersions": true, "default": "null"10+",
       "about": "The rack of the broker, or null if it has not been assigned to a racktopic name." }
   ]},
   { "name": "ClusterIdAllowAutoTopicCreation", "type": "stringbool", "nullableVersionsversions": "24+", "versionsdefault": "2+true", "ignorable": true, "default": "null"false,
     "about": "TheIf clusterthis IDis thattrue, respondingthe broker belongs to may auto-create topics that we requested which do not already exist, if it is configured to do so." },
   { "name": "ControllerIdIncludeClusterAuthorizedOperations", "type": "int32bool", "versions": "1+", "default": "-1", "ignorable": true, "entityType": "brokerId"8-10",
     "about": "TheWhether IDto ofinclude thecluster controllerauthorized brokeroperations." },
   { "name": "TopicsIncludeTopicAuthorizedOperations", "type": "[]MetadataResponseTopicbool", "versions": "08+",
     "about": "EachWhether to include topic inauthorized the responseoperations.", "fields": [
   },
   { "name": "ErrorCodeRackId", "type":  "int16string", "versions": "0+",
  14+", "default": "", "ignorable": true,
     "about": "TheRack topicID error,of orthe 0client ifmaking there was no errorthis request."}
 ]
}

Here, RackId is the new field.

MetadataResponse:

Code Block

language	js

{
 "apiKey": 3,

     { "name": "Name", "type": "stringresponse",
 "versionsname": "0+MetadataResponse", "mapKey": true, "entityType": "topicName", "nullableVersions": "12+",
       "about": "The topic name. Null for non-existing topics queried by ID. This is never null when ErrorCode is zero. One of Name and TopicId is always populated." },
     { "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true,
       "about": "The topic id. Zero for non-existing topics queried by name. This is never zero when ErrorCode is zero. One of Name and TopicId is always populated." },
     { "name": "IsInternal", "type": "bool", "versions": "1+", "default": "false", "ignorable": true,
       "about": "True if the topic is internal." },
     { "name": "Partitions", "type": "[]MetadataResponsePartition", "versions": "0+",
       "about": "Each partition in the topic.",
 // Version 1 adds fields for the rack of each broker, the controller id, and whether or not the topic is internal.
 //
 // Version 2 adds the cluster ID field.
 //
 // Version 3 adds the throttle time.
 //
 // Version 4 is the same as version 3.
 //
 // Version 5 adds a per-partition offline_replicas field. This field specifies
 // the list of replicas that are offline.
 //
 // Starting in version 6, on quota violation, brokers send out responses before throttling.
 //
 // Version 7 adds the leader epoch to the partition metadata.
 //
 // Starting in version 8, brokers can send authorized operations for topic and cluster.
 //
 // Version 9 is the first flexible version.
 //
 // Version 10 adds topicId.
 //
 // Version 11 deprecates ClusterAuthorizedOperations. This is now exposed
 // by the DescribeCluster API (KIP-700).
 // Version 12 supports topicId.
 // Version 13 supports top-level error code in the response.
 // Version 14 supports Diskless.
 "validVersions": "0-13",
 "flexibleVersions": "9+",
 "fields": [
       { "name": "ErrorCodeThrottleTimeMs", "type": "int16int32", "versions": "03+",
    "ignorable": true,
     "about": "The partition errorduration in milliseconds for which the request was throttled due to a quota violation, or 0zero if there was no error the request did not violate any quota." },
       { "name": "PartitionIndexBrokers", "type": "int32[]MetadataResponseBroker", "versions": "0+",
         "about": "The partition index." },
  A list of brokers present in the cluster.", "fields": [
     { "name": "LeaderIdNodeId", "type": "int32", "versions": "0+", "mapKey": true, "entityType": "brokerId",
         "about": "The broker ID of the leader broker." },
     {  { "name": "LeaderEpochHost", "type": "int32string", "versions": "7+", "default": "-1", "ignorable": true,
 0+",
        "about": "The leader epoch of this partitionbroker hostname." },
       { "name": "ReplicaNodesPort", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
         "about": "The set of all nodes that host this partitionbroker port." },
       { "name": "IsrNodesRack", "type": "[]int32string", "versions": "01+", "entityTypenullableVersions": "brokerId1+",
 "ignorable":        "abouttrue, "default": "null",
       "about": "The setrack of nodes that are in sync with the leader for this partitionthe broker, or null if it has not been assigned to a rack." },
    ]},
   { "name": "OfflineReplicasClusterId", "type": "[]int32"string", "nullableVersions": "2+", "versions": "52+", "ignorable": true, "entityTypedefault": "brokerIdnull",
         "about": "The setcluster ofID offlinethat replicasresponding ofbroker thisbelongs partitionto." },
       { "name": "PreferredProduceBrokersControllerId", "type": "[]int32", "versions": "141+", "default": "-1", "ignorable": true, "entityType": "brokerId",
         "about": "The orderedID list of brokers to which the client is recommended to send Produce requests." }
     ]controller broker." },
     { "name": "TopicAuthorizedOperationsTopics", "type": "int32[]MetadataResponseTopic", "versions": "8+", "default": "-21474836480+",
       "about": "32-bitEach bitfieldtopic toin represent authorized operations for this topic." },the response.", "fields": [
     { "name": "IsDisklessErrorCode", "type": "boolint16", "versions": "140+", "default": "false", "ignorable": true,
       "about": "TrueThe topic error, or 0 if thethere topicwas isno disklesserror." },
   ]},
   { "name": "ClusterAuthorizedOperationsName", "type": "int32string", "versions": "0+", "mapKey": true, "entityType": "8-10topicName", "defaultnullableVersions": "-214748364812+",
       "about": "32-bit bitfield to represent authorized operations for this cluster." },
The topic name. Null for non-existing topics queried by ID. This is never null when ErrorCode is zero. One of Name and TopicId is always populated." },
     { "name": "ErrorCodeTopicId", "type": "int16uuid", "versions": "1310+", "ignorable": true,
       "about": "The top-level error code, or 0 if there was no error." }
 ]
}

Here, the topic definition has the new field IsDiskless and the partition definition has the new field PreferredProduceBrokers.

Monitoring

The following metrics may be useful for operators:

Object upload:

count and rates;
object size average and percentiles;
upload traffic;
latency;
errors.

Object commit:

count and rates;
latency;
errors.

Read:

count and rates;
GET requests per Fetch request.

Command line tools

Existing tools to be adapted:

kafka-topics.sh must support the new topic configuration on creation and as a filter to list diskless topics only.
kafka-dump-log.sh must support Shared Log Segment files as input and parse its content correctly.

kafka-diskless-metadata.sh is a new tool, which does the following:

Diskless topic overview:

the offsets;
the size on the object storage;
the total size of object where the topic is part of;
the total Size of objects where topic was part of but batches are deleted;

Getting object metadata, including object key, total size, used size.

Compatibility, Deprecation, and Migration Plan

Existing users upgrading to a version of Kafka with support for diskless writes will not experience any change in behavior. All broker and topic configurations will have defaults which are consistent with the existing storage model. This will be a backwards-compatible upgrade. These users may downgrade without additional steps.

Users which configure diskless brokers but no diskless topics may experience failures related to those configurations if they are invalid (e.g. a plugin is not installed, or backing service is unavailable). If the configuration passes validation and the brokers are able to start, users will experience no change in behavior. These users may downgrade without additional steps.

Users which configure diskless brokers and diskless topics will be able to produce and consume data with the same semantics and consistency model as traditional topics (save for the explicitly outlined exceptions/limitations), but with a higher latency than expected. These users will need to migrate away from and delete all diskless topics before downgrading.

Test Plan

We need to create new integration tests that would cover the following:

Create a diskless topic, write and read from it → Check that data is written to object storage stub (MinIO / Localstack).
Set different commit intervals and check they are written within an error threshold.
Set different buffer max bytes values and check they are held within an error threshold.

We need to create new systems tests that are end to end but stubbed with MinIO where:

Have a cluster running with both types of topics.
Have a cluster running with diskless topics, stop the cluster and restart (no data loss).
Have a cluster with only diskless topics and monitor that disk is only used for metadata.

Various failure scenarios (e.g. the Diskless Coordinator or Object Storage are inaccessible or partially accessible) should be tested extensively, with the focus primarily on correctness.

The majority of the existing produce and consume tests should be adapted to diskless topics by parameterizing the topic configuration. However, some of the current tests fall in the “known limitations and exceptions”.

Documentation Plan

These are the areas where we need to add documentation:

Quick start needs to be enhanced to explain how to create diskless topics.
New configuration settings need to be documented (in Java class).
Expand subsection 3.2 – Topic Configs to explain the new configuration option for topic creation.
Add a new subsection under section 4 – Design. Something like 4.11 – Diskless topics architecture.
Add a new subsection under section 6 – Operations. Something like 6.13 – Diskless topics operations. This would include:

A brief explanation of what the feature entails.
A detailed explanation of the new config options.
Some deployment configurations and options (hybrid vs segregated brokers).

Rejected Alternatives

Use Tiered Storage as-is

Tiered storage only affects the behavior of inactive segments, while active segments continue to use block storage and replication. It is possible to avoid inter-zone replication by setting replication.factor=1, but causes the topic to inherit the durability of block storage, which may experience correlated failures within a single rack.

Tiered Storage as-is forces a trade-off between durability and replication costs.

Using “Aggressive Tiering”, an operator may configure their cluster to roll active segments quickly, reducing the total active segment size. This has the effect of also reducing the window that data is stored non-durably, such that it might be possible to delay acks until a segment is uploaded to tiered storage. However, as tiered storage uploads individual segments, these requests can incur a large I/O cost. In many cases, this will negate the positive effects from shrinking or eliminating the block storage.

Tiered Storage as-is forces a trade-off between durability and excessive I/O overhead.

Tiered topics still have leaders, and under the Kafka protocol, producers must produce to the leader, even if the leader is in a different rack, incurring cross-rack transfer costs. To mitigate this, all producers and brokers can be placed in a single rack.

Tiered storage as-is forces a trade-off between availability and producer ingress costs.

As a summary, it can be seen that the inter-AZ traffic can be avoided but with the cost of losing availability of producers and brokers and availability and durability of topic data. The local storage requirement can be minimized with certain penalties explained earlier but not fully eliminated. The authors are not aware of any mechanism to improve tiered storage to eliminate these shortcomings.

Tiered Storage metadata being abstracted to segment granularity and segments being limited to a single partition prevents the overall cost optimization of the system, and forces operators to sacrifice very important guarantees in order to unlock cost savings. Diskless topics maintain these guarantees, while still unlocking cost savings.

File system driver

One may argue that we could keep the Kafka code unchanged or minimally changed, but instead use a sophisticated file system driver. Brokers would use the local broker disk backed by the driver and write segment files and other file types and read from them normally. The driver would be responsible for transferring the IO to the object storage. The native Kafka replication could be prevented by setting replication.factor=1. This approach comes with a number of challenges that make us reject it.

Mapping block storage operations to object storage directly is inefficient, and requires some assumptions about access patterns. For example, object storages don’t widely support random writes or appends, so objects would need to be overwritten often. This would result in poor performance and high costs in hyperscalers which have a per-request cost.

There are log-structured file systems implemented on top of object storage, but these are either proprietary or immature. And because they operate on a lower-level abstraction than Diskless Shared Log Segments, they are not able to take advantage of application-specific access behaviors and optimizations.

A filesystem driver also does not cover the rack awareness functionality, and modifying the code would still be required to implement a similar feature. Kafka may also need to be aware of some external component as low level as the file system driver to collaborate with it and have brokers which are able to access data they have not replicated. Kafka is not prepared to manage multiple writers to a single local filesystem.

 topic id. Zero for non-existing topics queried by name. This is never zero when ErrorCode is zero. One of Name and TopicId is always populated." },
     { "name": "IsInternal", "type": "bool", "versions": "1+", "default": "false", "ignorable": true,
       "about": "True if the topic is internal." },
     { "name": "Partitions", "type": "[]MetadataResponsePartition", "versions": "0+",
       "about": "Each partition in the topic.", "fields": [
       { "name": "ErrorCode", "type": "int16", "versions": "0+",
         "about": "The partition error, or 0 if there was no error." },
       { "name": "PartitionIndex", "type": "int32", "versions": "0+",
         "about": "The partition index." },
       { "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
         "about": "The ID of the leader broker." },
       { "name": "LeaderEpoch", "type": "int32", "versions": "7+", "default": "-1", "ignorable": true,
         "about": "The leader epoch of this partition." },
       { "name": "ReplicaNodes", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
         "about": "The set of all nodes that host this partition." },
       { "name": "IsrNodes", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
         "about": "The set of nodes that are in sync with the leader for this partition." },
       { "name": "OfflineReplicas", "type": "[]int32", "versions": "5+", "ignorable": true, "entityType": "brokerId",
         "about": "The set of offline replicas of this partition." },
       { "name": "PreferredProduceBrokers", "type": "[]int32", "versions": "14+", "ignorable": true, "entityType": "brokerId",
         "about": "The ordered list of brokers to which the client is recommended to send Produce requests." }
     ]},
     { "name": "TopicAuthorizedOperations", "type": "int32", "versions": "8+", "default": "-2147483648",
       "about": "32-bit bitfield to represent authorized operations for this topic." },
     { "name": "IsDiskless", "type": "bool", "versions": "14+", "default": "false", "ignorable": true,
       "about": "True if the topic is diskless." }
   ]},
   { "name": "ClusterAuthorizedOperations", "type": "int32", "versions": "8-10", "default": "-2147483648",
     "about": "32-bit bitfield to represent authorized operations for this cluster." },
   { "name": "ErrorCode", "type": "int16", "versions": "13+", "ignorable": true,
     "about": "The top-level error code, or 0 if there was no error." }
 ]
}

Here, the topic definition has the new field IsDiskless and the partition definition has the new field PreferredProduceBrokers.

Monitoring

The following metrics may be useful for operators:

Object upload:

count and rates;
object size average and percentiles;
upload traffic;
latency;
errors.

Object commit:

count and rates;
latency;
errors.

Read:

count and rates;
GET requests per Fetch request.

Command line tools

Existing tools to be adapted:

kafka-topics.sh must support the new topic configuration on creation and as a filter to list diskless topics only.
kafka-dump-log.sh must support Shared Log Segment files as input and parse its content correctly.

kafka-diskless-metadata.sh is a new tool, which does the following:

Diskless topic overview:

the offsets;
the size on the object storage;
the total size of object where the topic is part of;
the total Size of objects where topic was part of but batches are deleted;

Getting object metadata, including object key, total size, used size.

Compatibility, Deprecation, and Migration Plan

Existing users upgrading to a version of Kafka with support for diskless writes will not experience any change in behavior. All broker and topic configurations will have defaults which are consistent with the existing storage model. This will be a backwards-compatible upgrade. These users may downgrade without additional steps.

Users which configure diskless brokers but no diskless topics may experience failures related to those configurations if they are invalid (e.g. a plugin is not installed, or backing service is unavailable). If the configuration passes validation and the brokers are able to start, users will experience no change in behavior. These users may downgrade without additional steps.

Users which configure diskless brokers and diskless topics will be able to produce and consume data with the same semantics and consistency model as traditional topics (save for the explicitly outlined exceptions/limitations), but with a higher latency than expected. These users will need to migrate away from and delete all diskless topics before downgrading.

Test Plan

We need to create new integration tests that would cover the following:

Create a diskless topic, write and read from it → Check that data is written to object storage stub (MinIO / Localstack).
Set different commit intervals and check they are written within an error threshold.
Set different buffer max bytes values and check they are held within an error threshold.

We need to create new systems tests that are end to end but stubbed with MinIO where:

Have a cluster running with both types of topics.
Have a cluster running with diskless topics, stop the cluster and restart (no data loss).
Have a cluster with only diskless topics and monitor that disk is only used for metadata.

Various failure scenarios (e.g. the Diskless Coordinator or Object Storage are inaccessible or partially accessible) should be tested extensively, with the focus primarily on correctness.

The majority of the existing produce and consume tests should be adapted to diskless topics by parameterizing the topic configuration. However, some of the current tests fall in the “known limitations and exceptions”.

Documentation Plan

These are the areas where we need to add documentation:

Quick start needs to be enhanced to explain how to create diskless topics.
New configuration settings need to be documented (in Java class).
Expand subsection 3.2 – Topic Configs to explain the new configuration option for topic creation.
Add a new subsection under section 4 – Design. Something like 4.11 – Diskless topics architecture.
Add a new subsection under section 6 – Operations. Something like 6.13 – Diskless topics operations. This would include:

A brief explanation of what the feature entails.
A detailed explanation of the new config options.
Some deployment configurations and options (hybrid vs segregated brokers).