?? Apache Kafka Internals-Part1

Kafka's internal architecture is designed to provide a robust, durable, and high-performance platform for handling real-time data feeds. Its distributed nature allows it to scale out and handle very large volumes of data while ensuring that messages are processed in the order they are received. The combination of these features makes Kafka a powerful tool for building real-time streaming data pipelines and applications.

Kafka's architecture is divided into two main layers: the storage layer and the compute layer.

• The storage layer is responsible for efficiently storing data in a distributed manner, ensuring that as storage needs grow, the system can scale to accommodate this growth. This is achieved through topics, which are divided into partitions that are distributed and replicated across the brokers to ensure fault tolerance and high availability.
• The compute layer consists of several APIs that enable Kafka's powerful application layer. These include the Producer API for writing events to topics and the Consumer API for reading events. On top of these, there are APIs for integration and processing, such as Kafka Connect for integrating with external systems and Kafka Streams for real-time stream processing.

?? What is the Apache ZooKeeper?

Apache ZooKeeper is a centralized service that is crucial for distributed applications. It provides a reliable way to coordinate and manage a cluster of machines. At its core, ZooKeeper keeps track of data in a tree-like structure called znodes, which can be thought of as files and directories. Znodes can be regular or ephemeral; regular znodes persist until they are explicitly deleted, while ephemeral znodes disappear when the session that created them ends.

ZooKeeper ensures that the entire distributed system has a consistent view of the data. It does this by following a simple protocol: when a process changes the data in a znode, ZooKeeper replicates the change across all nodes in the cluster, ensuring that all nodes see the same view of the data.

Processes can set watches on znodes. When a watched znode changes, ZooKeeper notifies all the watching processes of the change. This feature is particularly useful for coordination tasks like leader election, configuration management, and cluster management.ZooKeeper's design aims to be simple. It doesn't handle large data values but focuses on managing synchronization primitives such as locks and barriers efficiently.

?? What is the Cluster Membership?

Kafka cluster membership refers to the collection of brokers that form a Kafka cluster. These brokers work together to provide a distributed, fault-tolerant, and scalable messaging platform. Each broker in the cluster has a unique identifier, which is used to manage and maintain the cluster's overall health and performance. The brokers coordinate with each other to handle multiple data streams, ensuring low latency, data durability, and scalability.

An ephemeral node in the context of Kafka and ZooKeeper is a temporary node that is automatically removed when the session that created it ends. This is often used to represent the presence of a broker in the cluster; when the broker is active, its ephemeral node exists in ZooKeeper, and if the broker disconnects or crashes, the ephemeral node is deleted.

When a broker is disconnected from ZooKeeper, the ephemeral node associated with that broker is removed. This signals to the rest of the Kafka cluster that the broker is no longer available. As a result, Kafka will trigger a rebalance of the partitions that were managed by the disconnected broker, electing new leaders for those partitions if necessary.

If consumers or producers were using the disconnected broker, they would be affected by the disconnection. Kafka clients are designed to handle such failures by detecting broker unavailability and reconnecting to other available brokers in the cluster. They may experience a temporary disruption but will continue to function by connecting to other brokers.

Adding a new broker to the Kafka cluster can serve as an alternative pathway for producers and consumers. When a new broker is added, it is assigned a unique broker.id and configured to connect to the existing ZooKeeper ensemble. The new broker can then start handling new partitions and replicas. To balance the cluster load, partitions may be reassigned to the new broker using Kafka's partition reassignment tools. This process is dynamic and does not require a restart of the existing brokers or ZooKeeper nodes.

?? What is the Controller?

The Controller in a Kafka cluster is a broker with the responsibility of managing the states of partitions and replicas. It is a critical component that performs administrative tasks such as reassigning partitions and handling broker failures. There can only be one active controller at any time, and it is elected from among the brokers in the cluster. The controller's duties include:

• Leader Election: When a broker fails, the controller elects a new leader for the partitions that were led by the failed broker.
• Cluster Management: It manages the overall state of the cluster, including tracking which brokers are alive and coordinating updates to the cluster metadata.
• Replica Reassignment: The controller reassigns partition replicas to different brokers when necessary to balance the load or in response to broker failures.

A ZooKeeper watch is a notification mechanism that allows Kafka brokers to be informed about changes in the cluster state. For example, if a broker goes down, ZooKeeper can notify the other brokers through a watch, triggering the controller to elect a new leader for the affected partitions.

??What is the Replication?

Replication in Kafka is a fundamental feature that ensures data durability and high availability. It involves creating multiple copies of data across different brokers within the Kafka cluster. Each topic in Kafka is divided into partitions, and these partitions are replicated across the cluster's brokers based on a configurable replication factor. The replication factor determines how many copies of each partition are made.

In the context of replication, each partition has one special broker known as the leader. The leader handles all read and write requests for that partition. Other brokers that hold copies of the partition act as followers. The followers replicate the data from the leader and can take over as the new leader if the current leader fails. This mechanism ensures that even in the event of a broker failure, the data remains available and the system can continue to operate without data loss.

The leader is crucial because it maintains the integrity and consistency of the data. It ensures that all followers are up-to-date before confirming writes to the clients. This process is managed through a set of in-sync replicas (ISRs), which are the followers that have fully replicated the leader's log of messages.

There are two types of replicas:

1. Leader Replica: For each partition, one of the replicas is designated as the leader. The leader replica handles all read and write requests for that partition. It is also responsible for updating the followers with the latest data.

2. Follower Replica: These are the replicas that replicate the data from the leader. They do not serve client requests directly. Instead, they synchronize with the leader, ensuring that they have the same data as the leader.

?? What is the Request?

Requests are structured communications between clients (like producers and consumers) and the Kafka brokers. These requests are part of Kafka's binary protocol over TCP, which defines a set of API operations that clients can perform on the Kafka cluster.

Requests in Kafka are made up of a standard format that includes a header and a body. The header typically contains information such as the API key (which identifies the type of request), API version (which specifies the version of the protocol to be used), a correlation ID (a unique identifier for the request provided by the client), and a client ID (an identifier for the client making the request).

The Kafka protocol supports a variety of request types, including but not limited to:

• Produce: Sending messages to a broker.
• Fetch: Retrieving messages from a broker.
• Offsets: Requesting information about the available offsets for a topic partition.
• Metadata: Requesting information about the available brokers and the structure of the Kafka cluster.

The request processing involves several components that work together to handle client requests and responses. Here's a simplified explanation of each term:

• Acceptor Thread: The Acceptor Thread is part of the network layer that handles incoming connection requests from producers and consumers. When a client wants to establish a connection with a Kafka broker, it sends a connection request to the server's socket. The Acceptor Thread listens on this server socket and is responsible for accepting these connection requests. This is a thread that listens for new connections from clients. Once a connection is established, the acceptor thread hands it off to a processor thread for further handling.
• Network Thread: These threads handle network requests from clients, such as produce and fetch requests. They are responsible for reading requests from the socket, placing them in a request queue, and picking up responses from the response queue to send back to the clients.
• Queue: In the context of Kafka's request processing, there are typically two queues involved: a request queue and a response queue. The request queue holds incoming requests from clients that are waiting to be processed, and the response queue holds responses that are ready to be sent back to the clients.
• I/O Thread: I/O threads are responsible for the actual reading and writing of data to disk. They process the requests from the request queue, which often involves reading from or writing to the log, and then place the responses in the response queue.
• Response Queue: This is where responses are placed after being processed by the I/O threads. Network threads will pick up these responses and send them back to the appropriate clients.

?? Producer Request:

1. Message Creation: The producer creates a message to be sent to a Kafka topic. This message can optionally include a key that determines which partition the message will be sent to.
2. Partitioning: If a key is provided, the producer uses a partitioner to determine which partition of the topic the message should be sent to. If no key is provided, the partition is chosen either randomly or based on a round-robin algorithm.
3. Serialization: The message and key are serialized into bytes so they can be transmitted over the network. Serialization can be customized using different serializers provided by Kafka.
4. Batching: To improve efficiency, the producer batches multiple messages together into a single request. This batch is then sent to the broker that is the leader for the target partition.
5. Sending Request: The producer sends the produce request to the Kafka broker that is the leader for the chosen partition.
6. Acknowledgement: Depending on the producer's configuration, it may wait for an acknowledgement from the broker. This can be: acks=0: No acknowledgement is required, and the producer won't wait for a response. acks=1: The producer waits for the leader broker to write the message to its local log but not for any replicas to copy it. acks=all: The producer waits for all in-sync replicas to acknowledge the message. This provides the strongest guarantee that the message won't be lost but has the highest latency.
7. Response Handling: Once the broker processes the request, it sends back a response to the producer. If the message was successfully written to the log, the response will indicate success; otherwise, it will include error information.
8. Retries: If the produce request fails, the producer may retry sending the message based on its configuration settings for retries and retry back-off time.
9. Committing Messages: Once the message is acknowledged by the broker (based on the acks configuration), it is considered committed and will be available for consumers to read.

?? Fetch Request:

1. Consumer Request: The consumer sends a fetch request to the Kafka broker, specifying the topics, partitions, and offsets from where it wants to start fetching messages.
2. Broker Validation: The broker validates the fetch request, checking if the consumer is authorized to read from the specified partitions and if the requested offsets are valid.
3. Data Retrieval: The broker retrieves the messages starting from the given offsets up to the maximum bytes specified in the fetch request.
4. Message Set Preparation: The broker prepares a message set, which includes the messages to be sent back to the consumer. This set may also include any messages that are available up to the fetch size limit set by the consumer.
5. Response Sent: The broker sends the message set back to the consumer in the response to the fetch request.
6. Offset Update: Upon receiving the messages, the consumer processes them and updates its offsets, marking the messages as consumed.
7. Long Polling (Optional): If the fetch.min.bytes configuration is set, the broker may wait until enough data is available to meet this minimum amount before sending a response. This is known as long polling and can reduce the number of fetch requests made by the consumer.
8. Fetch Session Management (Optional): Kafka may use fetch sessions to optimize fetch requests from consumers. If a session is established, the consumer can fetch data using the session ID, reducing the overhead of repeatedly sending topic and partition information.
9. Error Handling: If an error occurs, such as an invalid offset or an authorization failure, the broker includes error codes in the fetch response. The consumer can then take appropriate action, such as refreshing metadata or seeking to a valid offset.