Page History

...

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 and metadata topic __remote_wal_log_metadata

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 to topic __remote_wal_log_metadata 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.

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 from multiple topic partitions and do the uploadof 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.

RemoteLogManager.RLMWalCombinerTask

...

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_wal_log_metadata;

Producer Acknowledgement

...

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.

...

• 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 onto metadata topic: __remote_wal_log_metadata 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.

...

When RLMWalCombineTask is uploading records to the object storage, it is also publishing multiple events to __remote_wal_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_wal_log_metadata topic) and therefore a separate metadata event for each topic partition needs to be published.

...

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.

...

This KIP is being developed as an evolution of KIP-405 and it is using many of the constructs introduced in KIP (e.g. RemoteLogManager, RLMTask, __remote_log_ metadata topic, ProducerManager, ConsumerManager, MetadataStore), however there are some differences between this KIP and KIP-405:

• While KIP-405 is uploading closed log segments to a slow object storage this KIP is uploading active log segment to a fast object storage.
• In KIP-405 we would need to upload various index files and producer snapshots alongside data logs to object storage, in this proposed KIP we only need to upload the data logs to object storage.  In KIP-405, all local logs on all brokers will be removed after local.retention.ms passed we would need to keep a consistent log snapshot (including all index lookup files) in object storage for future lookups and downloads;  In this proposed KIP the object storage is mostly used as an intermediate data hop between the leader and the follower, the follower will build up all the indexing files and producer snapshot files while it is appending batch records from object storage to local log by calling UnifiedLog.appendAsFollower (in the same way as today's follower when the follower builds indexes while calling UnifiedLog.appendAsFollower for the records coming from FetchResponse)
• Although both KIP-405 and this proposed KIP is using __remote_log_metadata topic to keep the metadata for each data log segment, the upload frequency is a lot more frequent in this proposed KIP.  In KIP-405, we are uploading a closed log segment (usually in the size of 1GB) every 5-10 minutes while in this proposed KIP we are uploading a smaller data segment and much more frequent (e.g. every 10ms).  With the cost optimization of combing data from multiple topic partitions, each uploaded combined active log segments are in the size of 100KB-300KB.  But the data retention of active log segment's metadata is much shorter so the overall data size of the topic is not very big, the active log segment only need to live on object storage before the data is uploaded to slow object storage through KIP-405 (happens about 5-10 minutes after the message was appended to the local log when the active log segment is closed due to log rotation).
  • We are planning on moving the metadata for active log segment into its own metadata topic __remote_wal_log_metadata to set a shorter retention on the metadata topic
  • If the topic's local.retention.ms is set to 1 hour, the local data logs for the topic will be removed after 1 hour since KIP-405's design assumes that the data will already be uploaded onto object storage by 1 hour mark.  For this reason we can set the retention for active log segment metadata topic to be 1 hour, by the 1 hour mark the active segment's data log on fast object storage will be removed and there is no need to keep its metadata in the metadata topic anymore.

...