You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 10 Next »

Status

Current stateUnder Discussion"

Discussion thread: here

JIRA: here

Released: <Kafka Version>

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

Motivation

The most expensive part of a Kafka cluster is probably its storage system. At LinkedIn we use RAID-10 for storage and set Kafka’s replication factor = 2. This setup requires 4X space to store data and tolerates up to 1 broker failure. We are at risk of data loss with just 1 broker failure, which is not acceptable for e.g. financial data. On the other hand, it is prohibitively expensive to set replication factor = 3 with RAID-10 because it will increase our existing hardware cost and operational cost by 50%.

The solution is to use JBOD and set replication factor = 3 or higher. It is based on the idea that Kafka already has replication across brokers and it is unnecessary to use RAID-10 for replication. Let’s say we set the replication factor = 4 with JBOD. This setup requires 4X space to store data and tolerate up to 3 broker failures in order not to lose any data. In comparison to our existing setup, this allows us to obtain 3X broker failure tolerance without increasing storage hardware cost.

We have evaluated the possibility of using other RAID setup in LinkedIn. But none of them addresses our problem as JBOD does. RAID-0 stops working entirely with just one disk failure. RAID-5 or RAID-6 has sizable performance loss as compared to RAID-0 and probably JBOD as well, due to their use of block-level striping with distributed parity.

Unfortunately, JBOD is not recommended for Kafka because some important features are missing. For example, Kafka lacks good support for tools as well as load balancing across disks when multiple disks are used. Here is a list of problems that need to be addressed for JBOD to be useful:

1) Broker will shutdown if any disk fails. This means a single disk failure can bring down the entire broker. Instead, broker should still serve those replicas on the good disks as long as there is good disk available.

2) Kafka doesn’t provide the necessary tools for users to manage JBOD. For example, Kafka doesn’t provide script to re-assign replicas between disks of the same broker. These tools are needed before we can use JBOD with Kafka.

3) JBOD doesn’t by itself balance load across disks as RAID-10 does. This will be a new problem for us to solve in order for JBOD setup to work well. We should have a better solution than round-robin which we are using to select disk to place a new replica. And we should probably figure out how to re-assign replicas across disks of the same broker if we notice load imbalance across disks of a broker.

For ease of discussion, we have separated the design of JBOD support into two different KIPs. This KIP address the first problem. See KIP - Support replicas movement between log directories for our proposal to address the second problem. 

Since Kafka configuration and implementation does not expose "disk", we will use log directory and disk interchangably in the rest of the KIP.

Goals

 

The goal of this KIP is to allow broker to serve replicas on good log directories even if some log directories have failed. This addresses the first problem raised in the motivation section. See KIP - Support replicas movement between log directories to read our proposal of how to address the second problem.

Proposed change

How to handle log directory failure

Problem statement:

Currently LeaderAndIsrRequest is used for two purpose: 1) create a new replica on a broker and 2) switch a replica between leader/follower of the partition. If a broker starts with some replicas unavailable because they are on a bad log directory, it will re-create those replicas on a good log directory when it receives LeaderAndIsrRequest from the controller. This is wrong. To avoid this, controller needs to know whether the replica has been created on the broker and explicitly specify whether broker should create replica in the LeaderAndIsrRequest.

Further, currently a replica is offline if and only if the broker hosting the replica is offline. This logic is no longer applicable with JBOD. Broker needs to explicitly tell controller the list of online replicas on its good log directories so that controller can elect leader from only online replicas. And admin tools should be developed to allow user to query which replicas are online or offline.

Solution:

The idea is for controller to keep track of weather a replica has been successfully created by getting this information from LeaderAndIsrResponse and persisting this information in the zookeeper. Then controller can explicitly tell broker whether to create replica or not by specifying a newly-added boolean field in the LeaderAndIsrRequest. Each broker can tell controller whether a replica is created and online by specifying the error_code per partition in the LeaderAndIsrResponse, which in turn allows broker to derive the offline replicas per partition and elect leader appropriately.

Here are a few clarification to make our solution easier to understand:
- Broker assumes a log directory to be good after it starts until there is IOException when broker attempts to access the log directory.
- Broker will be offline if all log directories are bad.

In the following we describe how our solution works under eight different scenarios. Some existing steps (e.g. kafka-topics.sh creates znode) are omitted for simplicity.

1. Topic gets created

- The controller creates znode /broker/topics/[topic]/partitions/[partitionId]/controller_managed_state with json-formatted data {"version" : 1, "created" : []} for this partition.
- The controller sends LeaderAndIsrRequest with create=True to the leader and followers.
- The leader and followers create replicas locally and sends LeaderAndIsrReponse to the controller with error=None.
- After receiving LeaderAndIsrResponse from leader and followers of this partition, the controller adds broker to the list "created" of this partition if error=None in LeaderAndIsrResponse. Otherwise, the replica is considered offline.

2. A log directory stops working on a broker during runtime

- The controller watches the path /log_dir_event_notification for new znode.
- The broker detects offline log directories during runtime.
- The broker notifies the controller by creating a sequential znode under path /log_dir_event_notification with data of the format {"version" : 1, "broker" : brokerId, "event" : LogDirFailure}.
- The controller reads the znode to get the brokerId and finds that the event type is LogDirFailure.
- The 
controller deletes the notification znode
- The controller sends LeaderAndIsrRequest to that broker to query the state of all topic partitions on the broker. The LeaderAndIsrResponse from this broker should tell the controller the list of partitions that are on the bad log directories.
- The controller updates the information of offline replicas in memory and trigger leader election as appropriate.
- The controller propagates the information of offline replicas to brokers by sending UpdateMetadataRequest.

3. Broker bootstraps with bad log directories

- Broker collects list of replicas found on the good log directories. If there is no good log directory the broker will exit.
- The controller sends LeaderAndIsrRequest for all partitions that should exist on this broker. If a replica is not in the created list of a partition, its create=True in the LeaderAndIsrRequest. Otherwise, its create=False.
- The broker will specify error=KafkaStorageException for those partitions that are in the LeaderAndIsrRequest with create=False but not found on any good log directory.
- The controller considers a replica on that broker to be offline if its error!=None in the LeaderAndIsrResponse.
- The controller updates the information of offline replicas in memory and triggers leader election as appropriate.
- The controller propagates the information of offline replicas to brokers by sending UpdateMetadataRequest.

4. The disk (or log directory) gets fixed

- The controller watches the path /log_dir_event_notification for new znode.
- User can either replace a bad disk with good disk, or remove the bad log directory from broker config.
- User restarts broker. Broker can read all log directories specified in its config.

- The controller sends LeaderAndIsrRequest with 
create=False to this broker because this replica has been created according to zookeeper.
- Broker will create replica on a good disk because it can access all log directories specified in the config. 


5. User queries replicas state
 
- Controller propagates the information of offline replicas to brokers by sending UpdateMetadataRequest. MetadataResponse will include offline replicas per partition.
- Kafka client can send MetadataRequest to any broker to query offline replicas for the specified partitions.
- kafka-topics script will display offline replicas when describing a topic partition. The offline replicas is the union of offline replicas on live brokers and replicas on dead brokers. kafka-topics script obtains offline replicas by sending MetadataRequest to any broker.
6. Controller failover
- Controller sends LeaderAndIsrRequest to all brokers. It specifies create=False for those replicas in the created list in the zookeeper. Controller specifies create=True for other replicas.
- Broker responds with LeaderAndIsrResponse and specifies error for offline replicas in the response
- Controller derives offline replicas from LeaderAndIsrResponse and make leader election among live replicas.

7. Partition reassignment
The procedure for partition reassignment will be almost the same as the current procedure. There are three changes:

- Controller will use both broker liveness and LeaderAndIsrResponse to determine a replica is online or not. StopReplicaRequest will fail if replica is offline.
- Controller will remove replica from created list of this partition in zookeeper if StopReplicaResponse indicates that this replica is successfully deleted.
- Controller will follow the same procedure specified in steps of "Topic gets created" when it creates replica on destination brokers, i.e. it will specify proper value for "create" in LeaderAndIsrRequest and manage created list of this partition in zookeeper.

8. Topic deleteion

Topic deletion should delete all replicas of all partitions of this topic and update metadata. If a replica is offline, then we handle it in the same way as if its broker is offline.

Public interface

Zookeeper

1) Add znode at /broker/topics/[topic]/partitions/[partitionId]/controller_managed_state with the following data format:

{
  "version" : int,
  "created" : [int]
}

These znodes will only be updated by controller. It will be updated in the follow scenario:

- When a replica is successfully created on the broker with LeaderAndIsrRequest(create=True), controller adds this replica (i.e. broker id) to the created list of this partition
- When a replica is successfully deleted on the broker with StopReplicaRequest(delete=True), controller deletes this replica from the created list of this partition.


2) Store data with the following json format in znode /log_dir_event_notification/log_dir_event_*

{
  "version" : int,
  "broker" : int,
  "event" : int    <-- This can be LogDirFailure
}

Protocol

Add a create field to LeaderAndIsrRequestPartitionState which will be used by LeaderAndIsrRequest 

LeaderAndIsrRequest => controller_id controller_epoch partition_states live_leaders
  controller_id => int32
  controller_epoch => int32
  partition_states => [LeaderAndIsrRequestPartitionState]
  live_leaders => [LeaderAndIsrRequestLiveLeader]
 
LeaderAndIsrRequestPartitionState => topic partition controller_epoch leader leader_epoch isr zk_version replicas
  topic => str
  partition => int32
  controller_epoch => int32
  leader => int32
  leader_epoch => int32
  isr => [int32]
  zk_version => int32
  replicas => [int32]
  create => boolean <-- NEW

Add a offline_replicas field to UpdateMetadataRequestPartitionState which will be used by UpdateMetadataRequest

UpdateMetadataRequest => controller_id controller_epoch partition_states live_brokers
  controller_id => int32
  controller_epoch => int32
  partition_states => [UpdateMetadataRequestPartitionState]
  live_brokers => [UpdateMetadataRequestBroker]
 
UpdateMetadataRequestPartitionState => topic partition controller_epoch leader leader_epoch isr zk_version replicas offline_replicas
  topic => string
  partition => int32
  controller_epoch => int32
  leader => int32
  leader_epoch => int32
  isr => [int32]
  zk_version => int32
  replicas => [int32]
  offline_replicas => [int32]  <-- NEW

Add a offline_replicas field to PartitionMetadata which will be used by MetadataResponse

MetadataResponse => brokers cluster_id controller_id topic_metadata
  brokers => [MetadatBroker]
  cluster_id => nullable_str
  controller_id => int32
  topic_metadata => TopicMetadata
 
TopicMetadata => topic_error_code topic is_internal partition_metadata
  topic_error_code => int16
  topic => str
  is_internal => boolean
  partition_metadata => [PartitionMetadata]
 
PartitionMetadata => partition_error_code partition_id leader replicas isr offline_replicas
  partition_error_code => int16
  partition_id => int32
  leader => int32
  replicas => [int32]
  isr => [int32]
  offline_replicas => [int32]  <-- NEW

Scripts

1) When describing a topic, kafka-topics.sh will show the offline replicas for each partition.

Changes in Operational Procedures

In this section we describe the expected changes in operational procedures in order to switch Kafka to run with JBOD instead of RAID. Administrators of Kafka cluster need to be aware of these changes before switching from RAID-10 to JBOD.

1) Need to reset broker's log directory state after disk is fixed.

As of current Kafka implementation, a broker will automatically create replica on a good log directory if it receives LeaderAndIsrRequest for a partition. No extra operation is needed e.g. if a broker is restarted after replacing failed disks with empty disks, or if broker is restarted on a new machine with empty disks, because the broker will automatically populate good log directories with data read from the partition leader. But after this KIP, administrator needs to explicitly execute the script kafka-log-dirs.sh to remove this broker from the list "created" of all partitions and restart broker after the hardware problem is fixed. This is needed to ensure that those replicas can be re-created on the good log directories of the broker after administrator has fixed the problem.

2) Need to adjust replication factor and min.insync.replicas

After we switch from RAID-10 to JBOD, the number of disks that can fail will be smaller if replication factor is not changed. Administrator needs to change replication factor and min.insync.replicas to balance the cost, availability and performance of Kafka cluster. With proper configuration of these two configs, we can have reduced disk cost or increased tolerance of broker failure and disk failure. Here are a few examples:

 - If we switch from RAID-10 to JBOD and keep replication factor to 2, the disk usage of Kafka cluster would be reduced by 50% without reducing the availability against broker failure. But tolerance of disk failure will decrease.
- If we switch from RAID-10 to JBOD and increase replication factor from 2 to 3, the disk usage of Kafka cluster would be reduced by 25%, the number of brokers that can fail without impacting availability can increase from 1 to 2. But tolerance of disk failure will still decrease.
- If we switch from RAID-10 to JBOD and increase replication factor from 2 to 4, the disk usage of Kafka would stay the same, the number of brokers that can fail without impacting availability can increase from 1 to 3, and number of disks that can fail without impacting availability would stay the same.

Compatibility, Deprecation, and Migration Plan

This KIP is a pure addition. So there is no backward compatibility concern.

The KIP changes the inter-broker protocol. Therefore the migration requires two rolling bounce. In the first rolling bounce we will deploy the new code but broker will still communicate using the existing protocol. In the second rolling bounce we will change the config so that broker will start to communicate with each other using the new protocol.

Test Plan

The new features will be tested through unit, integration, and system tests. In the following we explain the system tests only. 

 Note that we validate the following when we say "validate client/cluster state" in the system tests.

 - Brokers are all running and show expected error message
 - topic description shows expected results for all topics
 - kafka-log-dirs.sh can show the expected offline replicas for all brokers
 - A pair of producer and consumer can succcessfully produce/consume from a topic without message loss or duplication.

1) Log directory failure on leader during bootstrap

- Start 1 zookeeper and 3 brokers
- Create a topic of 1 partition with 3 replicas
- Start a pair of producer and consumer to produce/consume from the topic
- Kill the leader of the partition
- Remove the first log directory of the leader and create a file with the same path
- Start leader again
- Validated client/cluster state

2) Log directory failure on leader during runtime

 - Start 1 zookeeper and 3 brokers
- Create a topic of 1 partition with 3 replicas
- Start a pair of producer and consumer to produce/consume from the topic
- Remove the first log directory of the leader and create a file with the same path
- Validated client/cluster state
// Now validate that the previous leader can still serve replicas on the good log directories
- Create another topic of 1 partition with 3 replicas
- Kill the other two brokers
Start a pair of producer and consumer to produce/consume from the new topic
- Validated client/cluster state

3) Log directory failure on follower during runtime

- Start 1 zookeeper and 3 brokers
- Create a topic of 1 partition with 3 replicas
Start a pair of producer and consumer to produce/consume from the topic
- Remove the first log directory of a follower
- Validated client/cluster state
// Now validate that the follower can still serve replicas on the good log directories
- Create another topic of 1 partition with 3 replicas
- Kill the other two brokers
Start a pair of producer and consumer to produce/consume from the new topic
- Validated client/cluster state

Rejected Alternatives

- Let broker keep track of the replicas that it has created.

 The cons of this approach is that each broker, instead of controller, keeps track of the replica placement information. However, this solution will split the task of determining offline replicas among controller and brokers as opposed to the current Kafka design, where the controller determines states of replicas and propagate this information to brokers. We think it is less error-prone to still let controller be the only entity that maintains metadata (e.g. replica state) of Kafka cluster.


- Avoid adding "create" field to LeaderAndIsrRequest.
 If we don't add "create" field to LeaderAndIsrRequest, then broker will need to keep track of the list of replicas it has created and persists this information in either local disks or zookeeper.

Add a new field "created" in the existing znode /broker/topics/[topic]/partitions/[partitionId]/state instead of creating a new znode
If we don't include list of created replicas in the LeaderAndIsrRequset, the leader would need to read this list of created replicas from zookeeper before updating isr in the zookeeper. This is different from the current design where all information except isr are read from LeaderAndIsrRequest from controller. And it creates opportunity for race condition. Thus we propose to add a new znode to keep those information that can only be written by controller.

 

- Identify replica by 4-tuple (topic, partition, broker, log_directory) in zookeeper and various requests

This would require big change to both wire protocol and znode data format in order to specify log directory for every replica. And it requires Kafka to keep track of log directory of replica and update information in zookeeper every time a replica is moved between log directories on the same broker for load-balance purpose. We would like to avoid the additional code complexity and performance overhead.

- Setup one broker per disk and deploy multiple broker on the same machine
one-broker-per-disk would work and should require no major change in Kafka as compared to this KIP. So it would be a good short term solution. But it has a few drawbacks which makes it less desirable in the long term. Assume we have 10 disks on a machine. Here are the problems:
  

1) Our stress test result shows that one-broker-per-disk has 15% lower throughput

2) Controller would need to send 10X as many LeaderAndIsrRequest, MetadataUpdateRequest and StopReplicaRequest. This increases the burden on the controller and makes it increasingly a performance bottleneck.

3) Less efficient use of physical resource on the machine. The number of socket on each machine will increase by 10X. The number of connection between any two machine will increase by 100X.

4) Less efficient way to manage quota.

5) Rebalance between disks/brokers on the same machine will be less efficient and less flexible. Broker has to read data from another broker on the same machine via socket. It is also harder to do automatic load balance between disks on the same machine in the future.


Potential Future Improvement

1. Distribute segments of a given replica across multiple log directories on the same broker. It is useful but complicated. It is something that can be done later via a separate KIP.
2. Provide intelligent solution to select log directory to place new replicas and re-assign replicas across log directories to balance the load.
3. Have broker automatically rebalance replicas across its log directories. It is worth exploring separately in a future KIP as there are a few options in the design space.
4. Allow controller/user to specify quota when moving replicas between log directories on the same broker.

5. Let broker notify controller of ISR change and disk state change via RPC instead of using zookeeper
6. Handle various failure scenarios (e.g. slow disk) on a case-by-case basis. For example, we may want to detect slow disk and consider it as offline.
7. Allow admin to mark a directory as bad so that it will not be used/

 

 

  • No labels