...

Authors: Henry Cai, Thomas Thornton

Status

Current state: Under Discussion Withdrawn

Discussion thread: here 

JIRA: KAFKA-19225 

...

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

...

Cloud Storage for Active Log Segment

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. Note: for most cloud storage providers there will still be some costs (e.g., PUT/GET cost per request, transfer surcharge for performant object storage), but these are typically a fraction of the original 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 (example), 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 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 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.

RemoteLogManager.RLMWalCombinerTask

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

Since we are combining 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.

...

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

...

The proposal of using cloud storage for active log segment can also be implemented on traditional object storage (e.g. S3), you can implement both RemoteStorageManager and RemoteWalStorageManager on the same class.  Although S3 has higher latency than S3E1Z, S3 provides across-AZ durability which might be required for some use cases.  For an implementation using S3 as active log segment storage, you would need to configure the system to upload the segment in lower frequency with higher batch size.

AZ Availability

In the proposed design, we propose using 1 leader a typical setup would involve a leader broker in AZ1, 1 a follower broker in AZ2 and 1 S3E1Z , and an S3 Express One Zone (S3E1Z) bucket in AZ1.  The current design co-locate Co-locating the leader and the S3E1Z bucket in the same AZ to have a shorter minimizes upload latency.  This design will cover 2 AZs but with 5 replicas (S3E1Z is also This configuration provides robust data redundancy across two AZs with what is effectively five replicas of the data (given that S3E1Z is internally multi-way replicated in the same AZ), although we didn't cover AZ3 we feel this setup has a very strong data redundancy (with 5 replicas).  If the user worries about total AZ failure in within its AZ). For users concerned about a simultaneous failure of both AZ1 and AZ2, they can set up another follower in AZ3 to have complete 3 AZ coverage, note even with 3 brokers they can still get the the across-AZ traffic free.  The alternative design is putting the S3E1Z bucket in AZ3 to cover 3 AZs but this would introduce longer latency during message replication.

Data Durability

In acks=all flow, we maintain the same data durability as today by requiring the data gets to the follower before we can acknowledge back to the producer;

In acks=1 flow, we have similar data durability as today by waiting to update the high watermark until the data arrives on the follower broker and the consumer won't be able to read the new message until the high watermark moves passing the new record.  There is an extra latency for the message flows from object storage to the follower broker (in the order of 10ms for bigger batches), this extra latency would mean the consumer needs to wait longer and it also mean the producer would have a longer window for data loss.  In today's code, we can lose about 10ms' data for the data on the way from the leader to the follower when the leader crashes, in this proposed KIP we can lose maybe 20ms' data for the data on the way from the leader to the object storage and then to the follower broker when the leader crashes.  Although the data loss window doubles but it only happens on leader broker crash which is a rare event, most of the acks=1 data flow can tolerate this occasional data loss.

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
• remote.wal.log.manager.combiner.task.interval.ms: controls how frequent the combiner task runs
• remote.wal.log.manager.combiner.task.upload.bytes: control how big the upload packet is going to be;

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

an additional follower can be placed in a third AZ to achieve complete three-AZ coverage, still benefiting from the elimination of cross-AZ replication costs.

Leader Failover and Recovery in an AZ Outage

A critical scenario to consider is the failure of an entire Availability Zone, which would take down both the leader broker and its co-located S3E1Z bucket. In this rare event, the cluster remains available by electing a new leader in a different AZ. Once a new leader (e.g., in AZ2) is elected, replica reconciliation takes place. This is similar to standard Kafka. If followers have more events than the high watermark, they will need to truncate them. Below is an example to illustrate this:

Suppose there is a remote WAL enabled topic with replication factor 3. The current leader of a partition is B1. B1 has received data up to offset 10, written data to S3E1Z up to offset 10, and published metadata up to offset 10. One follower, B2, has replicated up to offset 5, and follower B3, up to offset 7. Suppose B1 dies (e.g., AZ goes down), and B2 is elected as the new leader. B2 overwrites the LEO/high-watermark based on its own LEO, which is 5. B2 will tell its followers to reset their LEO to 5. So B3, which had an LEO of 7, will discard 2 messages and reset 5 to match B2, the new leader. In addition, B2 will also publish metadata DELETE events for any metadata entries it receives that are after its LEO (e.g., for messages 6-10), since they are no longer valid.

In the case that B3 was elected, with the higher end offset of offset 7. It will perform replica reconciliation and discard its data that is greater than the high watermark of 5. This prevents any sacrifices in availability, and is equivalent to standard Kafka.

Data Durability

In acks=all flow, we maintain the same data durability as today by requiring the data gets to the follower before we can acknowledge back to the producer;

In acks=1 flow, we have similar data durability as today by waiting to update the high watermark until the data arrives on the follower broker and the consumer won't be able to read the new message until the high watermark moves passing the new record.  There is an extra latency for the message flows from object storage to the follower broker (in the order of 10ms for bigger batches), this extra latency would mean the consumer needs to wait longer and it also mean the producer would have a longer window for data loss.  In today's code, we can lose about 10ms' data for the data on the way from the leader to the follower when the leader crashes, in this proposed KIP we can lose maybe 20ms' data for the data on the way from the leader to the object storage and then to the follower broker when the leader crashes.  Although the data loss window doubles but it only happens on leader broker crash which is a rare event, most of the acks=1 data flow can tolerate this occasional data loss.

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
• remote.wal.log.manager.combiner.task.interval.ms: controls how frequent the combiner task runs
• remote.wal.log.manager.combiner.task.upload.bytes: control how big the upload packet is going to be;

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

public class RemoteWalCombinedLogSegmentMetadata extends RemoteLogMetadata {
    /**
     * Universally unique remote log segment id

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

    private final List<RemoteLogSegmentMetadataEntry> remoteLogSegmentMetadataEntries;

    /**
     * Custom metadata.
     * Custom metadata can be used to encode the object path in object storage and/or the bucket name in the object storage.  This is especially useful if you use multiple buckets in the object storage.
     */
    private final Optional<CustomMetadata>Uuid customMetadatacombinedSegmentId;

    /**
private final List<RemoteLogSegmentMetadataEntry> remoteLogSegmentMetadataEntries;

  * It indicates/**
 the state in which the* action is executed on this segmentCustom metadata.
     */
 Custom metadata  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;   // endcan be used to encode the object path in object storage and/or the bucket name in the object storage.  This is especially useful if you use multiple buckets in the object storage.
     */
    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

...

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

Future enhancement

Cloud native elasticity

This KIP is extending KIP-405: Tiered Storage.  KIP-405 moved all closed log segments to object storage and this KIP moves the active log segment to the object storage as well.  With all data now living on cloud storage, this opens the door for cloud-native-elasticity.  The consumers now can read directly from cloud storage (without connecting to the broker), and we can utilize cloud vendor's fan-out support to easily add a lot more consumer reading volume in a short amount of time (elastic consumer fan-out support).  By moving in this direction majority of the traffic (consumer traffic probably comprises 2/3 of the overall traffic) will be happening outside broker, there are much less resources (network bandwidth / memory) we need to allocate to the broker.  The cluster will become much easier to scale up as well.

Further reduction of latency broker resources

For simplicity reason, the current proposal is still recommending set up a follower broker as a hot standby in case the existing broker crashes and you need to fast switch to a new leader.  This adds more resource requirement on the broker and also add one more data hop in the latency cycle since the data needs to travel to the follower broker before acknowledging back to the producer (for acks=all).  But with a bit more work, we can remove the need for this hot standby follower.  With the conjunction of KIP-405, all the active and closed log segment are now living on the cloud storage, when the old broker dies any new broker can become the new leader by syncing up the data from object storage.  KIP-405 already supports downloading data from closed log segment from cloud, for the active log segment we need to download about 5-10 minutes more of data.  This download volume can be even smaller because in a well maintained data pipeline the consumer is usually only behind by a few seconds, this would mean we only need to download the last few seconds of data from the cloud to the new leader for the consumer fetch to resume.  The new leader can continue downloading the rest of data asynchronous on the background.

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

...

Baseline

...

S3E1Z

...

Storage/Instance Cost

...

$866

...

$776

...

Transfer Cost

...

$1612

...

$648

...

Total Cost

...

$2478

...

$1424

...

Latency

...

Single digit ms

...

Single digit ms

...

Across AZ

...

Yes

...

Yes

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

Future enhancement

Cloud native elasticity

This KIP is extending KIP-405: Tiered Storage.  KIP-405 moved all closed log segments to object storage and this KIP moves the active log segment to the object storage as well.  With all data now living on cloud storage, this opens the door for cloud-native-elasticity.  The consumers now can read directly from cloud storage (without connecting to the broker), and we can utilize cloud vendor's fan-out support to easily add a lot more consumer reading volume in a short amount of time (elastic consumer fan-out support).  By moving in this direction majority of the traffic (consumer traffic probably comprises 2/3 of the overall traffic) will be happening outside broker, there are much less resources (network bandwidth / memory) we need to allocate to the broker.  The cluster will become much easier to scale up as well.

Further reduction of latency broker resources

For simplicity reason, the current proposal is still recommending set up a follower broker as a hot standby in case the existing broker crashes and you need to fast switch to a new leader.  This adds more resource requirement on the broker and also add one more data hop in the latency cycle since the data needs to travel to the follower broker before acknowledging back to the producer (for acks=all).  But with a bit more work, we can remove the need for this hot standby follower.  With the conjunction of KIP-405, all the active and closed log segment are now living on the cloud storage, when the old broker dies any new broker can become the new leader by syncing up the data from object storage.  KIP-405 already supports downloading data from closed log segment from cloud, for the active log segment we need to download about 5-10 minutes more of data.  This download volume can be even smaller because in a well maintained data pipeline the consumer is usually only behind by a few seconds, this would mean we only need to download the last few seconds of data from the cloud to the new leader for the consumer fetch to resume.  The new leader can continue downloading the rest of data asynchronous on the background.

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.

Appendix: Performance Data

Configuration

Topic perf_cloud_native: 2-way replicated, one leader (AZ: use1-az4) writes to cloud storage, one reader (AZ: use1-az6) reads from cloud storage (uses KIP-1176 code)

Topic perf_local_disk: 2-way replicated (for fair comparison), standard replication path, leader (AZ: use1-az4) writes to local disk, follower (AZ: use1-az6) reads via cross AZ-transfer

Each topic has 1 partition.

Results

End to End Latency

A custom script was run to determine end-to-end-latency, which is defined here as the difference between system time when a record is consumed and system time when a record is created in memory on the producer side. The other producer/consumer metrics are all standard.

Workload

100 records, 1.6 KB avg record size, max batch size 32kb, acks=1, consumer client.rack=use1-az6 (read from follower)

Metrics

Kafka Producer/Consumer Perf Test Scripts

Workload

Used kafka/bin/kafka-producer-perf-test.sh and kafka/bin/kafka-consumer-perf-test.sh.

Producer: --num-records 100000 --record-size 2000 --throughput -1 --producer-props acks=1 retries=2 linger.ms=20 delivery.timeout.ms=300000 request.timeout.ms=30000

Consumer: --messages 100000 --timeout 240000 --consumer.config consumer.properties
The consumer.properties file contained client.rack=use1-az6 (the rack of the follower broker, so we consumed from the follower).

Metrics

Analysis

Overall we see comparable performance between a standard topic and KIP-1176 for both producer & consumer metrics, in particular throughput and latency. The producer path is near identical performance, i.e., an ack to the producer is returned in the same time. Consumer performance is slightly slower as data must travel an additional hop to/from S3E1Z. There are additional tunings we are aware of that could further boost performance (e.g., tuning the interval that the leader/follower remote WAL tasks run).

For latency, there are two latency metrics. E2E latency measures when the consumer sees the data. Producer's request-latency measures when the producer can move on after a message is sent. We can see the request-latency number is very small (<10ms) on both cloud-native and local-disk cases, this is where acks=1 setting excels at and this is also the niche of our KIP

The result is ~43% reduction in cost.

The details of the calculation can be provided per request.