...

Authors: Henry Cai, Thomas Thornton

Status

Current state: Under Discussion Withdrawn

Discussion thread: here 

JIRA: KAFKA-19225 

...

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

...

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 within its 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 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. For users concerned about a simultaneous failure of both AZ1 and AZ2, 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;

...

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)

...

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

...

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

...