...

1. 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.
2. 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.
3. 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.

...

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

...

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 of MetadataRequest and  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:

1. Send Produce requests to recommended produce brokers (in order and subject to availability).
2. Send Fetch requests as for classic topics: either to the leader or to the closest replica based on the rack.
3. 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.

...

• 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 doesn’t change , 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.

...

1. The Diskless Coordinator doesn’t need to be concerned about the object storage, to have credentials and the code to access it.
2. 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.

1. 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.
2. 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 are is 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-1165it 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

...

{
 "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:

...

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:

...