kafka

1. Introduction

Consumer Group

  • One or more consumer threads form a consumer group. Each message in a partition can only be consumed by one Consumer in a Consumer group. If a message can be consumed by multiple consumers, So these consumers have to be in different groups. A consumer group can consume all partitions, but only one consumer can consume one partition “at a time”. Other consumers cannot consume messages in the same partition at the same time. If multiple businesses need the data of this topic, it is better to have multiple consumer groups. They all read message sequentially, and the values of offset do not affect each other. In this way, there is no lock competition, giving full play to the horizontal scalability and high throughput. This has led to the concept of distributed consumption.
  • When a consumer group is started to consume a topic, no matter how many partitions there are in the topic or how many consumer threads are configured in the consumer group, All consumer threads in the consumer group must consume all partitions. Even if there is only one consumer thread in the consumer group, that one consumer thread will consume all partitions. Therefore, the optimal design is that the number of consumer threads in the consumer group is equal to the number of partitions, thus achieving the highest efficiency.
  • When the number of consumers in a consumer group is smaller than the number of partitions in this topic, a conusmer thread consumes multiple partitions in groupA and groupB. In short, all partitions under this topic will be consumed. If the number of consumers in a consumer group is equal to the number of partitions in this topic, as shown in the groupC below, the efficiency is the highest, and each partition has one consumer thread to consume. If the number of consumers in the consumer group is greater than the number of partitions in the topic, there will be one free consumer thread in the group shown in the following figure. Therefore, when you set a consumer group, you just need to specify how many consumers there are in it. You don’t need to specify the corresponding consumer partition number. Consumer rebalance automatically.

Trigger condition for Consumer Rebalance

  • Adding or deleting a Consumer triggers the Consumer Group Rebalance
  • The increase or decrease of brokers triggers Consumer Rebalance

Consumer

  • The Consumer reads messages sequentially from partitions. So you have to maintain offset information about where you last read. An offset is shared by a consumer group. When a consumer consumes a message, the offset of the whole consumer group is increased by one, so as to avoid repeated consumption of the same message by different consumers.
  • The number of Kafka consumers in each service is smaller than the number of partitions in the corresponding topic, but the total number of consumers in all services is equal to the number of partitions. This is because all the consumers of a distributed service come from a consumer group, (Consumers belonging to the same consumer group cannot process the same partition. Different consumer groups can process the same topic.) So if you’re processing messages sequentially, you’re going to be processing repeated messages. In this case, two different business logics will start two consumer groups to handle a topic.
  • If the flow of producer increases, the number of parition of the current topic equals the number of consumers. In this case, the solution is horizontal expansion: increase partitions under topic and increase consumers under this consumer group.

Delivery Mode

  • There is no need to maintain the offset information for the message, because the offset is an incremented ID and the Producer just sends the message. Kafka’s producer usually sends a large batch of messages to a topic at a time. The load balance is inserted into a partition, and the offset is used as the increment ID.
  • The Consumer side needs to maintain the offset information of the message currently consumed by this partition. The high level API is maintained on Zookeeper, while the low level API is maintained by its own program. Kafka can only display the consumer part of the high level API, because Kafka does not know the partition offset of the low level API. When using the high level API, the message handler is automatically committed to offset+1 (or manually), and Kakfa handles messages without locking. Therefore, if processing the message fails, the message will be consumed again after the Consumer Thread restarts without committing offset+1. However, as a real-time processing system with high throughput and high concurrency, it can be tolerated that the system will be processed at least once. If this is not tolerated, you need to use the low level API to maintain the offset information, so when you want to commit the offset+1, you can do it yourself.

Topic and Partition

  • Topic is equivalent to a queue in MQ, a traditional messaging system. Messages sent by the producer end must be specified to which Topic, but not to which partition under the Topic. Because Kafka loads the received message evenly across different partitions in the topic (Hash (message) % [number of brokers])

  • This topic is divided into one or more partitions, and each partiton is equivalent to a sub-queue. In terms of physical structure, each partition corresponds to a physical directory (folder). The folder name is [TopicName][partition][serial number]. A topic can have countless partitions, which can be set according to service requirements and data volume. The number of partitions for a topic can be configured by changing num. Partitions in the Kafka configuration file at any time, specifying the number of parittions when creating a topic. The partiton number can also be modified through the tools provided by Kafka after Topic creation.

  • In general

    • The number of partitions in a Topic is greater than or equal to the number of brokers to improve throughput.
    • Replicas of the same Partition are distributed to different machines to ensure high availability.
  • When a new partition is added, the message in the partition will not be reassigned. The message data in the original partition will not be changed. The new partition will be empty at first. Messages that enter the topic then re-participate in the load balance for all partitions.

Partition Replica

  • Each partition can store copies on other Kafka Broker nodes so that the failure of one Kafka Broker node does not affect the Kafka cluster. Replica copies are stored in the order of the Kafka broker. For example, if there are 5 Kafka Broker nodes and a topic has 3 partitions with 2 copies per partition, then partition1 stores broker1, Broker2, Broker2, Broker3… (The number of replicas must not be greater than the number of Kafka Broker nodes; otherwise, an error is reported. The number of replicas here is actually the total number of copies of the partition, including one leader and other copies. This way, if a broker goes down, the data in Kafka remains intact. However, the higher the number of replicas, the more stable the system will be, but the resources and performance will decline. Fewer replica copies may cause data loss in the system.

    • How to send a message:
      • The producer first sends messages to the partition leader, who in turn sends messages to the followers of other partitions. (It would be too slow if the producer sent to each replica)
    • Before sending an ACK to Producer, how many replicas need to ensure that the message has been received:
      • It depends on the number of ACK configurations
    • How to deal with a Replica not working:
      • If the faulty partition replica is not in the ACK list, the producer is sending a message to the partition leader, and the partition leader does not respond to the message sent to the partition follower. It won’t affect the whole system, and it won’t be a problem. If the faulty partition replica is in the ACK list, the producer sends a message waiting for the faulty partition to write a message successfully until time out, and finally returns a failure. Because the partition replica in an ACK list does not respond, Kafka automatically removes the faulty partition replica from the ACK list. In the future, there will be no non-working partition replica in the ACK list when producer sends messages.
    • How to deal with the Failed Replica recovery:
      • If the partition replica is not in the ACK list before, it can be managed by Zookeeper again after being started. Then, when the producer sends the message, The partition leader continues to send messages to the partition follower. If the partition replica is in the ACK list, manually add the partition replica to the ACK list after the replica is restarted. (The ACK list is manually added. When a faulty partition replica occurs, it is automatically removed from the ACK list.)

Partition the leader and the follower

  • A partition can also be a leader or a follower. The leader is the primary partition. When a producer writes kafka, he writes to the partition leader first, who then pushes the leader to the followers of other partitions. Information about the partition leader and followers is controlled by Zookeeper. If the broker node where the partition leader resides breaks down, Zookeeper turns followers into parition leader on the partition followers of other brokers.

Algorithm for Topic partition allocation and Partition Replica

  • Sort Broker (size=n) and Partition to be allocated.
  • Allocate the ith Partition to the (I %n) Broker.
  • Allocate the JTH Replica of the ith Partition to the (I + j) % n) Broker.

Message reliability

  • Message delivery reliabilityKafka provides three modes for determining how a message is successfully delivered:
    • Don’t care. Send it as a success. (There is no guarantee that messages will be successfully delivered to the broker.)
    • In the master-slave model, a message is successfully delivered only when the Master and all slaves receive the message. (The model provides the highest delivery reliability, but compromises performance)
    • As long as the Master acknowledges receiving the message, the delivery is successful. (In practical use, this model is selected according to application characteristics, and in most cases, reliability and performance will be selected)
  • The reliability of messages on the broker
    • Because messages persist to disk, data on a broker is not lost if it is stopped properly. If stop is not performed properly, messages that cannot be written to disks may be lost. This can be alleviated by setting flush page cache interval and threshold. However, frequent disk writes may affect performance.
  • Reliability of message consumption
    • Kafka provides a “At least once” model, because the message reading progress is provided by the offset, which can be maintained by the consumer or in ZooKeeper. However, when the consumer dies after the message is consumed, the offset is not written back immediately. This can be alleviated by adjusting the commit offset cycle, the threshold, or even the consumer’s own transaction resolution between consumption and Commit offset, but if your application doesn’t care about double consumption, don’t resolve it at all for maximum performance.

Partition ack

  • If ACK =1: indicates that producer successfully writes to the partition leader, the broker returns success regardless of whether the other partition followers have successfully written to the leader.
  • When ACK =2: indicates that the producer succeeded in writing to the partition leader and one of the followers, the broker returns success regardless of whether the other followers succeeded.
  • If ack=-1, kafka Broker returns a success message only when producer writes successfully.

Note: If ack=1, one of the brokers breaks down and causes the follower and leader of the partition to switch, data will be lost.

The message state

  • In Kafka, the message state is stored in the consumer. The broker does not care which message is consumed or by whom. Instead, it records an offset value that points to the partition where the message will be consumed next. A message on a broker may be consumed more than once.

The message persistence

  • Kafka persists messages to local file systems and is extremely efficient. IO reads are known to be resource-intensive and have the slowest performance, which is why database bottlenecks are often on IO, requiring SSD drives to be replaced. But Kafka, as a high-throughput MQ, can persist messages to files very efficiently. This is because Kafka writes sequentially with o(1) complexity and is very fast, which is the reason for the high throughput. Since write persistence of messages is sequential, messages are also consumed sequentially when consumed, ensuring that partition messages are consumed sequentially. General machine, single machine 100K data per second.

The validity of the message

  • Kafka retains messages for a long time so that consumers can consume them multiple times, although many of the details are configurable.

Produer

  • The Producer sends messages to a Topic without specifying the partition. Kafka uses partition acks to control whether a message is sent successfully and sends the message back to the producer. The producer can have as many threads as he or she wants. The delivery guarantee on the Producer end is At least once by default. The Producer can also be set to send asynchronously At most once. Producer can use primary key idempotency to achieve Exactly once.

Kafka has high throughput

  • The high throughput of Kafka is reflected in read and write. Distributed concurrent reads and writes are very fast. The write performance is reflected in sequential writes with o(1) time complexity. Read performance is reflected in sequential read with o(1) time complexity and partition of topics. Consume threads in consume group can perform sequential read with high performance.

Kafka Delivery Guarantee

  • The message may be lost At most once and never repeated
  • Messages are never lost, but may be transmitted repeatedly
  • Exactly once each message is transmitted once and only once, which is what the user wants.

Batch send

  • Kafka supports bulk sending of message sets to improve push efficiency.

push-and-pull

  • In Kafka, producers and consumers push and pull messages to the broker, while consumers pull messages from the broker. Both producers produce and consume messages asynchronously.

Relationships between brokers in a Kafka cluster

  • Instead of a master-slave relationship, brokers have the same status in the cluster and can be added or removed at will.

Load balancing

  • Kafka provides a metadata API to manage load between brokers (for Kafka0.8.x, zooKeeper is the main load balancer for 0.7.x).

Synchronous asynchronous

  • Producer adopts asynchronous push mode, which greatly improves the throughput of Kafka system (synchronous or asynchronous mode can be controlled by parameters).

Partition mechanism

  • Kafka’s broker supports message partitions. The Producer can decide which partition to send messages to. The order in which the Producer sends messages is the order in which the message is sent. A topic can have multiple partitions, and the number of partitions is configurable. The partition concept allows Kafka to scale horizontally as MQ with huge throughput. Replica copies exist on different Kafka broker nodes. The first partition is the leader and the others are followers. Messages are written to the leader first. Then the partition leader pushes it to the Parition follower. So Kafka can scale horizontally, that is, partition.

Offline data loading

  • Kafka is also ideal for loading data into Hadoop or a data warehouse due to its support for scalable data persistence.

Real-time data versus offline data

  • Kafka supports both offline and real-time data. Because Kafka messages are persisted to files and expiration dates can be set, Kafka can be used as an efficient store for offline data for later analysis. Of course, as a distributed real-time messaging system, most of the time it is still used for real-time data processing, but when the cosumer consumption power is down, the message persistence can be used to stack data in Kafka.

Plug-in support

  • A number of plugins have been developed by the active community to extend Kafka’s capabilities, such as those for Storm, Hadoop, and Flume.

The decoupling

  • It acts as an MQ that decouples systems from producers and consumers in asynchronous operations

redundant

  • Replica has multiple copies to ensure that a breakdown of one broker node does not affect the entire service

scalability

  • Broker nodes can expand horizontally, partitions can increase horizontally, and partition replicas can increase horizontally

peak

  • As Kafka expands horizontally as traffic surges, the app needs to continue to function

recoverability

  • If a component in the system fails, the replica copy of the partition does not affect the whole system.

Order guarantee

  • Since Kafka’s producers write messages and consumers read messages sequentially, efficient performance is guaranteed.

The buffer

  • Since the producer side might be very simple, and the consumer side might be complicated and have database operations, the producer would be faster than the consumer. Without Kafka, the producer would call the consumer directly. Then it will cause the whole system processing speed is slow, add a layer of Kafka as MQ, can play a buffer role.

Asynchronous communication

  • As MQ, the Producer communicates asynchronously with the Consumer

2.Kafka file storage mechanism