Apache Kafka article

Inter-service communication

You can learn more about the methods of internetworking at LINK.

Classic queuing services

Queuing systems typically consist of three basic components:

1. server(broker)
2. producers who send messages to a named queue, pre-configured by the administrator on the server
3. consumers who read the same messages as they are appearance

In web applications, queues are often used for deferred processing. events or as a temporary buffer between other services, so thereby protecting them from load surges.

Consumers receive data from the server (broker) using two different request models: pull or push.

pull model

— consumers themselves send a request once every n seconds to the server for receiving a new batch of messages. With this approach, clients can effectively control your own workload. Besides, The pull model allows you to group messages into batches, thus achieving better throughput. The disadvantages of the model can be attribute potential load imbalance between different consumers, as well as higher data processing latency.

push model (RabbitMQ)

— the server makes a request to the client, sending it a new portion of data. It reduces message processing latency and allows efficient balance the distribution of messages among consumers.

A typical message lifecycle in queuing systems is:

1. The producer sends a message to the server.
2. The consumer fetches (from the English fetch - bring) a message and its unique server identifier.
3. The server marks the message as in-flight. Messages in this state are still stored on the server, but are temporarily not delivered to others to consumers. The timeout of this state is controlled by a special setting.
4. The consumer processes the message following business logic. Then sends an ack or nack request back to the server using a unique identifier obtained earlier - thereby either confirming the successful processing of the message, or signaling error.
5. If successful, the message is deleted from the server forever. When error or state timeout in-flight message is delivered to the consumer for reprocessing.

Like queue processing services, Kafka conventionally consists of three components:

1. server (broker)
2. producers - they send messages to the broker
3. consumers - read these messages using the pull model

Kafka uses a pull model approach that allows users to request message packages from specific offsets. Users can use batch message processing to increasing throughput and efficient message delivery.

What is Apache Kafka used for?

Apache Kafka is best for streaming from A to B without complex routing, but with maximum throughput. The tool excels at stream processing and modeling changes in the system as a sequence of events. Kafka also can be used for data processing in a multi-stage pipeline processing. Kafka is a great solution if you need a framework for storing, reading, re-reading and analyzing streaming data. Her Our strong point is real-time data processing and analysis. The tool is ideal for permanently storing messages or for regularly checked systems.

Where can Apache Kafka be used?

• Logs and metrics: Kafka can be used for collecting, storing and analyzing logs and metrics to extract valuable information about the system.
• Microservice architecture: Apache Kafka can be used for communication and synchronization between microservices in a distributed system. It can be used to transfer messages and events between different services, ensuring reliable delivery and guarantee of message processing.
• Data stream processing: Apache Kafka can used to process data streams in real time. He can receive data streams from various sources, e.g. sensors, IoT devices, etc., and process them practically instantly. This makes it ideal for solving machine tool problems. training, data analysis and monitoring.
• Web consumer analysis: Apache Kafka can be used to analyze web consumers by processing data into real time from various sources. For example it could be used to build a recommendation system that processes data on user behavior and provides personalized recommendations.
• Integration and streaming data processing: Apache Kafka can be used to integrate and process data in real time with other systems such as databases, systems data management and other data sources. It provides powerful tools for performing streaming data transformations and aggregations.
• Persistence logs: Kafka can used as persistence logs to save and data recovery. It can be used to record and restoring the system state in case of failures, etc.

1. Installing the package and creating a directory system
   1. Create a folder for the cluster
   2. Let's create working directories for three nodes kafka_server_1, kafka_server_2 and kafka_server_3.
      
   3. Download, unpack and copy the contents to each broker directories https://downloads.apache.org/kafka/3.5 .0/kafka_2.13-3.5.0.tgz :
      
2. Setting up nodes
   1. Kafka broker configuration (from the kafka_server_1, kafka_server_2 and kafka_server_3 respectively)

      config/server.properties

      • broker id(0, 1, 2)
      • client port( 9092, 9093, 9094)
      • Zookeeper port(2181, 2182, 2183)
      • log directories(/tmp/kafka-logs-1, /tmp/kafka-logs-2, /tmp/kafka-logs-3)

      broker.id=0
                                   listeners=PLAINTEXT://:9092
                                   zookeeper.connect=localhost:2181
                                   log.dirs=/tmp/kafka-logs-1

   2. Zookeeper node configuration:

      
                                   config/server.properties

      • data directory (/tmp/zookeeper_1, /tmp/zookeeper_2, /tmp/zookeeper_3)
      • client port(2181, 2182, 2183)
      • maximum number of client connections and limits connections
      • ports for data exchange between nodes (i.e. we inform each node about existence of others)

      
                                  dataDir=/tmp/zookeeper_1
                                  clientPort=2181
                                  maxClientCnxns=60
                                  initLimit=10
                                  syncLimit=5
                                  tickTime=2000
                                  server.1=localhost:2888:3888
                                  server.2=localhost:2889:3889
                                  server.3=localhost:2890:3890
                                                  

   3. Creating directories for Zookeeper nodes, recording node ids in service files:

      
                                   mkdir -p /tmp/zookeeper_[1..3]
                                   echo "1" >> /tmp/zookeeper_1/myid
                                   echo "2" >> /tmp/zookeeper_2/myid
                                   echo "3" >> /tmp/zookeeper_3/myid
                                                   

3. Launching zookeeper nodes and brokers

   To start a cluster you need from each root directory( kafka_server_1, kafka_server_2 and kafka_server_3) run the scripts:

   
                        sudo bin/zookeeper-server-start.sh config/zookeeper.properties
                        sudo bin/kafka-server-start.sh config/server.properties
                                    

4. Creation of a topic, producer and consumer:

   
                        sudo bin/kafka-topics.sh --create --bootstrap-server localhost:9092,localhost:9093,localhost:9094 --replication-factor 3 --partitions 3 --topic TestTopic
                                    

   We created a topic (topic) with 3 partitions on 3 servers and duplicated all partitions on each server (replication factor).

   You can find out information about a topic with the following command:

   
                        bin/kafka-topics.sh --describe --topic TestTopic --bootstrap-server localhost:9094
                                    

   We get:

   
   • Leader - server with the main instance of the partition
   • replica - server on which information is duplicated
   • ISR - servers that take on the role of leaders in case of leader failure

1. First, let's create console producers and consumers:

   
                        sudo bin/kafka-console-producer.sh --broker-list localhost:9092,localhost:9093,localhost:9094 --topic TestTopic
                        sudo bin/kafka-console-consumer.sh --bootstrap-server localhost:9092,localhost:9093,localhost:9094 --from-beginning --topic TestTopic
                                    

   

   On the right is the producer who sends the message, on the left - consumer who reads.

2. Let's stop one of the kafka nodes (kafka_server_1):
   

   We see that the Leader and Isr parameters have changed. For Partition: 1 the leader changed from the first broker with id=0 to the second with id=1, as well as must be according to the Isr parameter.

   Let's try to write a message after the first broker stops:

   

   We see that the message has arrived.

3. Let's stop one of the kafka nodes (kafka_server_1):
   

   Since there is only one broker left to work, accordingly he become a leader for all parties.

   Let's try to write a new message:

   

   We see that this time too, the message got through. Thus we observe Kafka cluster fault tolerance.

1. Let's create a new topic with 3 ports and 3 replicas on 3 brokers to make the recording uniform across all partitions from one consumer needs to add the RoundRobin class to config/producer.properties

   
                        partitioner.class=org.apache.kafka.clients.producer.RoundRobinPartitioner
                                    

   

   There are 3 directories created in the kafka broker log directories:

   

   We have created three partitions in the topic, and each has its own directory in file system on each broker. In leader-epoch-checkpoint information about the current epochal number of the leader and his identifier, as well as information about the last applied and sent journal entries for each leader. This is used to control repeat operations and reorder records in case of emergency or change of leader. In partition.metadata contain information about each topic partition in the Kafka cluster. They include the following information:

   • Partition ID: unique identifier section within the topic.
   • Topic ID: ID of the topic to which belongs to the section.
   • Section Leader: the broker who is the leader of the section and is responsible for reading and writing data to this section.
   • Partition Replicas: All brokers that contain a copy this section.
   • Version: The current version of the section's metadata.
   • State: The state of the partition, such as "Online" or "Offline".

   The log-start-offset-checkpoint parameter is also visible, which used in Apache Kafka to store information about the latest recorded offset (offset) in the log of each partition, which is necessary to continue reading data from where you left off consumer. Also used for recovery after a failure. If Kafka broker crashes, it recovers from last saved checkpoint file. This allows you to restore data and state of the broker before the failure.

2. Let's add a few messages to the topic. Let's connect to the first broker producer, who is the leader of one of the parties and will be sent to he has several messages:
   

   Messages are written to the .log file, so we can see where the messages were recorded - in the TestTopic-1 port:

   

   Each new message in the partition is assigned an Id that is 1 more previous one. This Id is also called offset. At the first messages have an offset of 0, the second one has an offset of 1, etc., each next one is always 1 more than the previous one. Let's check our example with the tool kafka:

   
                       bin/kafka-run-class.sh kafka.tools.DumpLogSegments --deep-iteration --print-data-log --files /tmp/kafka-logs-1/TestTopic-1/00000000000000000000.log
                                   

   

   Here we see offset, Create Time, size key(keySize) and value(valueSize) , and also the message itself (payload).

3. Let's make sure that when connecting to different brokers - record carried out only in one partition, for which the broker is leader. Let's connect to the second broker and write several messages:
   

   Let's check which partition the data was written to:

   

   The size of the TestTopic-0 log file has changed, let's check if there is messages we sent to the second broker:

   

   We see that the sent messages were recorded in the TestTopic-0 partition.

   Let's repeat all the steps for the third broker:

   
   
   

   We conclude that the first broker is the leader of the TestTopic-1 partition, the second broker is TestTopic-0, and the third is TestTopic2.