This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
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]
Authors: Henry Cai, Thomas Thornton, Greg Harris
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
KIP-405 introduced remote tiered storage for Kafka for regular append-only logs. The support for compacted topics (log representing compacted K/V storage) is deferred for future KIPs. Here we propose a design to support compacted topics in tiered storage.
Configurations: This feature enables the combination of remote.storage.enable=true with cleanup.policy=compact. Previously, this combination would cause an exception to be thrown.
Interfaces: No changes to the client protocol or binary log format. No changes to existing CLI tools.
First we start with the background information on how current log compaction works and the lifecycle of the log segment files. In the next section we will describe how we enhance some of the steps.
We briefly describes the life cycles for append-only logs in tiered storage:
LogRetention thread will periodically delete the log segment files (and their accompanying index files) whose age passes local.retention.ms/bytes
This Confluent wiki has details on how log compaction works. In summary, this is the high level steps:
Log compaction on local disks is a well-established process. However, supporting it in tiered storage introduces complexity regarding remote storage, transaction handling and multi-segment processing.
To ensure stability and correctness, we opt for a design that maintains the existing log cleaning logic with minimal changes. The LogCleaner will:
This approach avoids implementing complex logic specific to tiered storage, and uses already well-tested local compaction code.
There will be minimal changes to LogCleaner code except we will download the remote log segment to local before the cleaning starts and upload the cleaned log segment after the cleaning is done.
The above diagram shows the local/remote log segments before and after the log compaction.
For each log for a topic partition, from the perspective of log compaction we can look at the log as composed of 5 contiguous regions:
Before compaction, log segments (A/B/C) are already uploaded to remote storage; Some log segments (B2/C/D/E) are at the moment stored in the local filesystem. Note a portion of the dirty region (B2) is still at the local storage.
During compaction, we first build an in-memory index map with key as the message key and the value as the last offset location of the message; And then we rewrite the log segments for all A+B using this map, for the messages whose key is present in the map it will only be written out if its offset matches with the offset value in the map. Note we only need to build the index map for the messages in region B but the log rewriting needs to happen for regions A and B.
After compaction, both region A and B are cleaned with no duplicate message keys and checkpoint location is moved to the end of B. Note the log segments of A,B after compaction might have different content than the ones before compaction since some messages might be compacted out.
For the log segments which still have a local presence (e.g. B2), we don’t have to fetch remote log segments. For the remote log segments which are being changed, a remote file delete followed by a remote file create/upload will occur, those will be detailed in a later section. There is also optimization of not downloading remote log segments if its content has no interaction with the index map.
For the current compacted topic setup with cleanup.policy=compact, Kafka never deletes local log segments based on time (only compaction reduces size). If we maintained this behavior, tiered storage would offer no benefit since all log segments are present on the local filesystem and there is no need to look up data in remote storage.
To make tiered storage usable in this setup, we will also perform time-based local retention for tiered compacted topics:
Local segments are deleted when they exceed local.retention.ms/bytes, same as append-only logs. The remote compacted segments persist according to the overall retention.ms
Local segments are deleted when they exceed local.retention.ms/bytes, regardless of compaction status. This means:
Local deletion is time-based only and does not wait for compaction to complete. This ensures
We make this trade-off intentionally to ensure durability is not compromised. This may increase downloads if local retention is consistently shorter than compaction. However, this is acceptable since compaction inherently requires re-processing old segments from remote storage (see later sections in Detailed Design on how we optimize this).
We will discuss some of the detailed design choices. First we will walk through the lifecycle for compacted topics in tiered storage and compare the flow with the local topic compaction.
When a compacted topic is enabled for remote tiered storage:
When the current active local log segment file is full and LogManagerrolls to a new log segment file, the log rotation event occurs. LogManager will close the log segment file and its index files and upload them to remote tiered storage through RemoteStorageManager.
The process is the same as the handling for append-only logs.
The LogCleaner thread runs at regular intervals:
For each logs to clean, group the log segmentsThis logic is the same as existing code
To prevent disk exhaustion when downloading multiple remote segments for compaction, we process the work in bounded chunks.
Chunk size is defined as min(log.segment.bytes, available_disk_space / 3). This ensures there is room for downloaded segments, offset map, and output segments.
To avoid downloading remote segments that contain no keys in the current offset map, we will build a bloom filter index per remote log segment.
On the local file filesystem, the swap stage consists of several stages:
On the remote storage, the swap is relatively simpler since we don’t directly access remote log segment files. Instead, we query RemoteLogMetadataManager for segment locations based on offset and leader epoch. The swap sequence is:
As long as we include the correct leader epoch and offset information in the RLMetadata when we upload the newly rewritten log segment file, the RemoteLogMetadataManager will update its cache to route the retrieval request to the latest segment file; the old log segment file essentially becomes an orphan node.
Although we could delete the old log segment file immediately after uploading the new segment, we defer deletion to the log retention thread. This ensures deletions are validated against the current leader epoch lineage, preventing zombie brokers from deleting segments that are still valid for the new leader. Orphaned segments from invalid compactions are filtered out by the same epoch validation mechanism used for checkpoint fencing (See the later discussion on fencing old broker on cleaner-offset-checkpoint file).
Same as existing code
Same as existing code. Tombstone records (value=null) are processed during compaction.
Same as existing code. All existing transaction handling logic is preserved
We will then discuss some of the design considerations.
For tiered storage topics, we will modify LogToClean to include remote segment sizes in the totalBytes calculation.
To method to determine the sum of remote segment sizes will:
For the classic topic, the log cleaning/compaction is happening independently on the leader and the follower broker at the same time. If we use the same approach for the tiered storage topic, the follower broker will need to download all the old log segments from remote storage on each cleaning cycle which is not desirable. Instead the follower broker will only do log cleaning/compaction on its local log segment files.
When the leadership switches to the follower broker, the follower needs to adjust its cleaner offset checkpoint to reflect the position from the old leader since this checkpoint marks all the messages before the mark has been cleaned. For this reason, the leader broker will need to persist its cleaner-offset-checkpoint on both local disk as well as remote tiered storage so this state can be transitioned to the follower on the leadership switch event.
When the leadership switches from broker A to broker B, broker B needs to figure out the current state of log compaction on the remote storage and what is the cleaner-offset-checkpoint on the overall log for the given topic partition. For this reason, broker A needs to persist its cleaner-offset-checkpoint to remote storage at the end of each compaction cycle. However we cannot simply upload the checkpoint file directly onto remote storage since broker A might lose the leadership by force because it became an unresponsive zombie. After the controller moves the leadership to broker B and while broker B is working on log compaction and uploading its new cleaner-offset-checkpoint file, broker A can become live again and try to finish uploading the old checkpoint file and thus overwrite the new checkpoint file.
To fence off the rogue leader, we will extend the concept of cleaner-offset-checkpoint in remote storage to be a vector of entries with each entry of tuple (leader-epoch, cleaner-offset-checkpoint-for-the-epoch). Those entries established the lineage of cleaner-offset-checkpoint with regards to leader epoch change. This design models the leader-epoch checkpoint file introduced in KIP-101.
The vector of cleaner-offset-checkpoint in remote storage will be persisted as part of the RemoteLogSegmentMetadata (very similar to how we include segmentLeaderEpochs in the LogSegmentMetadata):
public RemoteLogSegmentMetadata(RemoteLogSegmentId remoteLogSegmentId,
...
Map<Integer, Long> segmentLeaderEpochs,
Map<Integer, Long> cleanerOffsets) { |
The leader broker will publish out this list of cleanOffsets for each log segment it is uploading to remote storage. The value of the cleaner checkpoint corresponds to each compaction cycle and increases as the compaction cycle continues.
The RemoteStorageMetadataManager will have a cache of leaderEpoch/cleanerOffset as it tracks the RemoteLogMetadata publishing, much like how it builds the cache of leaderEpoch/highestOffset. The cache tracks the max valid cleaner offset for each leader epoch. Each uploaded segment contains the current cleanerOffset value for all leader epochs present in that segment, so the latest segment provides the checkpoint values (without needing to scan across multiple segments).
When the follower broker becomes the leader it will need to find the current cleaner-offset-checkpoint on the remote storage. It will look at the previous leaderEpoch and ask the RemoteStorageMetadataManager for the cleanerOffset for that epoch, this is very similar to how it currently resolves the highest offset for a leader epoch during leadership switch.
The follower also needs to remove its local log segments before the cleaner-offset-checkpoint to make its local content consistent with the remote storage.
When the follower proceeds with compaction and publishes its new cleaner-offset it will use its new leader epoch and essentially makes any publishing from zombie old broker invalid.
The following table shows the content of remote log metadata when the leadership switches from broker A to broker B. In the table, CO stands for cleaner-offset, LE stands for leader-epoch (broker A has LE-0 while broker B has LE-1), the cleaner-offset-map content of {LE-0 -> CO-100, LE-1 -> CO-155} means cleaner-offset for leader-epoch-0 ends at offset-100, cleaner-offset for leader-epoch-1 ends at offset-155.
Event | Broker A (Old Leader) | Broker B (New Leader) | RL Metadata | Note |
A finishes one compaction | Initial cleaner offset map: {LE-0:} Publish Seg-0, CO at 100 | Seg-0: {LE-0->CO-100} | ||
A becomes unresponsive and B becomes leader | Resolve cleaner-offset map as: {LE-0 -> CO-100, LE-1 ->} | |||
B finishes one compaction | Upload seg-2 and CO at 155 at the moment | Seg-2: {LE-0->CO-100, LE-1->CO-155} | ||
A becomes live and upload its pending compaction | Upload seg-1 and CO at 123 | Seg-1: {LE-0->CO-123} | Seg-1 & Seg-2 can arrive in any order. Regardless of arrival order, Seg-2’s cleaner offset map (LE-0-> 100, LE-1-> 155) establishes that LE-0 cleaned up to 100, and LE-1 cleaned up to 155. This means LE-0’s valid range ended at offset 100, invalidating Seg-1’s claim that LE-0 reached offset 123. The cache validation rejects Seg-1 regardless of arrival order. | |
Future reads from RemoteStorage | Will not read Seg-2: {LE-0->CO-100, LE-1->CO-155} |
Similar Fencing protection also needs to happen to the remote log segment upload from the old broker as well as the deletion of the old remote log segment. For remote log segment deletion, we will need to perform a tombstone operation instead of outright file deletion since that deletion request can come from a zombie broker A (while active broker B still needs to read the data from that log segment). So the metadata manager will mark the remote log segment as to be deleted (RemoteLogState.DELETE_SEGMENT_STARTED), then validates them using isRemoteSegmentWithinLeaderEpochs() before actual deletion. This prevents zombie brokers from deleting segments still valid under the current leader.
Besides the normal two-pass processing to compact the log segment in classic topic, there is extra overhead working with remote tiered storage:
When the cleanup policy is set to compact,delete: the local log segment file can still be removed by the LogRetention thread when the condition is met. This is the same as for append-only logs
When the retention policy is set to compact: we will remove local log segment file/indexes when all the messages in the file passed local.retention.ms and all messages in the file are cleaned (offset < cleaner-checkpoint).
If the cleanup policy of the topic is compact,delete, the remote log segment file can also be removed when the condition is met. This is the same as for append-only logs;
Unit tests will verify that LogCleaner correctly identifies and fetches remote segments, and that deletion and upload of compacted log segments occur.
Integration tests will verify the creation of remote storage topics with compaction enabled, produce data with duplicate keys to force compaction, and verify data on remote storage is compacted, and that consumer reads only receive the messages post-compaction.
System tests will be run with `compact,delete` to verify that both policies when enabled simultaneously are stable under load.
An alternative design is to delay the log segment file uploading to remote tiered storage until the file finishes compaction/cleaning. This design would reduce the need to download the file from remote, going through compaction and re-uploading. However this design deeply couples the LogRotation/Uploading with LogCleaning/LogRetention, when the log cleaning is throttled, errored out or blocked for other reasons we cannot upload the file to remote which affects durability of the log segment and runs the danger of running out disk space. Similarly any slowdown or errors during log retention processing (removing log segment) would also block the log file uploading and runs the danger of running out of disk space.