Authors: Henry Cai, Thomas Thornton

Status

Current state: Under Discussion

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

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

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

Motivation

In KIP-405, the community has proposed and implemented the tiered storage for old Kafka log segment files, when the log segments is older than local.retention.ms, it becomes eligible to be uploaded to cloud's object storage and removed from the local storage thus reducing local storage cost.  KIP-405 only uploads older log segments but not the most recent active log segments (write-ahead logs). Thus in a typical 3-way replicated Kafka cluster, the 2 follower brokers would still need to replicate the active log segments from the leader broker. It is common practice to set up the 3 brokers in three different AZs to improve the high availability of the cluster. This would cause the replications between leader/follower brokers to be across AZs which is a significant cost (various studies show the across AZ transfer cost typically comprises 50%-60% of the total cluster cost). Since all the active log segments are physically present on three Kafka Brokers, they still comprise significant resource usage on the brokers. The state of the broker is still quite big during node replacement, leading to longer node replacement time. KIP-1150 recently proposes diskless Kafka topic, but leads to increased latency and a significant redesign. In comparison, this proposed KIP maintains identical performance for acks=1 producer path, minimizes design changes to Kafka, and still slashes cost by an estimated 43%.

One of the reasons why Kafka tiered storage is not using object storage for active log segments is due to the latency and cost of uploading active log segments to traditional object storage (e.g. Amazon S3).  A typical S3 object upload usually takes 100ms and the cost of uploading will quickly add up astronomically if we need to upload many small log segments to S3 to keep up with the incoming write traffic to a Kafka topic.

However, with the new development of S3 express one Zone bucket (S3E1Z) and EBS, there is a possible path to store active log segment files to the cloud.

Proposed Changes

In this KIP, we propose to extend the tiered storage support for active log segment (write-ahead logs) as well. The active log segments will be eligible to be uploaded onto object storage once they get to a certain size or pass a certain retention time.  

Overall architecture and Fast Tiered Object Storage

The overall architecture is mainly centered around using background tasks to upload a section of active log segments from the leader broker to the object storage and download them onto the follower broker.  As a result the follower broker no longer directly reads the data from the leader broker during FetchRequest/Response flow. Instead the data flows from the leader broker to the object storage and then to the follower broker without paying for across-AZ transfer cost.

For the active log segments, due to the fast and frequent updating nature of those write ahead logs, we would need to upload them into a fast object storage.  In terms of object cloud storage choice, there are fast storage types that  provide single digit ms upload latency and with about half of the cost of file uploading. On the other hand, EBS would also be a good candidate for storing active Kafka log segments since EBS also offers single digit ms latency access and offers free transfer cost if the IOPS is under a certain threshold. EBS requires additional set up to allow for multiple hosts to read the same volume, while S3E1Z works out of the box. S3E1Z and EBS are the choice in AWS world, GCS and Azure have similar competing products as well.

With the advancement of S3E1Z bucket and EBS, we can choose to replicate the most recent active log segment (we call them WriteAheadLog shortnamed WAL Log) onto those fast short-term cloud storage.  Using S3E1Z bucket to store active log segment files is not a new idea and in fact several commercial cloud native Kafka offerings (Confluent Freight, WarpStream, AutoMQ) are using this bucket type to implement stateless Kafka brokers.

These fast cloud storage (S3E1Z or EBS) are internally multi-way replicated, once the kafka log data is uploaded into these cloud storage, there is not much need to replicate further into other Kafka follower brokers.  Therefore in theory, you can reduce the replication factor of a Kafka topic to 1 to save the cost of 2 Kafka brokers.  In practice and for the initial version of this KIP, you might want to set up one or more Kafka follower brokers as a hot-standby or serve as an extra read replica for the downstream consumers in the same AZ (as the follower broker) to save across AZ traffic cost for downstream consumers.  But the replication from cloud storage to those extra Kafka follower brokers are free.  For a future enhancement of this KIP, we can support a mode which ends the replication when the data is uploaded to object storage and elect a new leader from any brokers when the old leader crashes. The new leader would need to sync up with the object storage for the last few seconds of data when it starts up as the new leader (and read the rest of the data async in the background).

Key Classes and Constructs

We are proposing to reuse and extend the data structures and constructs introduced in KIP-405 to support active log segments uploading. Most of the new classes in this KIP follow their counterparts from KIP-405 with the convention of adding WAL (WriteAheadLog) in the classes name or config parameters to indicate they are to support active write-ahead log segments.  We briefly mention a few classes in this section to introduce the new design, the details of the class design are in the next section Public Interfaces.

Configs

If a topic is configured with remote.wal.storage.enable=true (similar to remote.storage.enable in KIP-405), we will trigger the background uploading/downloading of the active log segment for the topic

RemoteWalStorageManager

Similar to RemoteStorageManager in KIP-405, we introduce an interface RemoteWalStorageManager to support uploading/downloading active WAL log segments.  The main methods of the class are copyLogSegmentData and fetchLogSegments.

RemoteLogManager.RLMWalTask

Similar to RemoteLogManager.RLMTask in KIP-405, RLMWalTask is a background task which is responsible to upload a section of active log segments on leader broker and download the active log segments on follower broker;

The section of active log segment being copied is org.apache.kafka.common.record.FileRecords which represents the batch records from the last offset it was uploaded to the current log end of the log segment.

RemoteWalCombinedLogSegmentMetadata

Similar to RemoteLogSegmentMetadata which is published to topic __remote_log_metadata topic when each log segment is uploaded onto object storage in KIP-405, we are publishing a RemoteWalCombinedLogSegmentMetadata when the log segment is uploaded onto object storage.  The reason it is a combined log segment is for performance reasons.  If we only upload one log segment for one topic partition, we would incur a huge cost of S3 transfer and the log segment will be too small.  So in the implementation we would combine the log segments from multiple topic partitions and do the upload.

RemoteLogManager.RLMWalCombinerTask

Since we are combining log segments from multiple partitions, RLMWalTask will actually submit the candidate batch records from a topic partition into a queue and RLMWalCombinerTask will read those records from the queue and combine them into Combined LogSegment and upload to object storage.

Producer Path to The Leader Broker

When a Kafka broker receives a ProduceRequest from producer, KafkaApis.handleProduceRequest is called, a MemoryRecords byte buffer will be allocated to hold the produce batch and be appended to the active log segments of the specified topic partition.

If the topic is configured with remote.wal.enable=true, the async task RemoteLogManager.RLMWalTask will be periodically woken up to copy the remaining records in the current active Log segment files to cloud storage and publish a RemoteLogSegmentMetadata to metadata topic __remote_log_metadata;

Producer Acknowledgement

Acks=1

When the producer produces using acks=1, the append of produce records onto the log segment file on the local disk will mark the receiving of produce records and a ProduceResponse reply will be sent out in this request/reply cycle.  The async copy of active log segments onto Object storage will not affect the message acknowledgement.

Since we added the object storage layer in the active log segment producer path, if a leader is lost, then the data that is lost is equivalent to the amount of data that is waiting to be written to object storage. This is similar to the original Kafka design where the amount of data lost is the data on the leader broker but not yet replicated to the followers. The amount of data lost  can be mitigated by uploading to object storage more frequently.

Acks=-1

When the producer produces using acks=-1, the append of produce records to local log segment will not mark the completion of the message receiving, instead a DelayedProduce purgatory action will be triggered (this is the current Kafka implementation). There are two options for when to mark the produce request as successful:

1. The leader broker will wait until all in-sync follower replicas have reached the given offset before the acknowledgement signal can be sent back.  The async upload of active log segments onto object storage and the async download of the active log segment will be indirectly involved in the acks=-1 reply cycle; they will add one more hop (the object storage layer) in the cycle.
2. The leader broker will wait only until the data is replicated to the object storage layer. Since the data is already in the object storage layer, it is multi-way replicated. There will be some additional time if the leader is lost for the follower to catch up from reading from object storage, but should at most be a few seconds of data. The result will be comparable performance to the current acks=-1 latency.

There is additional analysis of these two options in Performance Considerations below.

The role of object storage for replication factor

The WAL log segment files on fast cloud storage can serve as the remote replica for the given topic partition.  In theory, the replication factor of the topic can be 1 (i.e. only the leader broker is configured).  When the leader broker is crashed, a new leader broker can be configured and re-bootstrapped with the records coming from the fast cloud storage.  

The above bootstrapping process might take some time to finish, for the use cases which are sensitive to time delay the user might want to configure one or more follower brokers as a hot standby.  Alternatively as an enhancement of this KIP, we can optimize the bootstrap time of the new leader by having it sync with the object storage on the last few seconds of the data when it becomes the new leader (and sync the rest of the data slowly async in the background).  In a well operated Kafka pipeline, the consumer is usually only behind on the tail of the Kafka queue by a few seconds, once the new leaders get these few seconds of data from object storage the consumer can resume the consumption flow.

Replication and Download on follower broker

When the topic’s replication factor is > 1, the follower broker will send the normal FetchRequest to the lead broker to replicate data.  However the leader broker will just respond with empty MemoryRecords as the reply.  The follower broker will retrieve log segment metadata from __remote_log_metadata topic and ask RemoteLogManager to transfer the records from the corresponding WAL log segment files.  This data transfer will not incur the across AZ network transfer cost. The main purpose of FollowerFetchReqeust/FollowerFetchResponse is now just to update the offsets and high watermark between leader and follower.

Life Cycle of the active log segments on the cloud storage

• The active log segment are created and uploaded onto object storage on time interval from the leader broker;
• The active log segment on the fast object storage (e,g, S3E1Z) stays to be consumed by all follower brokers;
• At the same time, KIP-405 will also be working to upload older log segments onto slower long-term object storage (e.g. S3) when local.retention.ms passed for the log segments;  When the main log segments uploads finishes for KIP-405, the active log segments on fast object storage can be deleted by the log cleaner thread;
• Since we are combining multiple log segments from multiple topic partitions during the active log segment uploads, the log cleaner will actually need to wait until segments from all topic partitions are uploaded to long term object storage before deleting the combined log segment file. The retention of the log segment file is equal to the max of all topics’ local.retention.ms.

Performance Considerations

Reliance on PageCache

Readers will notice that we are not proposing a diskless implementation as in KIP-1150, instead we are using the cloud storage as an intermediate data hop before the data is replicated onto follower brokers.  Both the leader broker and follower broker still build a list of log segment files and consumer clients still fetch those log segments which are cached in Page Cache.  This design maintains the same performance as traditional Kafka by relying on its high performance throughput core tenet: page cache.

Same performance for acks=1 producer path and close performance for acks=-1

The background upload of active log segments to object storage (and subsequent download) is performed asynchronously behind the scenes, and can be tuned for larger batch sizes.  For acks=1 producers (which is still a majority of use cases in many companies), the data transfer to object storage has no impact on producer acknowledgement, the producer will be able to receive the reply as soon as the data is written into the page cache of the lead broker.

For acks=-1 producer path, there is a slight delay because data is transferred onto object storage first before it is downloaded onto the follower broker.  This delay can be reduced by tuning the batch size of the upload.  And for some use cases, the user might decide to treat data synced to object storage as acks=-1 completed since the data on object storage is multi-way replicated already.  For this we can add a mode to acknowledge back to the producer as soon as the data is uploaded onto object storage.  This would give us the same performance with current Kafka implementation.  The extra work we would need to do to complete this mode is to have the new leader (elected from the follower broker during leadership change) to pause and catch up with the object storage when it becomes the new leader.  ..

Performance tuning of uploading Active Log Segments onto Cloud Storage

Uploading log segments onto S3 is a tradeoff, faster shorter upload will reduce the latency but will pay for the cost of more uploads and smaller log segments.  To achieve a reasonable transfer cost and file size, we would need to combine the log segments from multiple topic partitions and increase the batch size of the upload (and thus increase the latency).

Public Interfaces

We are proposing to reuse and extend the data structures and constructs introduced in KIP-405 to support active log segments uploading.  Most of the new classes in this KIP follow their counterparts from KIP-405 with the convention of adding WAL (WriteAheadLog) in the classes name or config parameters to indicate they are to support active write-ahead log segments.  

org.apache.kafka.common.config.TopicConfig

• If a topic is configured with remote.wal.storage.enable=true, we will trigger the background uploading/downloading of the active log segment for the topic

org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig

• remote.log.wal.storage.manager.impl.prefix: config prefix for RemoteWalStorageManager configuration parameters
• remote.log.wal.storage.system.enable: boolean parameter to enable the remote wal storage systems
• remote.log.wal.storage.manager.class.name: classname for RemoteWalStorageManager
• remote.log.wal.storage.manager.class.path: classpath to load RemoteWalStorageManager
• remote.wal.log.manager.thread.pool.size: thread pool size for RLMWalTask treads
• remote.wal.log.manager.combiner.thread.pool.size: thread pool size for RLMWalCombiner task threads
• Remote.wal.log.manager.combiner.queue.size: queue size for RemoteWalCombiner task

org.apache.kafka.server.log.remote.storage.RemoteWalCombinedLogSegmentMetadata

• Similar to RemoteLogSegmentMetadata which is published to topic __remote_log_metadata topic when each log segment is uploaded onto object storage in KIP-405, we are publishing a RemoteWalCombinedLogSegmentMetadata when the log segment is uploaded onto object storage.  The reason it is a combined log segment is for performance reasons.  If we only upload one log segment for one topic partition, we would incur a huge cost of S3 transfer and the log segment will be too small.  So in the implementation we would combine the log segments from multiple topic partitions and do the upload.

public class RemoteWalCombinedLogSegmentMetadata extends RemoteLogMetadata {
    /**
     * Universally unique remote log segment id.
     */
    private final Uuid combinedSegmentId;

    private final List<RemoteLogSegmentMetadataEntry> remoteLogSegmentMetadataEntries;

    /**
     * Custom metadata.
     */
    private final Optional<CustomMetadata> customMetadata;

    /**
     * It indicates the state in which the action is executed on this segment.
     */
    private final RemoteLogSegmentState state;

    /**
     * Metadata entry for each log segment
     */
    public static class RemoteLogSegmentMetadataEntry {
        RemoteLogSegmentId remoteLogSegmentId;
        long startFilePosition; // start position in the file for this segment
        long endFilePosition;   // end position in the file for this segment

    }
}

org.apache.kafka.server.log.remote.storage.RemoteWalCombinedLogSegmentMetadataUpdate

Similar to class RemoteLogSegmentMetadataUpdate class in KIP-405, this MetadataUpdate class captures the start and end file position for each log segment in the combined log segments and is constructed when RemoteWalStorageManager#copyLogSegments is called.

public class RemoteWalCombinedLogSegmentMetadataUpdate extends RemoteLogMetadata {

    /**
     * Universally unique remote log segment id.
     */
    private final Uuid combinedSegmentId;

    private final List<RemoteLogSegmentMetadataEntry> remoteLogSegmentMetadataEntries;

    /**
     * Custom metadata.
     */
    private final Optional<CustomMetadata> customMetadata;

    /**
     * It indicates the state in which the action is executed on this segment.
     */
    private final RemoteLogSegmentState state;

org.apache.kafka.server.log.remote.storage.RemoteWalStorageManager

Similar to RemoteStorageManager in KIP-405, we introduce an interface RemoteWalStorageManager to support uploading/downloading active WAL log segments.  The main methods of the class are copyLogSegmentData and fetchLogSegments.

package org.apache.kafka.server.log.remote.storage;

   /**
     * Copies the given List of {@link Records} provided for the given {@code remoteWalCombinedLogSegmentMetadata}.
     * {@code remoteWalCombinedLogSegmentMetadata} contains the metadata for a list of segments to be copied.  The list
     * of {@link Records} is the list of {@link org.apache.kafka.common.record.FileRecords} data objects be copied for
     * each log segment.
     * <p>
     * This operation is expected to be idempotent. If a copy operation is retried and there is existing content already written,
     * it should be overwritten, and do not throw {@link RemoteStorageException}
     *
     * @param remoteWalCombinedLogSegmentMetadata metadata about the combined remote wal log segment.
     * @param recordsList           a list of {@link Records} to be copied
     * @return RemoteWalCombinedLogSegmentMetadataUpdate which contains the metadata for the copied segments
     * @throws RemoteStorageException if there are any errors in storing the data of the segment.
     */
public interface RemoteWalStorageManager extends Configurable, Closeable {

   RemoteWalCombinedLogSegmentMetadataUpdate copyLogSegmentData(
RemoteWalCombinedLogSegmentMetadata remoteWalCombinedLogSegmentMetadata,                                                               List<Records> recordsList)
            throws RemoteStorageException;

   /**
     * Returns the remote log segment data file/object as InputStream for the given {@link RemoteWalLogSegmentMetadata}
     * starting from the given startPosition. The stream will end at the smaller of endPosition and the end of the
     * remote log segment data file/object.
     *
     * @param remoteLogSegmentMetadata metadata about the remote log segment.
     * @param startPosition            start position of log segment to be read, inclusive.
     * @param endPosition              end position of log segment to be read, inclusive.
     * @return input stream of the requested log segment data.
     * @throws RemoteStorageException          if there are any errors while fetching the desired segment.
     * @throws RemoteResourceNotFoundException the requested log segment is not found in the remote storage.
     */
    InputStream fetchLogSegment(RemoteWalLogSegmentMetadata remoteLogSegmentMetadata,int startPosition,int endPosition) throws RemoteStorageException;
}

kafka.log.remote.RemoteLogManager.RLMWalTask

Similar to RemoteLogManager.RLMTask in KIP-405, RLMWalTask is a background task which is responsible to upload a section of active log segments on leader broker and download the active log segments on follower broker;

The section of active log segment being copied is from the last offset it was uploaded to the current log end of the log segment.  The section of batch records is represented by Records (and FileRecords are the actual implementation) 

Since we are combining log segments from multiple topic partitions, RLMWalTask (one for each topic partition) is submitting the records for the given topic partition through in-memory queue remoteSegmentQ, the next task RLMWalCombineTask is going to combine the records from multiple topic partitions and submit them as one CombinedLogSegment to object storage.

class RLMWalTask extends CancellableRunnable {

        private final TopicIdPartition topicIdPartition;
        private final Logger logger;
        private final BlockingQueue<MetaQueueItem> remoteSegmentQ;

        public RLMWalTask(TopicIdPartition topicIdPartition,
                          BlockingQueue<MetaQueueItem> remoteSegmentQ) {


   static class MetaQueueItem {
        private final RemoteWalLogSegmentMetadata metadata;
        private final Records records;

kafka.log.remote.RemoteLogManager.RLMWalCombineTask

Since we are combining log segments from multiple topic partitions, RLMWalTask (one for each topic partition) is submitting the records for the given topic partition through in-memory queue remoteSegmentQ, the next task RLMWalCombineTask is going to combine the records from multiple topic partitions and submit them as one CombinedLogSegment to object storage.

class RLMWalCombineTask extends CancellableRunnable {

        private final int combinerSize = 1;
        private final BlockingQueue<MetaQueueItem> remoteMetadataQ;
        private final int customMetadataSizeLimit;
        private final Logger logger;

        public RLMWalCombineTask(BlockingQueue<MetaQueueItem> remoteMetadataQ, int customMetadataSizeLimit) {

When RLMWalCombineTask is uploading records to the object storage, it is also publishing multiple events to __remote_log_metadata topic.  It is submitting one RemoteWalCombinedLogSegMetadata event followed by a RemoteWalLogSegmentMetadata event for each topic partition.  The reason multiple events are submitted is because each ConsumerTask from KIP-405 is only listening for events for its own topic partition (it listens for a specific Metadata Partition in __remote_log_metadata topic) and therefore a separate metadata event for each topic partition needs to be published.

And follow the same pattern from RLMTask in KIP-405, a RemoteLogSegmentMetatadata event is first published to metadata topic to mark COPY_SEGMENT_STARTED state and then after object uploads completes a RemoteLogSegmentMetadataUpdate event is published with COPY_SEGMENT_FINISHED state which also contains the extra information (e.g. custom metadata or file start/end position) retrieved from the object storage upload.

Here is a code snippet for the upload events during records upload:

           for (MetaQueueItem item : metas) {
                RemoteWalLogSegmentMetadata copySegmentStartedRlsm = item.getMetadata().createWithUpdate(combinedSegmentId);
                
// For each topic partition, publish out the Metadata start event
remoteLogMetadataManager.addRemoteWalLogSegmentMetadata(copySegmentStartedRlsm).get();

                remoteLogSegmentMetadataEntries.add(new RemoteWalCombinedLogSegmentMetadata.RemoteLogSegmentMetadataEntry(
                        copySegmentStartedRlsm.remoteLogSegmentId(), -1, -1));
                recordsList.add(item.getRecords());
            }

            RemoteWalCombinedLogSegmentMetadata startMetadata = new RemoteWalCombinedLogSegmentMetadata(combinedSegmentId,
                    remoteLogSegmentMetadataEntries, brokerId, time.milliseconds());
            
// Publish the Combined Metadata Start Event
remoteLogMetadataManager.addRemoteWalCombinedLogSegmentMetadata(startMetadata).get();

            // Upload the objects to object storage
            RemoteWalCombinedLogSegmentMetadataUpdate metadataUpdate = remoteWalLogStorageManager.copyLogSegmentData(
                    startMetadata, recordsList);
            }
                       
// Publish out the Combined Metadata finished event
remoteLogMetadataManager.updateRemoteWalCombinedLogSegmentMetadata(metadataUpdate).get();

            for (RemoteWalCombinedLogSegmentMetadataUpdate.RemoteLogSegmentMetadataEntry entry : metadataUpdate.remoteLogSegmentMetadataEntries()) {
                RemoteWalLogSegmentMetadataUpdate copySegmentFinishedRlsm = new RemoteWalLogSegmentMetadataUpdate(
                        entry.remoteLogSegmentIdSegmentId(), time.milliseconds(), metadataUpdate.customMetadata(),
                        RemoteLogSegmentState.COPY_SEGMENT_FINISHED, brokerId,
                        combinedSegmentId, combinedCoordinatorTopicPartition, entry.startFilePosition(), entry.endFilePosition());
                
//Publish the Metadata finished event for each topic partition
remoteLogMetadataManager.updateRemoteWalLogSegmentMetadata(copySegmentFinishedRlsm).get();
            }

Compatibility, Deprecation, and Migration Plan

• We are introducing new classes and new configs to support replicating active log segments onto object storage, no existing features are changed.

Test Plan

We implemented a LocalWalStorageManager to test the E2E flow using a shared local filesystem.

Rejected Alternatives

This KIP was being developed independently without much knowledge of Aiven’s proposal of KIP-1150 (Diskless Kafka Topic).  Since KIP-1150 was published earlier, we would draw some comparisons between our proposal and KIP-1150:

• KIP-1150 is a major overhaul of Kafka storage system and KIP-1150 is just the first of the 7-8 KIPs proposed by Aiven;

• KIP-1150 completely removes all local storage, thus removes the core tenant of Kafka storage system: page cache, the main reason behind a fast Kafka;

• KIP-1150 has a long acknowledgement cycle on producer message publishing, the message cannot be acknowledged back to the producer until it is uploaded onto object storage.  This performance degradation becomes more severe on acks=1 producer path where the data acknowledgement was used to be fast (as soon as the record is written to lead broker’s page cache).  Although Kafka has promoted users to use acks=-1 as default for many years, acks=-1’s performance is significantly slower than acks=1 due to multiple acks signals needing to be collected from all follower brokers.  For this reason, many company’s logging pipelines are still using acks=1 since occasional data loss are usually tolerable for logging use cases;

• KIP-1150 understands the diskless topic cannot replace all use cases since the upload onto object storage is going to be slower than local disk performance;  In its blog, it targets its main usage for applications running in 200-400 ms latency

In contrast, our implementation is a gradual evolution of KIP-405 tiered storage.  We are reusing many data structures and classes from KIP-405 (e.g. RemoteLogManager, RLMTask, RemoteLogSegmentMetadata, __remote_log_metadata topic).  The amount of code change and new concepts introduced are more manageable in our proposal.

We understand the importance of page cache in traditional Kafka design and we still based our implementation of using local log segment files and its page cache, the object storage for active log segments are added to remove the across-AZ transfer cost but the local files stay to provide a caching layer for fast data retrieval to clients.  For this reason the performance of our system under acks=1 can still maintain the single-digit ms latency;

Our proposal is in a way also trying to reduce the state size of Kafka broker by uploading more recent log segments onto object storage but we are implementing it in a more conservative way.

Note that we mentioned KIP-1150 in the Rejected Alternative section but we don't view this KIP and KIP-1150 are all-or-nothing alternatives, they are optimized for different use cases (this KIP proposal are optimizing for low-latency acks=1 use case while KIP-1150 is optimizing for a longer latency but diskless implementation), we believe some of the ideas or implementations can be learned or adopted by both KIPs.

Appendix: Cost Estimate for a typical workload

Workload example

For one of our sample 3-way replicated kafka cluster of 90 i3en.2xlarge nodes, for each node we are getting 30MB/s inbound traffic, which is about 112GB per hour of data needs to be stored;  And we are targeting to upload every 10ms which is about 100 IOPS.

For the proposed new architecture, we are targeting to have one writer (leader kafka broker) writes to cloud storage, one reader (one follower kafka broker) to read from the cloud storage and plan to store 1 hour of data in the cloud storage (the storage cost is usually small so storing 2 or 3 hours of data wouldn’t add much cost).

Comparison Table

See the following table for cost comparison.  The charge is per month per kafka broker

The result is ~43% reduction in cost.

The details of the calculation can be provided per request.