Tuesday, April 12, 2016

KAFKA PERFORMANCE TUNING

We all know the power and advantages of KAFKA. Apache Kafka is publish-subscribe messaging system which basically has three major components
  • KAFKA CONSUMER
  • KAFKA PRODUCER and
  • KAFKA BROKER
This blog is all about how we can achieve maximum throughput while planning to have KAFKA in production or in POCs.
There are many organizations running Kafka in their production and also they have provided default configuration to maximize Kafka performance. I’ll be using some of the existing configuration details along with some extra information to describe the configuration we need to take care in setting up the cluster configuration.
Before diving into Kafka tuning pool lets get some idea what exactly Kafka is and what feature it provides-
INTRODUCTION
Apache Kafka is an open-source message broker project, which was developed by Apache Software Foundation.
Kafka provides distributed, partitioned and replicated commit log-service with functionality of a messaging system.
The features which makes Kafka more reliable and fast are mentioned below:
High Throughput
A single Kafka broker is capable of handling hundreds of megabytes of reads and writes per second from thousands of clients.
Scalable
Kafka is designed in such a way that it can be elastically and transparently expanded without any downtime. It means you can add new broker node at any point of time without shutting down your Kafka cluster.
Durability
All the messages are persisted on the disk and also has a replica of it within the cluster to avoid data loss at any point. Brokers are capable enough of handling terabytes of data without affecting the performance.
Distributed
Kafka also provides distributed processing of messages and its cluster-centric design offers you strong durability and fault-tolerance. Kafka allows us to have partition of any topic which will help us to increase throughput of the system.
So, at very high level we can say that Producers produced the messages and sends them to the Kafka cluster over the network, which will be consumed by respective Consumers who has subscribed for it.
kafka architecture
KAFKA USECASE DIAGRAM
KafkaUsecase
TUNING KAFKA
When we talk about tuning Kafka, there are few configuration parameters to be considered. The most important configurations to improve performance are the one, which controls the disk flush rate.
We can also divide these configurations on component basis as well.
Most important configurations which needs to be taken care at Producer side are –
  • Compression
  • Batch size
  • Sync or Async
The important configuration at Consumer side is –
  • Fetch size
We can also have multiple Consumers running to fetch maximum data from partitioned topic available on Kafka Brokers.
Apache has maintained a very good documentation to describe all the configurations parameter and their definitions. To get the details please follow the link.
Production server configurations
Here are some the configurations parameter and their values –
Property NameDefaultDescription
num.replica.fetchers1Number of threads used to replicate messages from leaders. Increasing this value can increase the degree of I/O parallelism in the follower broker.
replica.fetch.max.bytes1048576The number of bytes of messages to attempt to fetch each partition in the fetch requests. The replicas are sent to the leader.
replica.fetch.wait.max.ms500The maximum amount of time to wait time for data to arrive on the leader in the fetch requests sent by the replicas to the leader.
replica.high.watermark.checkpoint.interval.ms5000The frequency with which each replica saves its high watermark to disk to handle recovery.
replica.socket.timeout.ms30000The socket timeout for network requests to the leader for replicating data.
replica.socket.receive.buffer.bytes65536The socket receive buffer for network requests to the leader for replicating data.
replica.lag.time.max.ms10000If a follower hasn’t sent any fetch requests for this window of time, the leader will remove the follower from ISR (in-sync replicas) and treat it as dead.
replica.lag.max.messages4000If a replica falls more than this many messages behind the leader, the leader will remove the follower from ISR and treat it as dead.
controller.socket.timeout.ms30000The socket timeout for commands from the partition management controller to the replicas
controller.message.queue.size10The buffer size for controller-to-broker-channels
num.partitions8The default number of partitions per topic if a partition count isn’t given at topic creation time
message.max.bytes1000000This is largest message size Kafka will allow to be appended to this topic. Note that if you increase this size you must also increase your consumer’s fetch size so they can fetch such large messages.
auto.create.topics.enabletrueEnable auto creation of topic on the server. If this is set to true then attempts to produce data or fetch metadata for a non-existent topic will automatically create it with the default replication factor and number of partitions.
log.index.interval.bytes4096The byte interval at which we add an entry to the offset index. When executing a fetch request the server must do a linear scan for up to this many bytes to find the correct position in the log to begin and end the fetch. So setting this value to be larger will mean larger index files (and a bit more memory usage) but less scanning. However the server will never add more than one index entry per log append (even if more than log.index.interval worth of messages are appended). In general you probably don’t need to mess with this value.
log.index.size.max.bytes10485760The maximum size in bytes we allow for the offset index for each log segment. Note that we will always pre-allocate a sparse file with this much space and shrink it down when the log rolls. If the index fills up we will roll a new log segment even if we haven’t reached the log.segment.bytes limit. This setting can be overridden on a per-topic basis
log.flush.interval.ms10000The maximum time between fsync calls on the log. If used in conjuction with log.flush.interval.messages the log will be flushed when either criteria is met.
log.flush.interval.messages20000The number of messages written to a log partition before we force a fsync on the log. Setting this lower will sync data to disk more often but will have a major impact on performance. We generally recommend that people make use of replication for durability rather than depending on single-server fsync, however this setting can be used to be extra certain.
log.flush.scheduler.interval.ms2000The frequency in ms that the log flusher checks whether any log is eligible to be flushed to disk.
log.retention.check.interval.ms300000The period with which we check whether any log segment is eligible for deletion to meet the retention policies.
log.segment.bytes1073741824The log for a topic partition is stored as a directory of segment files. This setting controls the size to which a segment file will grow before a new segment is rolled over in the log. This setting can be overridden on a per-topic basis
num.io.threads8The number of I/O threads that the server uses for executing requests. You should have at least as many threads as you have disks
num.network.threads8The number of network threads that the server uses for handling network requests. You probably don’t need to change this.
socket.request.max.bytes104857600The maximum request size the server will allow. This prevents the server from running out of memory and should be smaller than the Java heap size.
socket.receive.buffer.bytes1048576The socket receive buffer for network requests
socket.send.buffer.bytes1048576The SO_SNDBUFF buffer the server prefers for socket connections
queued.max.requests16The number of requests that can be queued up for processing by the I/O threads before the network threads stop reading in new requests
fetch.purgatory.purge.interval.requests100The purge interval (in number of requests) of the fetch request purgatory
producer.purgatory.purge.interval.requests100The purge interval (in number of requests) of the producer request purgatory
All the above configurations are broker side configuration. In most of the cases these configuration works well but there are few configurations, which you can modify as per the availability of the cluster environment and machine configuration which are listed below:
num.replica.fetchers
This configuration parameter defines the number of threads which will be replicating data from leader to the follower. Value of this parameter can be modified as per availability of thread. If we have threads available we should have more number of replica fetchers to complete replication in parallel.
replica.fetch.max.bytes
 This parameter is all about how much data you want to fetch from any partition in each fetch request. It’s good to increase value for this parameter so that it helps to create replica fast in the followers.
replica.socket.receive.buffer.bytes
In case of less thread available for creating replica, we can increase the size of buffer. It will help to hold more data if replication thread is slow as compared to the incoming message rate.
num.partitions
This is the very important configuration which we should be taken care while having Kafka in live. As many partitions are there, we can have that level of parallelism and write data in parallel which will automatically increase the throughput.
Increasing number of partition can also slow down your performance and throughput if the system configuration is not capable of handling it.
Now the question that rises from the above statement is How??
Yes, this is true if system does not have sufficient threads or just have single disk then it does not make sense in creating lots of partition for better throughput. Creating more partition for a topic is directly dependent on available threads and disk.
num.io.threads
 Setting value for I/O threads directly depends on how much disk you have in your cluster. These threads are used by server for executing request. We should have at least as many threads as we have disks.
 Apart from these configurations there are few other factors which I have mentioned earlier in the blog like batch size and sync/async mode of message transfer.
 When we think about batch size its always confusing what batch size will be optimal. Large batch size may be great to have high throughput but you might feel latency issue in that. So, we can conclude that latency and throughput is inversely proportional to each other.
ThroughputVsLatency
But there are ways to have low latency with high throughput where we have to choose a proper batch-size. We can also use queue-time or refresh-interval to find the required right balance.
References:

No comments:

Post a Comment