Page History

...

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. In the case of cloud providers where cross-AZ data transfer is free (e.g., Azure), there won't be any cross AZ savings, but there would still be broker savings. Since this approach only needs two replicas, the typical three replicas can be reduced.

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 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). Additionally, the follower uses the leader epoch to filter out any messages sent by the old leader. Only metadata from the latest leader is processed.

Note although this proposal is promoting to use fast cloud storage (S3E1Z or EBS) to store active log segment, the interfaces proposed in this proposal can also be easily implemented on a traditional cloud storage (e.g. S3) with a high latency.

...

We chose to create a separate metadata topic __remote_wal_log_metadata because the uploading of active log segment is much more frequent comparing to closed log segment.  We envision we will upload a log segment in the size of 300-500KB every 10-20ms, due to the frequency of the metadata publishing and the ensuing size of the topic, it is recommended to set the retention of the topic to small value.  The minimum value of the topic retention needs to be bigger than all tiered storage topic's local.retention.ms because when the time exceeds tiered storage topic local.retention.ms, the closed log segments of the topic will be removed from local storage since they are assumed to uploaded to slow tiered storage already.  Once they are on tiered storage object storage there is no need to keep active log segments anymore in the fast object storage.

Reads are done on a byte range of a combined object (metadata message contains the byte range), and read costs are only billed only on that byte range. Byte range requests are supported across the major cloud storage providers.

RemoteLogManager.RLMWalCombinerTask

...

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_wal_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.

The current API does support sending metadata with empty records object. However, the interface may be expanded to have a cleaner way of supporting this (e.g., only return metadata & directly update replicated offset).Image Modified

Life Cycle of the active log segments on the cloud storage

...