Authors: Greg Harris, Ivan Yurchenko, Jorge Quilcate, Giuseppe Lillo, Anatolii Popov, Juha Mynttinen, Josep Prat, Filip Yonov

Status

Current state: Under Discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]19161

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

Motivation

KIP-1150: Diskless Topics introduces the concepts of diskless topics. KIP-1163 describes the core functionality of diskless topics, such as the produce and consume paths. However, KIP-1163 left the Batch Coordinator interface and reference implementation unspecified. This KIP fills this gap.

Glossary

• Diskless Topic: A topic which does not append to or serve from block storage devices.
• 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.
• Batch: A container for Kafka records and a unit of record representation in the network protocol and storage format in both status quo Kafka and diskless topics.
• Shared Log Segment Object: An object containing a shared log segment 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 object, at some byte range.
• 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.
• Diskless Manager: The set of components internal to the broker that handle diskless topics operations, such as creating objects, uploading objects to object storage, etc.

Proposed Changes

This KIP has two purposes:

1. Define the Batch Coordinator plugin interface.
2. Describe the reference topic-based implementation of this interface that will be shipped with Kafka.

Batch Coordinator

The Batch Coordinator is the source of truth about objects, partitions, and batches in diskless topics. It does the following:

1. Chooses the total ordering for writes, assigning offsets without gaps or duplicates.
2. Serves requests for log offsets.
3. Serves requests for batch coordinates.
4. Serves requests for atomic operations (creating partitions, deleting topics, records, etc.)
5. Manages data expiry and soft deletion.
6. Manages object physical deletion.

The three main entities the Batch Coordinator is concerned about are objects, partitions, and batches. We’re interested mainly in the following information about objects:

1. The object key: the object name/path of the object in the object storage.
2. The object state: “uploaded” or “deleted” (for soft deletion).
3. The size and the actually used size for tracking when the object becomes empty to be deleted.

About partitions, we’re interested mainly in the following:

1. The topic ID and name, partition number.
2. The log start offset and the high watermark: to track the beginning and end of the partition.

About batches, we’re interested mainly in the following:

1. The reference to the log (the topic ID + partition).
2. The reference to the object where the batch is located.
3. The base and last offsets.
4. The byte offset in the object.
5. The batch size in bytes.
6. The timestamps: log append time and batch max timestamp.
7. The information necessary to support idempotent producers.

We propose to make the Batch Coordinator pluggable. The plugin interface is described in the Public Interfaces section.

Batch Coordinator implemented with an internal topic

This is a reference implementation that will be shipped with Kafka. It uses an internal, non-compacted topic called __diskless-metadata for storing diskless metadata. This topic is created by default with one partition, but the number of partitions can be increased later (see the Scaling the number of Batch Coordinators section.)

The reference implementation of the Batch Coordinator is based upon the coordinator runtime framework.

Given the nature of diskless topics, clients can fetch messages from these topics from every broker in the cluster. This characteristic requires that the batch coordinates should be retrieved in an efficient way, without the need to contact the leader of the metadata partition for every fetch request. 

In order to solve this problem, the Batch Coordinator makes use of read-only coordinators. Read-only Batch Coordinators are run on the brokers that are in-sync replicas for the __diskless-metadata topic. They can serve read requests, but cannot serve write requests.

For simplicity, let’s assume there’s only one partition, __diskless-metadata-0. The Batch Coordinator instance will run on the broker that is the leader of this partition, while read-only Batch Coordinators instances will run on all the brokers that are in-sync replicas for this partition. The Batch Coordinator will handle all the requests that modify the state (such as batch commit and delete requests), while the read-only Batch Coordinators can serve requests that only need to read the state. 

Image Removed

The topic is the storage and replication medium, but it cannot be directly queried in a performant way. For this, the state needs to be materialized locally. This, in turn, poses another challenge: the state can become too large to fit into memory. That means, the materialization mechanism must be backed by a disk. This set of requirements suggests using some embedded disk-based database engine. We propose to use SQLite. SQLite is an open-source, self-contained SQL database engine that can be embedded into an application. It’s widely used in the software industry and can be easily integrated into any codebase. Using SQLite offers the possibility for structured querying and indexing of the state, allowing fast and easy access to the Batch Coordinator state without the need for implementing all the data structures and algorithms needed for managing the state on disk.

Protocol

...

: Diskless Core describes how Diskless topics work. One of the principal components that was mentioned but not described was the Diskless coordinator. This KIP closes this gap and describes the Diskless coordinator (DC) in detail.

Role of Diskless coordinator

The Diskless coordinator manages metadata specific to Diskless topics, such as batches, WAL files, producer and transaction state for Diskless topics. Brokers rely on DC to perform operations on Diskless topics. For example:

1. When a broker writes a WAL file, it sends a commit request to DC so it stores the metadata about the file and batches.
2. When a broker serves a consumer, it queries DC for partitions’ offsets and existing batches and also files where the batches are located.
3. When a partition is deleted or truncated, DC deletes the corresponding batches and marks WAL files as free.

To perform this and other operations, brokers need to use the request-response API of the Diskless coordinator.

Proposed Changes

This KIP proposes the following:

1. The Diskless coordinator broker API.
2. The approach to the Diskless coordinator implementation based on topics and the coordinator framework.

Diskless coordinator API

The Diskless coordinator API will be a part of the broker API. However, it is not expected to be used directly by client applications. It will require the CLUSTER_ACTION permission.

The Diskless coordinator will expose the following operations:

1. DisklessCreatePartitions: Create new partitions, could be used both for creating new topics and for increasing the number of partitions in an existing one.
2. DisklessDeleteTopics: Delete existing topics and their partitions.
3. DisklessCommitFile: Commit a WAL file with batches. During this operation, various checks are performed (e.g. for producer idempotence, transactional checks) and offsets are assigned to batches.
4. DisklessDeleteRecords: Delete tail records in partitions.
5. DisklessListOffsets: List offsets for specified timestamps, including special timestamps for “earliest”, “latest” for finding the log start offset and high watermark. This is a read-only operation with strict consistency (i.e. it should reflect the state after applying all the previous mutating operations).
6. DisklessFindBatches: Find batches in the specified partitions starting from the specified offset. This is a read-only operation with less strict consistency requirements (i.e. could be served from stale state and by followers).
7. DisklessDescribeFile: Check if the coordinator has live batches in a WAL file.

The request-response protocol for these operations will be explained in the Public interfaces section.

Some operations remain out of scope of this KIP, such as the ones for transaction management or offloading to tiered storage. They will be explained in the corresponding KIPs.

Topic-based implementation

The KIP proposes to use a new internal Kafka topic __diskless_metadata with multiple partitions as the primary storage. The implementation and deployment and operation models will be built around this fact.

Partitioning of __diskless_metadata

The __diskless_metadata topic will have multiple partitions, effectively creating multiple coordinators and sharding the space of user partitions for scalability. Following the current terminology, we will be saying “Diskless coordinators” and “the Diskless coordinator for partition N” when referring to Diskless code and (meta-)data associated with these partitions.

The __diskless_metadata partitions and Diskless coordinators will be independent in all senses: placement, Diskless operations, etc.

User partitions will be assigned to Diskless coordinators / metadata partitions during creation. This mapping will be stored as a part of partition metadata. Adding partitions to __diskless_metadata will be possible to increase the total capacity of the cluster, but only newly created user partitions will be able to take advantage of the new metadata partitions / Diskless coordinators.

Moving of user partition from Diskless coordinator to another is possible, but remains out of scope of this KIP.

The default number of partitions will be TBD.

Leadership and deployment

The implementation will follow the model set by other coordinators in Kafka.

__diskless_metadata partition leaders will be coordinators. They will serve as an entry point for all mutating operations for the corresponding user partitions. The brokers which need to communicate with a coordinator will be discovering it using FindCoordinator API.

It will be possible to control which brokers host the coordinators through partition placement of the __diskless_metadata topic. If necessary, this will allow dedicating a subset of brokers to serving as Diskless coordinators.

Local state machine

The source of truth for the coordinator will be the __diskless_metadata topic. Similar to other coordinators in Kafka, the Diskless coordinator will need a local state machine materializing the metadata log to facilitate performing of the operations.

In contrast to other coordinators, the expected size of the state of DC is big (up to hundreds of megabytes or even gigabytes), so it’s impractical to keep it in memory. We propose using SQLite as the way to locally materialize the metadata log. SQLite is an open-source widely-used ACID embedded DBMS. Using SQLite offers the possibility for structured querying and indexing of the state, allowing fast and easy access to the Diskless metadata state without the need for implementing all the data structures and algorithms needed for managing the state on disk.

The SQLite DB will be a local metadata cache, not the source of truth. It could be dropped when needed to be refilled again from the log and snapshots.

How operations are performed

The high-level idea of how operations are performed is simple and resembles what happens in the KRaft controller:

1. Do check against the current state (e.g. whether the operation could be performed at all, fully or partially, etc.)
2. For mutating operations, append one or several metadata records to the log.
3. Wait for the metadata records to be replicated to the followers.
4. Reply to the client.

However, the question is in which order to wait for metadata record replication and to apply the changes to the local state. There are two somewhat conflicting requirements:

1. We want to perform operations in the pipelined manner to reduce the operation latency. That is, to be able to start working on the following operation while the current one is still waiting to be replicated. This requires the local state to be consistent with the previous pending (non-replicated) operations.
2. We don’t want to modify the local state until we’re sure it’s consistent with the log to reduce the recovery time in case of failures (reduce the probability of rebuilding the state from scratch). This requires the local state to be updated only with the replicated operations.

While these two requirements contradict each other, they can be reconciled by doing the pre-operation checks against the local state and pending operations at the same time.

From the coordinator point of view, the API request processing logic will be the following:

1. The broker API receives a request (for example, DisklessCommitFile).
2. The broker checks that it hosts the coordinator (i.e. it is the leader of the corresponding partition of __diskless_metadata) and passes the request to it. Otherwise, NOT_COORDINATOR error will be returned.
3. The coordinator speculatively selectively applies the current pending operations to the current committed local state and does the necessary checks against the resulting speculative state. The speculative application can be done either fully in memory or within a to-be-rejected SQLite transaction. At this point, the request may turn out to be fully or partially erroneous, which will be reflected in the response.
4. The coordinator generates one or more metadata records and publishes them to the local log. These records become pending for the future operations until they are committed to the local state. If there are multiple records, they will be appended to the local log atomically within the same batch. The record formats are explained in the Public interfaces section.
5. The coordinator waits until these records are replicated to in-sync replicas (the acks=all way), i.e. until the high watermark advances past them.
6. The coordinator applies the replicated records to the local state for real.
7. The coordinator sends the response to the client.

Read-only operations that require a consistent up-to-date view of data (e.g. DisklessListOffsets) will skip step 4, but will still wait for the preceding records to be replicated. Some other read-only operations could be served from a stale view of data (e.g. DisklessFindBatches) and could be served right away after step 2, also by a follower.

When an operation is applied to the local state, atomically with this the corresponding offset in the metadata log is stored to facilitate potential recovery from the known point.

Metadata log size management

By far the biggest contributor to the total metadata log size will be metadata records about committed batches and WAL files. The Diskless system is designed to work in cooperation with the tiered storage system by periodically combining batches from Diskless topics into Kafka segments and offloading them to tiered storage. This means that even with infinite data retention, the lifetime of each batch and WAL file metadata inside DC is finite and determined by user configuration (mainly segment.ms and segment.bytes).

The primary mechanism to keep the metadata log size contained will be snapshotting and pruning of the log. Periodically, the leader will take snapshots of the local state asynchronously and the followers will be able to fetch these snapshots. Once a metadata log offset is in a snapshot, it could be pruned. This mechanism is identical to the one in KRaft (see KIP-630).

In KRaft, snapshots are created from an in-memory metadata image. In the Diskless coordinator, the local state will be stored in an SQLite database. To make an asynchronous snapshot point-in-time consistent snapshot, SQLite read-only transactions or the backup API will be used.

WAL file management

Technically it’s not impossible (however, not desirable and we’ll discuss below how to avoid this) that a WAL file contains data for Diskless partitions that belong to different Diskless coordinators, i.e. there’s no one owner that controls the file lifetime.

To overcome this difficulty, we propose the following approach:

1. When a broker is about to commit a file, it gathers the IDs of all the DCs it’s going to send the commit requests to into the list of owners. The list of owners is randomly permuted.
2. The list of owners is included as the field in DisklessCommitFile request. Now each DC knows what other DCs claim the file.
3. The first DC in the owner list becomes the owner.
4. Once the last batch in this DC for the file is deleted, the DC hands over the file to the next owner in the list.
5. Once this happens to the last owner, the file is deleted.

The handover and deletion operations must not interfere with other DC activity, i.e. they must be performed asynchronously. The status of the file is changed in the local state (but not in the metadata log, because this information is already implicitly present there and the followers will know it) and a background worker is started for the corresponding operation (handover or deletion). If the target broker for handover or the remote storage is not available, the background worker will retry the operation indefinitely. If the DC leadership changes, the background workers on the current broker will be stopped and started again on the new leader (as the new leader has been reading the same metadata log and knows the status of files).

When a file is to be deleted, it makes sense to allow for some grace period and not delete it right away to allow consumers finish possible ongoing reads from them.

Orphan files

It’s possible that a broker uploads a WAL file, but fails to commit it with any DC and also fails to do on-the-spot cleanup. The most obvious example is a broker crash right after uploading. These files will be a dead weight, making users pay for the extra storage. To deal with these orphan files, a special background worker will periodically scan the remote storage in order to find such files and set them for deletion. The algorithm will be the following:

1. Before each scan, the worker will ask each DC for the timestamp of the oldest file it has batches in. The grace period will be added to the oldest of all the timestamps, forming the timestamp threshold.
2. The worker will scan the remote storage for the files older than the threshold.
3. As an additional safety measure, the working will ask each DC whether particular files are known to it.
4. If a file is not known to any DC, it will be physically deleted by the worker.

The scan frequency must be configurable. An orphan file is expected to be a rare event, so the default scan frequency should be correspondingly low.

Reducing multi-DC commit operations

Any broker can handle a Produce request for any Diskless partition. As mentioned above, this opens the possibility for a WAL file to contain data from partitions belonging to different Diskless coordinators. In this case, the broker will have to commit this file against multiple DCs. This situation is undesirable because in the worst case committing one WAL file will result in n_dcs (number of DCs) outbound network commit calls, which increases the chances for partial failure and higher tail latencies. This section is dedicated to alleviating this problem.

The problem is partially alleviated by the fact that one broker may host several DCs, so logical commit requests can be bundled together into fewer physical ones. However, the worst case is still bad enough: the upper bound of outbound network commit calls will be n_brokers (number of brokers).

If we direct producers the right way, we will be able to concentrate Diskless Produce requests so that brokers need to do fewer network commit calls. KIP-1163 proposes to expand the Metadata request/response for newer clients, particularly to add the PreferredProduceBrokers field. The value of this field will be dynamically calculated on the broker side.

For each Diskless coordinator, one or several brokers in each rack will be selected as “produce gateways”. This means, when a producer needs to send a Produce request to a Diskless partition managed by this DC, it’ll be encouraged to send it to particular brokers depending on the raсk, not just to any broker. 

Provided there’s enough brokers in the cluster, we can make sure that each broker is the produce gateway for only one DC (in its rack). Let’s consider some scenarios to clarify the idea.

Scenario 1: the number of brokers is less than the number of DCs.
Racks: 3. Brokers: 3. DCs: 12.
For each DC, we select one broker in each rack as the produce gateway. That means, each broker will be a gateway for 12 DCs (12 DCs * 1 * 3 racks / 3 brokers). However, there are fewer brokers than DCs and each broker will have to do only up to n_brokers outbound network commit calls per WAL file.

Scenario 2: same, just more brokers.
Racks: 3. Brokers: 6. DCs: 12.
For each DC, we select one broker in each rack as the produce gateway. That means, each broker will be a gateway for 6 DCs (12 DCs * 1 * 3 racks / 6 brokers). However, there’s fewer brokers than DCs and each broker will have to do only up to n_brokers outbound network commit calls per WAL file.

Scenario 3: the number of brokers is equal to the number of DCs.
Racks: 3. Brokers: 12. DCs: 12.
For each DC, we select one broker in each rack as the produce gateway. That means, each broker will be a gateway for 3 DCs (12 DCs * 1 * 3 racks / 12 brokers). Each broker will have to do up to n_brokers outbound network commit calls per WAL file.

Scenario 4: the number of brokers is greater than the number of DCs.
Racks: 3. Brokers: 24. DCs: 12.
For each DC, we select one broker in each rack as the produce gateway. That means, each broker will be a gateway for 2 DC (12 DCs * 1 * 3 racks / 24 brokers = 1.5). Each broker will have to do up to 2 outbound network commit calls per WAL file.

Scenario 5: same, just more brokers.
Racks: 3. Brokers: 36. DCs: 12.
For each DC, we select one broker in each rack as the produce gateway. That means, each broker will be a gateway for 1 DC (12 DCs * 1 * 3 racks / 36 brokers). Each broker will have to do up to 1 outbound network commit calls per WAL file.

Scenario 6: same, just more brokers.
Racks: 3. Brokers: 108. DCs: 12.
For each DC, we select three brokers in each rack as produce gateways. That means, each broker will be a gateway for 1 DC (12 DCs * 3 * 3 racks / 18 brokers). Each broker will have to do up to 1 outbound network commit calls per WAL file.

The number of produce gateways per DC per rack can be calculated as max(1, n_brokers / n_dcs / n_racks).

By this, provided we’re having enough brokers, each broker will have to do less than n_brokers outbound network commit calls per WAL file, going down to 1. The saturation point is where n_brokers = n_dcs * n_racks.

A further important optimization is possible. In scenarios before the saturation point (too few brokers), we can mirror co-location of DCs by co-location of their produce gateways. That means if DCs dc1, dc2, dc3 are co-located on some broker, their produce gateways should also be co-located on some brokers in other racks. With this, we lower the upper bound of outbound network commit calls per WAL file from n_brokers to n_racks.

And last optimization but not least is that sending requests to DCs on the same broker doesn’t need to be a network call, so the upper bound is in any case n_brokers-1 or n_racks-1. This matters because DC can be its own produce gateway in its own rack.

Public Interfaces

The proposed public interface changes are subject to change during the community discussion.

Diskless coordinator API

DisklessCreatePartitions

{
  "apiKey": 93,
  "type": "request",
  "listeners": ["controller"],
  "name": "DisklessCreatePartitionsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]DisklessCreatablePartitions", "versions": "0+",
      "about": "Each topic where to create partitions.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+", "mapKey": true,
        "about": "The unique topic ID." },
      { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "NumPartitions", "type": "int32", "versions": "0+",
        "about": "The number of partitions to create for the topic." }
    ]},
    { "name": "TimeoutMs", "type": "int32", "versions": "0+", "default": "60000",
      "about": "How long to wait in milliseconds before timing out the request." }
  ]
}

{
  "apiKey": 93,
  "type": "response",
  "name": "DisklessCreatePartitionsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "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": "Topics", "type": "[]DisklessCreatablePartitionsResult", "versions": "0+",
      "about": "Results for each topic.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+", "mapKey": true,
        "about": "The unique topic ID." },
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code, or 0 if there was no error." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "ignorable": true,
        "about": "The error message, or null if there was no error." }
    ]}
  ]
}

DisklessDeleteTopics

{
  "apiKey": 94,
  "type": "request",
  "listeners": ["controller"],
  "name": "DisklessDeleteTopicsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]DisklessDeletableTopicState", "versions": "0+",
      "about": "Each topic to delete.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+", "mapKey": true,
        "about": "The unique topic ID." }
    ]},
    { "name": "TimeoutMs", "type": "int32", "versions": "0+", "default": "60000",
      "about": "The length of time in milliseconds to wait for the deletions to complete." }
  ]
}

{
  "apiKey": 94,
  "type": "response",
  "name": "DisklessDeleteTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "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": "Responses", "type": "[]DisklessDeletableTopicResult", "versions": "0+",
      "about": "The results for each topic we tried to delete.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+", "mapKey": true,
        "about": "The unique topic ID." },
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The deletion error, or 0 if the deletion succeeded." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "ignorable": true, "default": "null",
        "about": "The error message, or null if there was no error." }
    ]}
  ]
}

DisklessCommitFile

{
  "apiKey": 95,
  "type": "request",
  "listeners": ["controller"],
  "name": "DisklessCommitFileRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ObjectKey", "type": "string", "versions": "0+",
      "about": "The object key identifying the file in object storage." },
    { "name": "ObjectFormat", "type": "int8", "versions": "0+",
      "about": "The format of the object (e.g. 1 = WRITE_AHEAD_MULTI_SEGMENT)." },
    { "name": "FileSize", "type": "int64", "versions": "0+",
      "about": "The total size of the file in bytes." },
    { "name": "OwnerIds", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
      "about": "The list of Diskless Coordinator IDs that share ownership of this file." },
    { "name": "Batches", "type": "[]DisklessCommitBatch", "versions": "0+",
      "about": "The batches contained in the file to commit.", "fields": [
      { "name": "Magic", "type": "int8", "versions": "0+",
        "about": "The record batch magic byte." },
      { "name": "RequestId", "type": "int32", "versions": "0+",
        "about": "The ID of the original produce request this batch belongs to." },
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The unique  ID of the topic this batch belongs to." },
      { "name": "PartitionIndex", "type": "int32", "versions": "0+",
        "about": "The partition index this batch belongs to." },
      { "name": "ByteOffset", "type": "int32", "versions": "0+",
        "about": "The byte offset of this batch within the file." },
      { "name": "Size", "type": "int32", "versions": "0+",
        "about": "The size of the batch in bytes." },
      { "name": "BaseOffset", "type": "int64", "versions": "0+",
        "about": "The base offset of the record batch." },
      { "name": "LastOffset", "type": "int64", "versions": "0+",
        "about": "The last offset of the record batch." },
      { "name": "BatchMaxTimestamp", "type": "int64", "versions": "0+",
        "about": "The maximum timestamp in the batch." },
      { "name": "TimestampType", "type": "int8", "versions": "0+",
        "about": "The timestamp type (0 = CreateTime, 1 = LogAppendTime)." },
      { "name": "ProducerId", "type": "int64", "versions": "0+",
        "about": "The producer ID, or -1 if not set." },
      { "name": "ProducerEpoch", "type": "int16", "versions": "0+",
        "about": "The producer epoch, or -1 if not set." },
      { "name": "BaseSequence", "type": "int32", "versions": "0+",
        "about": "The base sequence number, or -1 if not set." },
      { "name": "LastSequence", "type": "int32", "versions": "0+",
        "about": "The last sequence number, or -1 if not set." }
    ]},
    { "name": "TimeoutMs", "type": "int32", "versions": "0+", "default": "60000",
      "about": "How long to wait in milliseconds before timing out the request." }
  ]
}

{
  "apiKey": 95,
  "type": "response",
  "name": "DisklessCommitFileResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "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": "Batches", "type": "[]DisklessCommitBatchResult", "versions": "0+",
      "about": "Results for each batch, in the same order as the request batches.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code, or 0 if there was no error." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "ignorable": true, "default": "null",
        "about": "The error message, or null if there was no error." },
      { "name": "AssignedBaseOffset", "type": "int64", "versions": "0+",
        "about": "The base offset assigned by the coordinator, or -1 on error." },
  

The Diskless Manager uses inter-broker RPCs to communicate with a Batch Coordinator, using a new set of APIs: 

• InitDisklessTopics
• CommitBatches
• DeleteDisklessTopics
• DeleteDisklessRecords
• FindBatches
• ListDisklessOffsets

Scaling the number of Batch Coordinators

The number of Batch Coordinators can be dynamically increased by creating new partitions for the __diskless-metadata topic.

Topic-partitions are assigned to one specific Batch Coordinator. This mapping is stored in the KRaft metadata.

Image Removed

Risks

The __diskless-metadata topic health is essential for the topic-based Batch Coordinator to function correctly. This creates some risks when operating this type of Batch Coordinator:

1. If the topic becomes non-writable (e.g. too few in-sync replicas), the Coordinator cannot perform operations that modify the metadata. This involves, first and foremost, committing shared log segment objects. That means, in case of an outage, the write path will stall. In case of multi-partition __diskless-metadata topic, the outage will be contained only to the affected partitions.
2. If the topic becomes fully unavailable, Batch Coordinator instances would be able to serve read-only operations from their local materialized state, which isn’t guaranteed to be up-to-date.
3. If the topic is deleted (e.g. by an accident), the metadata is lost and the previously written data becomes inaccessible.

Public Interfaces

Batch Coordinator pluggable interface

public record CreateTopicAndPartitionsRequest(
    Uuid topicId,
    String topicName,
    int numPartitions) {
}

public record CommitBatchRequest(
    int requestId,
    TopicIdPartition topicIdPartition,
    int byteOffset,
    int size,
    long baseOffset,
    long lastOffset,
    long batchMaxTimestamp,
    TimestampType messageTimestampType,
    long producerId,
    short producerEpoch,
    int baseSequence,
    int lastSequence) {
}

public record CommitBatchResponse(
    Errors errors,
    long assignedBaseOffset,
    long logAppendTime,
    long logStartOffset,
    boolean isDuplicate,
    CommitBatchRequest request) {
}

public record FindBatchRequest(
    TopicIdPartition topicIdPartition,
    long offset,
    int maxPartitionFetchBytes) {
}

public record FindBatchResponse(
    Errors errors,
    List<BatchInfo> batches,
    long logStartOffset,
    long highWatermark) {
}

public record BatchInfo(
    long batchId,
    String objectKey,
    BatchMetadata metadata) {
}

public record BatchMetadata (
    TopicIdPartition topicIdPartition,
    long byteOffset,
    long byteSize,
    long baseOffset,
    long lastOffset,
    long logAppendTimestamp,
    long batchMaxTimestamp,
    TimestampType timestampType,
    long producerId,
    short producerEpoch,
    int baseSequence,
    int lastSequence) {
}

public record ListOffsetsRequest(
    TopicIdPartition topicIdPartition,
    long timestamp
) {

    public static final long EARLIEST_TIMESTAMP = org.apache.kafka.common.requests.ListOffsetsRequest.EARLIEST_TIMESTAMP;
    public static final long LATEST_TIMESTAMP = org.apache.kafka.common.requests.ListOffsetsRequest.LATEST_TIMESTAMP;
    public static final long MAX_TIMESTAMP = org.apache.kafka.common.requests.ListOffsetsRequest.MAX_TIMESTAMP;
    public static final long EARLIEST_LOCAL_TIMESTAMP = org.apache.kafka.common.requests.ListOffsetsRequest.EARLIEST_LOCAL_TIMESTAMP;
    public static final long LATEST_TIERED_TIMESTAMP = org.apache.kafka.common.requests.ListOffsetsRequest.LATEST_TIERED_TIMESTAMP;
}

public record ListOffsetsResponse(
    Errors errors,
    TopicIdPartition topicIdPartition,
    long timestamp,
    long offset
)

public record DeleteRecordsRequest(
    TopicIdPartition topicIdPartition,
    long offset) {
}

public record DeleteRecordsResponse(
    Errors errors,
    long lowWatermark) {
}

public record FileToDelete(
    String objectKey,
    Instant markedForDeletionAt) {
}

public record DeleteFilesRequest(
    Set<String> objectKeyPaths) {
}

public interface BatchCoordinator extends Closeable, Configurable {
    /**
     * This operation is called when a Diskless partition 
     * (or a topic with one or more partitions) is created in the cluster.
     * The Batch Coordinator initializes the corresponding logs.
     *
     * @throws KafkaException if an unexpected error occurs
     */
    void createTopicAndPartitions(
        Set<CreateTopicAndPartitionsRequest> requests);

    /**
     * This operation is called by a broker after uploading the 
     * shared log segment object to the object storage.
     *
     * <p>The Batch Coordinator:
     * <ol>
     * <li>Performs the necessary checks for idempotent produce.
     * <li>Accordingly increases the high watermark of the affected logs.
     * <li>Assigns offsets to the batches.
     * <li>Saves the batch and object metadata.
     * <li>Returns the result to the broker
     *
     * @throws KafkaException if an unexpected error occurs
     */
    List<CommitBatchResponse> commitFile(
        String objectKey,
        int uploaderBrokerId,
        long fileSize,
        List<CommitBatchRequest> batches);

    /**
     * This operation is called by a broker when it needs to serve a Fetch request.
     * <p>The Batch Coordinator collects the batch coordinates to satisfy
     * this request and sends the response back to the broker.
     *
     * @throws KafkaException if an unexpected error occurs
     */
    List<FindBatchResponse> findBatches(
        List<FindBatchRequest> findBatchRequests,
        int fetchMaxBytes);


    /**
     * This operation allows the broker to get the information about log offsets:
     * earliest, latest, etc. The operation is a read-only operation.
     *
     * @throws KafkaException if an unexpected error occurs
     */
    List<ListOffsetsResponse> listOffsets(
        List<ListOffsetsRequest> requests);

    /**
     * This operation is called when a partition needs to be truncated by the user.
     * <p>The Batch Coordinator:
     * <ol>
     * <li>Modifies the log start offset for the affected partitions (logs).
     * <li>Deletes the batches that are no longer needed due to this truncation.
     * <li>If some objects become empty after deleting these batches,
     *     they are marked for deletion as well.
     *
     * @throws KafkaException if an unexpected error occurs
     */
    List<DeleteRecordsResponse> deleteRecords(
        List<DeleteRecordsRequest> requests);

    /**
     * This operation is called when topics are deleted. 
     * It’s similar to deleting records, but all the associated batches 
     * are deleted and the log metadata are deleted as well.
     *
     * @throws KafkaException if an unexpected error occurs
     */
    void deleteTopics(
        Set<Uuid> topicIds);

    /**
     * This operation allows a broker to get a list of soft deleted objects 
     * for asynchronous physical deletion from the object storage.
     *
     * @throws KafkaException if an unexpected error occurs
     */
    List<FileToDelete> getFilesToDelete();

    /**
     * This operation informs the Batch Coordinator that certain soft deleted
     * objects were also deleted physically from the object storage.
     * <p>The Batch Coordinator removes all metadata about these objects.
     *
     * @throws KafkaException if an unexpected error occurs
     */
    void deleteFiles(
        DeleteFilesRequest request);

    boolean isSafeToDeleteFile(
        String objectKey);
}

BatchCoordinator topic-based implementation

FindCoordinator API

The KIP introduces version 7.

Request schema

Version 7 adds the new key type of FindCoordinatorRequest.CoordinatorType.BATCH with value 3, with the key of "operation:topicId:partition". “operation” can be either "write" or "read", to indicate whether the Batch Coordinator needs to write new metadata (write) or it just needs to read metadata.

{
  "apiKey": 10,
  "type": "request",
  "listeners": ["broker"],
  "name": "FindCoordinatorRequest",
  // Version 1 adds KeyType.
  //
  // Version 2 is the same as version 1.
  //
  // Version 3 is the first flexible version.
  //
  // Version 4 adds support for batching via CoordinatorKeys (KIP-699)
  //
  // Version 5 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
  //
  // Version 6 adds support for share groups (KIP-932).
  // For key type SHARE (2), the coordinator key format is "groupId:topicId:partition".
  // Version 7 adds support for batch coordinator (KIP-1164).
  // For key type BATCH (3), the coordinator key format is "operation:topicId:partition", where operation is "read" or "write".
  "validVersions": "0-7",
  "flexibleVersions": "3+",
  "fields": [
    { "name": "KeyLogAppendTime", "type": "stringint64", "versions": "0-3+",
        "about": "The coordinator keylog append timestamp, or -1 if not applicable." },
      { "name": "KeyTypeLogStartOffset", "type": "int8int64", "versions": "10+", "default": "0", "ignorable": false,

        "about": "The log start coordinatoroffset keyof type.the (grouppartition, transaction, share, batch)or -1 on error." },
      { "name": "CoordinatorKeysIsDuplicate", "type": "[]stringbool", "versions": "40+",
        "about": "The coordinator keys." True if this batch was a duplicate of an already committed batch." }
    ]}
  ]
}

Response schema

Version 7 is the same as version 6.

InitDisklessTopics API

Request schema

DisklessDeleteRecords

{
  "apiKey": 9396,
  "type": "request",
  "listeners": ["brokercontroller"],
  "name": "InitDisklessTopicsDisklessDeleteRecordsRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]InitDisklessTopic", "versions// Version 0 is the initial version.
  "validVersions": "0+",
      "aboutflexibleVersions": "Topics to initialize.0+",
  "fields": [
      { "name": "TopicIdTopics", "type": "uuid[]DisklessDeleteRecordsTopic", "versions": "0+",
        "about": "The unique ID of this topic." },Each topic that we want to delete records from.", "fields": [
      { "name": "NumPartitionsTopicId", "type": "int32uuid", "versions": "0+",
        "aboutmapKey": "The number of partitions of this topic" }
      ]true,
    }
  ]
}

Response schema

{
  "apiKey": 93,
  "typeabout": "response",
  "listenersThe unique topic ID." },
      { "name": ["brokerPartitions"],
  "nametype": "InitDisklessTopics[]DisklessDeleteRecordsPartition",
  "validVersionsversions": "0+",
        "flexibleVersionsabout": "0+",
 Each partition that we want to delete records from.", "fields": [
        { "name": "ResponsesPartitionIndex", "type": "[]InitDisklessTopicResultint32", "versions": "0+", "mapKey": true,
          "about": "The results for each topic we tried to initialize.", "fields": [
partition index." },
        { "name": "TopicIdOffset", "type": "uuidint64", "versions": "0+",
          "about": "The unique topic ID."},
   deletion offset. All records with offsets less than this value will be deleted." }
      ]}
    ]},
    { "name": "ErrorCodeTimeoutMs", "type": "int16int32", "versions": "0+",
 "default": "60000",
      "about": "The initialization error, or 0 if the initialization succeededHow long to wait for the deletion to complete, in milliseconds." }
    ]
}
  ]
}

CommitBatches API

Request schema

{
  "apiKey": 9496,
  "type": "requestresponse",
  "listenersname": ["brokerDisklessDeleteRecordsResponse"],
  // Version  "name": "CommitBatches",0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "BrokerIdThrottleTimeMs", "type": "int32", "versions": "0+", "entityTypeignorable": "brokerId"true,
      "about": "The IDduration ofin themilliseconds requestingfor broker." },
    { "name": "Batches", "type": "[]Batches", "versions": "0+",
      "about": "Batches to commit.", "fields": [
  which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "Topics", "type": "[]CommitBatchesTopicDisklessDeleteRecordsTopicResult", "versions": "0+",
        "about": "Each topic that we wantwanted to commitdelete batchesrecords forfrom.", "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+",
 "mapKey": true,
        "about": "The unique topic ID." },
  
      { "name": "Partitions", "type": "[]CommitBatchesTopicPartitionDisklessDeleteRecordsPartitionResult", "versions": "0+",
          "about": "Each partition that we wantwanted to commitdelete batchesrecords forfrom.", "fields": [
          { "name": "PartitionIndex", "type": "int32", "versions": "0+",
            "aboutmapKey": "The partition index." }true,
          { "name": "ObjectKey", "type": "string", "versions": "0+",
            "about": "The key of this object that contains this batchpartition index." },
          { "name": "ByteOffsetLowWatermark", "type": "int32int64", "versions": "0+",
            "about": "The startingpartition byte offset for this batch within the object."low water mark." },
          { "name": "SizeErrorCode", "type": "int64int16", "versions": "0+",
            "about": "Size ofThe deletion error code, or 0 if the batchdeletion in bytessucceeded." },
          { "name": "BaseOffsetErrorMessage", "type": "int64string", "versions": "0+",
 "nullableVersions": "0+", "ignorable": true, "default": "null",
          "about": "The error message, or null if there was no error." }
      ]}
    ]}
  ]
}

DisklessListOffsets

{
  "apiKey": 97,
  "type": "request",
  "listeners": ["controller"],
  "name": "DisklessListOffsetsRequest base offset of this batch."},
          { "name": "LastOffset", "type": "int64", "versions": "0+",
  // Version 0 is the initial  version.
   "aboutvalidVersions": "0"The,
 last offset of this batch (inclusive)."}"flexibleVersions": "0+",
      "fields": [
    { "name": "BatchMaxTimestampTopics", "type": "int64[]DisklessListOffsetsTopic", "versions": "0+",
            "about": "MaxEach timestamptopic orin log append time of this batchthe request."},
 "fields": [
        { "name": "MessageTimestampTypeTopicId", "type": "int8uuid", "versions": "0+",
    "mapKey": true,
        "about": "The messageunique timestamptopic typeID." },
          { "name": "ProducerIdPartitions", "type": "int64[]DisklessListOffsetsPartition", "versions": "0+", "entityType": "producerId", 
            "about": "ProducerIdEach partition ofin thisthe batchrequest."},
 "fields": [
        { "name": "ProducerEpochPartitionIndex", "type": "int16int32", "versions": "0+",
 "mapKey": true,
          "about": "The current epoch associated with the producer IDpartition index." },
          { "name": "BaseSequenceTimestamp", "type": "int32int64", "versions": "0+",
            "about": "Base sequence number of this batch."},",
          "about": "The timestamp to query. Use -2 for earliest, -1 for latest, -3 for max timestamp, -4 for earliest local, -5 for latest tiered." }
      ]}
    ]},
    { "name": "LastSequenceTimeoutMs", "type": "int32", "versions": "0+",
      "default": "60000",
      "about": "Last sequence number of this batch."How long to wait in milliseconds before timing out the request." }
  ]
}

{
  "apiKey": 97,
   ]}"type": "response",
  "name": "DisklessListOffsetsResponse",
  // ]}
Version 0 is the ]}
  ]
}

Response schema

{initial version.
  "apiKeyvalidVersions": 94"0",
  "typeflexibleVersions": "response0+",
  "listenersfields": ["broker"],
    { "name": "CommitBatchesThrottleTimeMs",
  "type": "int32", "validVersionsversions": "0+", "ignorable": true,
      "flexibleVersionsabout": "0+",
  "fields": [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": "Topics", "type": "[]CommitBatchesTopicResultsDisklessListOffsetsTopicResponse", "versions": "0+",
      "about": "TheEach resulttopic forin eachthe topicresponse.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+", "mapKey": true,
        "about": "The unique topic ID." },
      { "name": "Partitions", "type": "[]CommitBatchesTopicPartitionResultsDisklessListOffsetsPartitionResponse", "versions": "0+",
        "about": "TheEach resultpartition forin eachthe partitionresponse.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0+", "mapKey": true,
          "about": "The partition index." },
        { "name": "ErrorCode", "type": "int16", "versions": "0+",
          "about": "The resultpartition error code, or zero if there was no error." },
        { "name": "AssignedBaseOffset", "type": "int64", "versions": "0+",
          "about": "The assigned base offset." if there was no error." },
        { "name": "LogAppendTimeErrorMessage", "type": "int64string", "versions": "0+", "nullableVersions": "0+", "ignorable": true, "default": "null",
          "about": "The timestamp returned by broker after appending the messages."error message, or null if there was no error." },
            { "name": "LogStartOffsetTimestamp", "type": "int64", "versions": "0+", "default": "-1",
          "about": "The log start timestamp associated with the returned offset." },
        { "name": "IsDuplicateOffset", "type": "boolint64", "versions": "0+",
          "aboutdefault": "Whether the batch was already written by the same producer id.""-1",
          "about": "The returned offset." }
      ]}
    ]}
  ]
}

DeleteDisklessTopics API

Request schema

DisklessFindBatches

{
  "apiKey": 9598,
  "type": "request",
  "listeners": ["brokercontroller"],
  "name": "DeleteDisklessTopicsDisklessFindBatchesRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
 fields": [
    { "name": "FetchMaxBytes", "type": "int32", "versions": "0+",
      "about": "Maximum bytes to fetch across all partitions." },
    { "name": "TopicsMaxBatchesPerPartition", "type": "[]DeleteDisklessTopicint32", "versions": "0+",
       "about": "TheMaximum IDnumber of batches theper topicspartition to deletereturn.", "fields": [
    },
    { "name": "TopicIdTopics", "type": "uuid[]DisklessFindBatchesTopic", "versions": "0+",
         "about": "TheEach topic uniquein topicthe IDrequest." }
  , "fields": [
   ]}
   ]
}

Response schema

{
  "apiKeyname": 95"TopicId",
  "type": "responseuuid",
  "listenersversions": ["broker0+"],
  "namemapKey": "DeleteDisklessTopics"true,
        "validVersionsabout": "0",
The topic "flexibleVersionsID.": "0+"},
  "fields": [
    { "name": "ResponsesPartitions", "type": "[]DeleteDisklessTopicResultDisklessFindBatchesPartition", "versions": "0+",
        "about": "TheEach resultspartition forin each topic we tried to deletethe request.", "fields": [
        { "name": "TopicIdPartitionIndex", "type": "uuidint32", "versions": "0+", "mapKey": true,
          "about": "The uniquepartition topic IDindex." },
        { "name": "ErrorCodeOffset", "type": "int16int64", "versions": "0+",
          "about": "The deletionoffset error,to orstart 0 if the deletion succeededfetching from." }
    ]}
  ]
}

DeleteDisklessRecords API

Request schema

{
  "apiKey": 96,
  "type": "request",
  "listeners": ["broker"],
  "name": "DeleteDisklessRecords",
  "validVersions,
        { "name": "MaxPartitionFetchBytes", "type": "int32", "versions": "0+",
          "flexibleVersionsabout": "0+",Maximum bytes to fetch for this partition." }
      ]}
  "fields": [  ]},
    { "name": "TopicsTimeoutMs", "type": "[]DeleteDisklessTopicRecordsint32", "versions": "0+", "default": "60000",
      "about": "Each topic that we want to delete records from.", "fields": [
      { "name": "TopicId",How long to wait in milliseconds before timing out the request." }
  ]
}

{
  "apiKey": 98,
  "type": "uuidresponse",
  "versionsname": "0+DisklessFindBatchesResponse",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"The,
 topic ID."flexibleVersions": }"0+",
  "fields": [
    { "name": "PartitionsThrottleTimeMs", "type": "[]DeleteDisklessPartitionRecordsint32", "versions": "0+",
 "ignorable": true,
      "about": "Each partition that we wantThe duration in milliseconds for which the request was throttled due to deletea recordsquota from.", "fields": [
    violation, or zero if the request did not violate any quota." },
    { "name": "PartitionIndexTopics", "type": "int32[]DisklessFindBatchesTopicResponse", "versions": "0+",
          "about": "The partition indexEach topic in the response." },
 "fields": [
      { "name": "OffsetTopicId", "type": "int64uuid", "versions": "0+",
 "mapKey": true,
        "about": "The deletiontopic offsetID." },
      ]}
    ]}
  ]
}

Response schema

{
  "apiKeyname": 96"Partitions",
  "type": "response[]DisklessFindBatchesPartitionResponse",
  "listenersversions": ["broker0+"],
        "nameabout": "DeleteDisklessRecords",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
Each partition in the response.", "fields": [
        { "name": "ResponsesPartitionIndex", "type": "[]DeleteDisklessRecordsResultint32", "versions": "0+", "mapKey": true,
          "about": "The results for each topic we tried to delete records for.", "fields": [
partition index." },
        { "name": "TopicIdErrorCode", "type": "uuidint16", "versions": "0+",
          "about": "The unique topic ID."partition error code, or 0 if there was no error." },
        { "name": "PartitionsLogStartOffset", "type": "[]DeleteDisklessPartitionRecordsResultint64", "versions": "0+", "default": "-1",
          "about": "The resultslog forstart eachoffset topicof wethe triedpartition, toor delete-1 recordsif forunknown.", "fields": [},
        { "name": "PartitionIndexHighWatermark", "type": "int32int64", "versions": "0+", "default": "-1",
          "about": "The high watermark of the partition index, or -1 if unknown." },
        { "name": "ErrorCodeBatches", "type": "int16[]DisklessFindBatchesBatchInfo", "versions": "0+",
          "about": "The deletionbatches error,found orfor 0 if the record deletion succeeded." }
  this partition.", "fields": [
    ]}
    ]}
  ]
}

FindBatches API

Request schema

{
  "apiKeyname": 97,
  "type": "requestBatchId",
  "listenerstype": ["broker"],
  "name": "FindDisklessBatches"int64",
  "validVersionsversions": "0+",
            "flexibleVersionsabout": "0+"The batch ID." },
     "fields": [
    { "name": "TopicsObjectKey", "type": "[]FindDisklessBatchTopicstring", "versions": "0+",
            "about": "EachThe topicobject thatkey wein wantobject tostorage findcontaining batchesthis forbatch." }, "fields": [

          { "name": "TopicIdMagic", "type": "uuidint8", "versions": "0+",
            "about": "The topic IDrecord batch magic byte." },
          { "name": "PartitionsByteOffset", "type": "[]FindDisklessBatchTopicPartitionint64", "versions": "0+",
            "about": "EachThe partitionbyte thatoffset weof wantthe tobatch findwithin batchesthe forobject." },
 "fields": [
        { "name": "PartitionIndexByteSize", "type": "int32int64", "versions": "0+",
            "about": "The partition index size of the batch in bytes." },
          { "name": "OffsetBaseOffset", "type": "int64", "versions": "0+",
            "about": "The startingbase offset of the record batch." },
          { "name": "MaxBytesLastOffset", "type": "int32int64", "versions": "0+",
            "about": "The maximumlast bytesoffset to fetch for this of the topicrecord partitionbatch." },
      ]}
    ]},
    { "name": "MaxBytesLogAppendTimestamp", "type": "int32int64", "versions": "0+",
            "about": "The maximumlog bytes to fetch for all the topic partition." }
  ]
}

Response schema

{
  "apiKey": 97,
  "type": "response",
  "listeners": ["broker"],
  "name": "FindDisklessBatches",
  "validVersionsappend timestamp." },
          { "name": "BatchMaxTimestamp", "type": "int64", "versions": "0+",
            "flexibleVersionsabout": "0+",
  "fields": [
The maximum timestamp in the batch." },
          { "name": "ResponsesTimestampType", "type": "[]FindDisklessBatchTopicResultint8", "versions": "0+",
      "about": "The results for each topic we searched batches for.", "fieldsabout": [
"The timestamp type (0   { "name": "TopicId", "type": "uuid", "versions": "0+",
= CreateTime, 1 = LogAppendTime)." }
        ]}
  "about": "The unique topic ID."]},
    ]}
  ]
}

DisklessDescribeFiles

{
 "name "apiKey": 100,
  "type": "Partitionsrequest",
  "typelisteners": ["controller"[]FindDisklessBatchTopicPartitionResult",
  "versionsname": "0+DisklessDescribeFilesRequest",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"The,
 results for each topic we searched batches for.", "flexibleVersions": "0+",
  "fields": [
        { "name": "PartitionIndexFiles", "type": "int32[]DisklessDescribeFileEntry", "versions": "0+",
          "about": "TheEach file partitionto indexdescribe." },
 "fields": [
      { "name": "ErrorCodeObjectKey", "type": "int16string", "versions": "0+",
 "mapKey": true,
        "about": "The object error,key oridentifying 0the iffile thein searchobject succeededstorage." },
    ]},
    { "name": "LogStartOffsetTimeoutMs", "type": "int64int32", "versions": "0+",
    "default": "60000",
      "about": "TheHow long currentto logwait start offset."},
        { "name": "HighWatermark",in milliseconds before timing out the request." }
  ]
}

{
  "apiKey": 100,
  "type": "int64response",
  "versionsname": "0+DisklessDescribeFilesResponse",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"The,
 current high water mark."}"flexibleVersions": "0+",
    "fields": [
    { "name": "ObjectKeyThrottleTimeMs", "type": "stringint32", "versions": "0+",
            "about": "The key of this object that contains this batch." }
   "ignorable": true,
     { "nameabout": "ByteOffset", "type": "int32", "versions": "0+",
          "about": "The starting byte offset for this batch within the object."},
    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": "SizeFiles", "type": "int64[]DisklessDescribeFileResult", "versions": "0+",
          "about": "SizeResults offor the batch in byteseach file."},
 "fields": [
      { "name": "BaseOffsetObjectKey", "type": "int64string", "versions": "0+",
 "mapKey": true,
        "about": "The base offset of this batch."object key identifying the file in object storage." },
        { "name": "LastOffsetErrorCode", "type": "int64int16", "versions": "0+",
          "about": "The last offset of this batch (inclusive)."},
   error code, or 0 if there was no error." },
      { "name": "BatchMaxTimestampErrorMessage", "type": "int64string", "versions": "0+", "nullableVersions": "0+",
   "ignorable": true, "default": "null",
        "about": "Max timestampThe error message, or lognull appendif timethere ofwas thisno batcherror." },
        { "name": "MessageTimestampTypeHasLiveBatches", "type": "int8bool", "versions": "0+",
          "about": "True if the file still has live batches tracked by this "The message timestamp type"},
        { "name": "ProducerId",coordinator." }
    ]}
  ]
}

Metadata records

DisklessPartitionCreatedRecord

{
  "apiKey": 100,
  "type": "int64data",
 "versions": "0+", "entityTypename": "producerIdDisklessPartitionCreatedRecord", 
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"ProducerId,
 of this batch."}"flexibleVersions": "0+",
    "fields": [
    { "name": "ProducerEpochTopicId", "type": "int16uuid", "versions": "0+",
          "about": "The current epoch associated with the producerunique topic ID." },
        { "name": "BaseSequenceName", "type": "int32string", "versions": "0+",
 "entityType": "topicName",
        "about": "Base sequence number of this batch." "The topic name." },
        { "name": "LastSequencePartitionIndex", "type": "int32", "versions": "0+",
          "about": "LastThe sequence number of this batch."}
      ]}
    ]partition index." }
  ]
}

ListDisklessOffsets API

Request schema

DisklessTopicDeletedRecord

{
  "apiKey": 98101,
  "type": "request",
  "listeners": ["broker"]data",
  "name": "ListDisklessOffsetsDisklessTopicDeletedRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]ListDisklessOffsetsTopic", "versions// Version 0 is the initial version.
  "validVersions": "0+",
      "aboutflexibleVersions": "Each topic that we want to list offsets for.",0+",
  "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The unique topic ID." },
  ]
}

DisklessFileCommittedRecord

TODO: optimize for storing fewer bytes per record

{
  "apiKey": 102,
 { "nametype": "Partitionsdata",
  "typename": "[]ListDisklessOffsetsTopicPartitionDisklessFileCommittedRecord",
  // Version "versions0 is the initial version.
  "validVersions": "0+",
        "aboutflexibleVersions": "Each partition that we want to list offsets for.",0+",
  "fields": [
        { "name": "PartitionIndexObjectKey", "type": "int32string", "versions": "0+",
          "about": "The partition index object key identifying the file in object storage." },
        { "name": "TimestampObjectFormat", "type": "int64int8", "versions": "0+",
          "about": "The currentformat timestamp." }
      ]}
    ]}
  ]
}

Response schema

{
  "apiKey": 98,
  "type": "response",
  "listeners": ["broker"],
  "name": "DeleteDisklessRecords",
  "validVersionsof the object (e.g. 1 = WRITE_AHEAD_MULTI_SEGMENT)." },
    { "name": "FileSize", "type": "int64", "versions": "0+",
      "flexibleVersionsabout": "0+",
  "fields": [The total size of the file in bytes." },
    { "name": "ResponsesUploaderBrokerId", "type": "[]ListDisklessOffsetsTopicResultint32", "versions": "0+", "entityType": "brokerId",
      "about": "The resultsbroker forthat eachuploaded topic we tried to list offsets forthe file.", "fields": [
   },
    { "name": "TopicIdOwnerIds", "type": "uuid[]int32", "versions": "0++", "entityType": "brokerId",
        "about": "The unique topic ID." list of Diskless Coordinator IDs that share ownership of this file." },
      { "name": "PartitionsTopics", "type": "[]ListDisklessOffsetsTopicPartitionResultDisklessCommittedTopic", "versions": "0+",
        "about": "The results for eachEach topic wewith triedaccepted tobatches listin offsetsthis forfile.", "fields": [
        { "name": "PartitionIndexTopicId", "type": "int32uuid", "versions": "0+",
          "about": "The partitionunique topic indexID." },
        { "name": "ErrorCodePartitions", "type": "int16[]DisklessCommittedPartition", "versions": "0+",
          "about": "TheEach deletionpartition error,with oraccepted 0batches iffor the record list offsets succeeded." },this topic.", "fields": [
        { "name": "TimestampPartitionIndex", "type": "int64int32", "versions": "0+",
          "about": "The timestamp associated with the returned offsetpartition index." },
        { "name": "OffsetBatches", "type": "int64[]DisklessCommittedBatch", "versions": "0+",
          "about": "The returned offset accepted batches for this partition." },
 "fields": [
    ]}
    ]}
  ]
}

AssignPartitionToBatchCoordinator API

Request schema

{
  "apiKeyname": 99"Magic",
  "type": "requestint8",
  "listenersversions": ["controller0+"],
            "nameabout": "AssignPartitionToBatchCoordinator",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
The record batch magic byte." },
          { "name": "TopicsByteOffset", "type": "[]AssignTopicToBatchCoordinatorint32", "versions": "0+",
,
            "about": "EachThe topicbyte thatoffset weof wantthis tobatch assignwithin Batchthe Coordinator tofile." }, "fields": [

          { "name": "TopicIdByteSize", "type": "uuidint32", "versions": "0+",
            "about": "The topic ID size of the batch in bytes." },
          { "name": "PartitionsAssignedBaseOffset", "type": "[]AssignTopicPartitionToBatchCoordinatorint64", "versions": "0+",
            "about": "EachThe partitionbase thatoffset weassigned wantby to assign Batch Coordinator tothe coordinator." },
 "fields": [
        { "name": "PartitionIndexAssignedLastOffset", "type": "int32int64", "versions": "0+",
            "about": "The partition index last offset assigned by the coordinator." },
          { "name": "BatchCoordinatorPartitionIndexLogAppendTimestamp", "type": "int32int64", "versions": "0+",

            "about": "The Batch Coordinator partition index log append timestamp assigned by the coordinator." },
      ]}
    ]}   
  ]
}

Response schema

{
  "apiKeyname": 99"BatchMaxTimestamp",
  "type": "responseint64",
  "listenersversions": ["broker0+"],
    "name        "about": "AssignPartitionToBatchCoordinator",
  "validVersions": "0"The maximum timestamp in the batch." },
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ResponsesTimestampType", "type": "[]AssignTopicToBatchCoordinatorResultint8", "versions": "0+",
            "about": "The results for each topic we tried to assign Batch Coordinator to.", "fields": [
 "The timestamp type (0 = CreateTime, 1 = LogAppendTime)." },
          { "name": "TopicIdProducerId", "type": "uuidint64", "versions": "0+",
            "about": "The unique topic ID." producer ID, or -1 if not set." },
          { "name": "PartitionsProducerEpoch", "type": "[]ListDisklessOffsetsTopicPartitionResultint16", "versions": "0+",
            "about": "The resultsproducer forepoch, eachor topic-1 weif tried to assign Batch Coordinator tonot set." },
 "fields": [
        { "name": "PartitionIndexBaseSequence", "type": "int32", "versions": "0+",
            "about": "The partition index": "The base sequence number, or -1 if not set." },
          { "name": "ErrorCodeLastSequence", "type": "int16int32", "versions": "0+",
            "about": "The last sequence errornumber, or 0-1 if thenot assignment succeededset." }
        ]}
      ]}
    ]}
  ]
}

...

DisklessRecordsDeletedRecord

{
  "apiKey": 29103,
  "type": "metadatadata",
  "name": "TopicPartitionToBatchCoordinatorMappingDisklessRecordsDeletedRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
   { "name": "PartitionIndex", "type": "int32", "versions": "0+",
     "about": "The partition index." },
// Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicId", "type": "uuid", "versions": "0+",
      "about": "The unique topic ID of the topic." },
    { "name": "BatchCoordinatorPartitionIndexPartitionIndex", "type": "int32", "versions": "0+",
      "about": "The Batch Coordinator partition index." },
    {  ]
}

Future Work

Having many diskless topics with infinite retention would mean that metadata keeps growing infinitely. This leads to metadata partitions having to store large amounts of metadata, which also impacts Batch Coordinator startup time. Enabling Tiered Storage on the __diskless-metadata topic is a possible way to mitigate these issues. Another possible way to solve these problems is to introduce the concepts of Diskless Metadata Snapshots. These would be periodic snapshots of a metadata partition that can be stored on object storage, so that the internal topic only needs to store the deltas compared to the last available snapshot.

Compatibility, Deprecation, and Migration Plan

Batch Coordinator is only needed for diskless topics. No migration is therefore needed.

Test Plan

The default implementation of the Batch Coordinator will be tested in-depth with unit tests. Performance tests will be done in order to understand how many requests a single Batch Coordinator is able to sustain, which is a useful metric to consider in regards to deciding how many Batch Coordinator is necessary to have in a deployment.

Tests of the Batch Coordinator in an integrated system will be performed through the integration and system tests that are planned for KIP-1163.

Rejected Alternatives

Topic-based Batch Coordinator using high number of partitions by default

In this option the Batch Coordinator topic is created by default with a high number of partitions, for example 50 like the other coordinators already available.

The advantage of this option is that the assignment of a topic-partition to its Batch Coordinator can be static, removing the need for storing the assignment inside KRaft metadata.

The drawback of this option instead consists of the high fan-out of requests from brokers to Batch Coordinators. Having 50 partitions means that there are 50 Batch Coordinators, and this drastically increases the likelihood of one broker having to contact every broker in the cluster every time a new Shared Log Segment Object is created. Every Shared Log Segment Object contains multiple topic-partitions, therefore in order to commit this object it’s necessary to contact all the Batch Coordinators that are assigned to the topic-partitions present in the object.

This KIP instead proposes using only one Batch Coordinator by default to avoid paying the high fan-out cost even when it’s not necessary for a cluster to have many Batch Coordinators.

Use an external storage for metadata

"name": "NewLogStartOffset", "type": "int64", "versions": "0+",
      "about": "The new log start offset after deletion. All batches with last offset less than this value are removed." }
  ]
}

Configuration

Broker configuration

More to be added as discussion unrolls

Monitoring

More to be added as discussion unrolls

Compatibility, Deprecation, and Migration Plan

There’s no impact on existing users and no existing behavior will be changed, no migration will be required.

Test Plan

The feature will be thoroughly tested with unit, integration and system tests. We will also carry out performance testing both to understand the performance of the Diskless coordinator, and also to understand the impact on brokers with it.

Rejected Alternatives

Use external system

To keep Kafka self-sufficient, we don’t propose to use an external system for managing the Diskless metadata.

Use cluster KRaft quorum

It is possible to put Diskless metadata in the same KRaft quorum that manages the cluster metadata. This was rejected because this would mean mixing low-throughput relatively slow changing cluster metadata with high-throughput Diskless metadata, which may potentially create performance issues with such critical operations like partition leadership changes.

Use separate KRaft quorum

Instead of using the cluster KRaft quorum, it’s possible to run a separate quorum exclusively for Diskless metadata. The proposed solution relies on some useful mechanisms from KRaft like snapshotting. However, we don’t see any benefit in using the real KRaft mechanism over a normal Kafka topic for this taskOne can argue that a Kafka topic is not the most convenient medium for storing this type of metadata. Using a relational or key-value database may make it more convenient to implement and provide certain performance benefits. We reject this alternative because Kafka tends to be self-sufficient and not rely on external systems (see e.g. the ZooKeeper removal in KIP-500). However, the pluggability of the batch coordinator allows using such implementations.