Apache Kafka Survival Guide

🧠 Free Recall ❓ Quiz 🔍 Cues 🔮 Simulation 🥷 ShadowBox 📐 Encode Cards

Service Goals

Throughput

To optimize for throughput, the producers and consumers need to move as much data as they can within a given amount of time.

For high throughput, you are trying to maximize the rate at which this data moves. This data rate should be as fast as possible.

A topic partition is the unit of parallelism in Kafka, and messages to different partitions can be sent in parallel by producers, written in parallel by different brokers, and read in parallel by different consumers.

In general, a higher number of topic partitions results in higher throughput, and to maximize throughput, you want enough partitions to distribute them across the brokers in your Confluent Cloud cluster.

Next, let’s discuss the batching strategy of Kafka producers. Producers can batch messages going to the same partition, which means they collect multiple messages to send together in a single request.

The most important step you can take to optimize throughput is to tune the producer batching to increase the batch size and the time spent waiting for the batch to fill up with messages. Larger batch sizes result in fewer requests to Confluent Cloud, which reduces the load on producers as well as the broker CPU overhead to process each request.

With the Java client, you can configure the batch.size parameter to increase the maximum size in bytes of each message batch. To give more time for batches to fill, you can configure the linger.ms parameter to have the producer wait longer before sending. The delay allows the producer to wait for the batch to reach the configured batch.size. The trade-off is tolerating higher latency since messages are not sent as soon as they are ready to send.

You can also easily enable compression, which means a lot of bits can be sent as fewer bits. Enable compression by configuring the compression.type parameter, which can be set to one of the following standard compression codecs: lz4, snappy, zstd, and gzip.

For performance, we generally recommend using lz4. We strongly recommend not using gzip because it’s much more compute intensive relative to the other codecs, so your application may not perform as well. Compression is applied on full batches of data, so better batching results in better compression ratios. When Confluent Cloud receives a compressed batch of messages from a producer, it always decompresses the data in order to validate it. Afterwards, it considers the compression codec of the destination topic.
If the compression codec of the destination topic are left at the default setting of producer, or if the codecs of the batch and destination topic are the same, Confluent Cloud takes the compressed batch from the client and writes it directly to the topic’s log file without taking cycles to recompress the data
Otherwise, Confluent Cloud needs to recompress the data to match the codec of the destination topic, and this can result in a performance impact; therefore, keep the compression codecs the same if possible

When a producer sends a message to Confluent Cloud, the message is sent to the leader broker for the target partition. Then the producer awaits a response from the leader broker (assuming acks is not set to 0, in which case the producer will not wait for any acknowledgment from the broker at all) to know that its message has been committed before proceeding to send the next messages.

There are automatic checks in place to make sure consumers cannot read messages that haven’t been committed yet. When leader brokers send those responses, it may impact the producer throughput: the sooner a producer receives a response, the sooner the producer can send the next message, which generally results in higher throughput.

So producers can set the configuration parameter acks to specify the number of acknowledgments the leader broker must have received before responding to the producer with an acknowledgment. Setting acks=1 makes the leader broker write the record to its local log and then acknowledge the request without awaiting acknowledgment from all followers.

The trade-off is you have to tolerate lower durability because the producer does not have to wait until the message is replicated to other brokers.

Kafka producers automatically allocate memory for the Java client to store unsent messages. If that memory limit is reached, then the producer will block on additional sends until memory frees up or until max.block.ms time passes.

You can adjust how much memory is allocated with the configuration parameter buffer.memory. If you don’t have a lot of partitions, you may not need to adjust this at all. However, if you have a lot of partitions, you can tune buffer.memory—while also taking into account the message size, linger time, and partition count—to maintain pipelines across more partitions. This in turn enables better utilization of the bandwidth across more brokers.

Likewise, you can tune consumers for higher throughput by adjusting how much data they get from each fetch from the leader broker in Confluent Cloud. You can increase how much data the consumers get from the leader for each fetch request by increasing the configuration parameter fetch.min.bytes. This parameter sets the minimum number of bytes expected for a fetch response from a consumer. Increasing this will also reduce the number of fetch requests made to Confluent Cloud, reducing the broker CPU overhead to process each fetch, thereby also improving throughput. Similar to the consequence of increasing batching on the producer, there may be a resulting trade-off to higher latency when increasing this parameter on the consumer. This is because the broker won’t send the consumer new messages until the fetch request has enough messages to fulfill the size of the fetch request, i.e., fetch.min.bytes, or until the expiration of the wait time, i.e., configuration parameter fetch.max.wait.ms.

Assuming the application allows it, use consumer groups with multiple consumers to parallelize consumption. Parallelizing consumption may improve throughput because multiple consumers can balance the load, processing multiple partitions simultaneously. The upper limit on this parallelization is the number of partitions in the topic.

Summary of Configurations for Optimizing Throughput

Producer:
batch.size: increase to 100000–200000 (default 16384)
linger.ms: increase to 10–100 (default 0)
compression.type=lz4 (default none, i.e., no compression)
acks=1 (default 1)
buffer.memory: increase if there are a lot of partitions (default 33554432)

Consumer:
fetch.min.bytes: increase to ~100000 (default 1)
Use consumer groups with multiple partitions

Latency

Confluent has guidelines on how to choose the number of partitions. Because a partition is a unit of parallelism in Kafka, an increased number of partitions may increase throughput. However, there is a trade-off in that an increased number of partitions may also increase latency. It may take longer to replicate a lot of partitions shared between each pair of brokers and consequently take longer for messages to be considered committed. No message can be consumed until it is committed, so this can ultimately increase end-to-end latency.

Producers automatically batch messages, which means they collect messages to send together. The less time that is given waiting for those batches to fill, then generally there is less latency producing data to Confluent Cloud. By default, the producer is tuned for low latency and the configuration parameter linger.ms is set to 0, which means the producer will send as soon as it has data to send. In this case, it is not true that batching is disabled—messages are always sent in batches—but sometimes a batch may have only one message (unless messages are passed to the producer faster than it can send them).

Consider whether you need to enable compression. Enabling compression typically requires more CPU cycles to do the compression, but it reduces network bandwidth utilization. So disabling compression typically spares the CPU cycles but increases network bandwidth utilization. Depending on the compression performance, you may consider leaving compression disabled with compression.type=none to spare the CPU cycles, although a good compression codec may potentially reduce latency as well.

You can tune the number of acknowledgments the producer requires the leader broker in the Confluent Cloud cluster to have received before considering a request complete. (Note that this acknowledgment to the producer differs from when a message is considered committed—more on that in the next section.) The sooner the leader broker responds, the sooner the producer can continue sending the next batch of messages, thereby generally reducing producer latency. Set the number of required acknowledgments with the producer acks configuration parameter. By default, acks=1, which means the leader broker will respond sooner to the producer before all replicas have received the message. Depending on your application requirements, you can even set acks=0 so that the producer will not wait for a response for a producer request from the broker, but then messages can potentially be lost without the producer even knowing.

Similar to the batching concept on the producers, you can tune consumers for lower latency by adjusting how much data it gets from each fetch from the leader broker in Confluent Cloud. By default, the consumer configuration parameter fetch.min.bytes is set to 1, which means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive, i.e., the configuration parameter fetch.max.wait.ms. Looking at these two configuration parameters together lets you reason through the size of the fetch request, i.e., fetch.min.bytes, or the age of a fetch request, i.e., fetch.max.wait.ms.
If you have a Kafka event streaming application or are using ksqlDB, there are also some performance enhancements you can make within the application. For scenarios where you need to perform table lookups at very large scale and with a low processing latency, you can use local stream processing. A popular pattern is to use Kafka Connect to make remote databases available local to Kafka. Then you can leverage the Kafka Streams API or ksqlDB to perform very fast and efficient local joins of such tables and streams, rather than requiring the application to make a query to a remote database over the network for each record. You can track the latest state of each table in a local state store, thus greatly reducing the processing latency as well as reducing the load of the remote databases when doing such streaming joins.

Kafka Streams applications are founded on processor topologies, a graph of stream processor nodes that can act on partitioned data for parallel processing. Depending on the application, there may be conservative but unnecessary data shuffling based on repartition topics, which would not result in any correctness issues but can introduce

performance penalties. To avoid performance penalties, you may enable topology optimizations for your event streaming applications by setting the configuration parameter topology.optimization. Enabling topology optimizations may reduce the amount of reshuffled streams that are stored and piped via repartition topics.

Producer:
linger.ms=0 (default 0)
compression.type=none (default none, i.e., no compression)
acks=1 (default 1)
Consumer:
fetch.min.bytes=1 (default 1)
Streams:
StreamsConfig.TOPOLOGY_OPTIMIZATION: StreamsConfig.OPTIMIZE (defaultStreamsConfig.NO_OPTIMIZATION)
Streams applications have embedded producers and consumers, so also check those configuration recommendations

Durability

Durability is all about reducing the chance for a message to get lost. Confluent Cloud enforces a replication factor of 3 to ensure data durability.

Producers can control the durability of messages written to Kafka through the acks configuration parameter. This parameter was discussed in the context of throughput and latency optimization, but it is primarily used in the context of durability. To optimize for high durability, we recommend setting it to acks=all (equivalent to acks=-1), which means the leader will wait for the full set of in-sync replicas (ISRs) to acknowledge the message and to consider it committed. This provides the strongest available guarantees that the record will not be lost as long as at least one in-sync replica remains alive. The trade-off is tolerating a higher latency because the leader broker waits for acknowledgments from replicas before responding to the producer.
Producers can also increase durability by trying to resend messages if any sends fail to ensure that data is not lost. The producer automatically tries to resend messages up to the number of times specified by the configuration parameter retries (default MAX_INT) and up to the time duration specified by the configuration parameter delivery.timeout.ms (default 120000), the latter of which was introduced in KIP-91. You can tune delivery.timeout.ms to the desired upper bound for the total time between sending a message and receiving an acknowledgment from the broker, which should reflect business requirements of how long a message is valid for.

There are two things to take into consideration with these automatic producer retries: duplication and message ordering.

1. Duplication: if there are transient failures in Confluent Cloud that cause a producer retry, the producer may send duplicate messages to Confluent Cloud
2. Ordering: multiple send attempts may be “in flight” at the same time, and a retry of a previously failed message send may occur after a newer message send succeeded

To address both of these, we generally recommend that you configure the producer for idempotency, i.e., enable.idempotence=true, for which the brokers in Confluent Cloud track messages using incrementing sequence numbers, similar to TCP. Idempotent producers can handle duplicate messages and preserve message order even with request pipelining—there is no message duplication because the broker ignores duplicate sequence numbers, and message ordering is preserved because when there are failures, the producer temporarily constrains to a single message in flight until sequencing is restored. In case the idempotence guarantees can’t be satisfied, the producer will raise a fatal error and reject any further sends, so when configuring the producer for idempotency, the application developer needs to catch the fatal error and handle it appropriately.

However, if you do not configure the producer for idempotency but the business requirements call for it, you need to address the potential for message duplication and ordering issues in other ways. To handle possible message duplication if there are transient failures in Confluent Cloud, be sure to build your consumer application logic to process duplicate messages. To preserve message order while also allowing resending failed messages, set the configuration parameter max.in.flight.requests.per.connection=1 to ensure that only one request can be sent to the broker at a time. To preserve message order while allowing request pipelining, set the configuration parameter retries=0 if the application is able to tolerate some message loss.

Instead of letting the producer automatically retry sending failed messages, you may prefer to manually code the actions for exceptions returned to the producer client, e.g., the onCompletion() method in the Callback interface in the Java client. If you want manual retry handling, disable automatic retries by setting retries=0. Note that producer idempotency tracks message sequence numbers, which makes sense only when automatic retries are enabled. Otherwise, if you set retries=0 and the application manually tries to resend a failed message, then it just generates a new sequence number so the duplication detection won’t work. Disabling automatic retries can result in message gaps due to individual send failures, but the broker will preserve the order of writes it receives.
Confluent Cloud provides durability by replicating data across multiple brokers. Each partition will have a list of assigned replicas (i.e., brokers) that should have copies the data. The list of replicas that are caught up to the leader are called in-sync replicas (ISRs). For each partition, leader brokers will automatically replicate messages to other brokers that are in their ISR list. When a producer sets acks=all (or acks=-1), then the configuration parameter min.insync.replicas specifies the minimum threshold for the replica count in the ISR list. If this minimum count cannot be met, then the producer will raise an exception. When used together, min.insync.replicas and acks allow you to enforce greater durability guarantees. A typical scenario would be to create a topic with replication.factor=3, topic configuration override min.insync.replicas=2, and producer acks=all, thereby ensuring that the producer raises an exception if a majority of replicas do not receive a write.
You also need to consider what happens to messages if there is an unexpected consumer failure to ensure that no messages are lost as they are being processed. Consumer offsets track which messages have already been consumed, so how and when consumers commit message offsets are crucial for durability. You want to avoid a situation where a consumer commits the offset of a message, starts processing that message, and then unexpectedly fails. This is because the subsequent consumer that starts reading from the same partition will not reprocess messages with offsets that have already been committed.

By default, offsets are configured to be automatically committed during the consumer’s poll() call at a periodic interval, and this is typically good enough for most use cases. But if the consumer is part of a transactional chain and you need strong message delivery guarantees, you may want the offsets to be committed only after the consumer finishes completely processing the messages. You can configure whether these consumer commits happen automatically or manually with the configuration parameter enable.auto.commit. For extra durability, you may disable the automatic commit by setting enable.auto.commit=false and explicitly call one of the commit methods in the consumer code (e.g., commitSync() or commitAsync()).
For even stronger guarantees, you may configure your applications for EOS transactions, which enable atomic writes to multiple Kafka topics and partitions. Since some messages in the log may be in various states of a transaction, consumers can set the configuration parameter isolation.level to define the types of messages they should receive. By setting isolation.level=read_committed, consumers will receive only non-transactional messages or committed transactional messages, and they will not receive messages from open or aborted transactions. To use transactional semantics in a consume-process-produce pattern and ensure each message is processed exactly once, a client application should set enable.auto.commit=false and should not commit offsets manually, instead using the sendOffsetsToTransaction() method in the KafkaProducer interface. You may also enable exactly once for your event streaming applications by setting the configuration parameter processing.guarantee.

There just so happens to be one such configuration: min.insync.replicas. The min.insync.replicas configuration enforces the number of replicas that must be in sync for the write to proceed. Note that the min.insync.replicas configuration is set at the broker or topic level and is not a producer configuration. The default value for min.insync.replicas is one. So to avoid the scenario described above, in a three-broker cluster, you’d want to increase the value to two.

If the number of replicas that are in sync is below the configured amount, the lead broker won’t attempt to append the record to its log. The leader throws either a NotEnoughReplicasException or NotEnoughReplicasAfterAppendException, forcing the producer to retry the write. Having replicas out of sync with the leader is considered a retryable error, so the producer will continue to retry and send the records up to the configured delivery timeout.

So by setting the min.insync.replicas and producer acks configurations to work together in this way, you’ve increased the durability of your data.

Summary of Configurations for Optimizing Durability

Producer:
replication.factor=3
acks=all (default 1)
enable.idempotence=true (default false), to prevent duplicate messages and out-of-order messages
max.in.flight.requests.per.connection=1 (default 5), to prevent out of order messages when not using an idempotent producer

Consumer:
enable.auto.commit=false (default true)
isolation.level=read_committed (when using EOS transactions) Streams:
StreamsConfig.REPLICATION_FACTOR_CONFIG: 3 (default 1)
StreamsConfig.PROCESSING_GUARANTEE_CONFIG: StreamsConfig.EXACTLY_ONCE (default StreamsConfig.AT_LEAST_ONCE)
Streams applications have embedded producers and consumers, so also check
those configuration recommendations

Availability

To optimize for high availability, you should tune your Kafka application to recover as quickly as possible from failure scenarios. 
When a producer sets acks=all (or acks=-1), the configuration parameter

min.insync.replicas specifies the minimum number of replicas that must acknowledge a write for the write to be considered successful. If this minimum cannot be met, then the producer will raise an exception. In the case of a shrinking ISR, the higher this minimum value is, the more likely there is to be a failure on producer send, which decreases availability for the partition. On the other hand, by setting this value low (e.g., min.insync.replicas=1), the system will tolerate more replica failures. As long as the minimum number of replicas is met, the producer requests will continue to succeed, which increases availability for the partition.
On the consumer side, consumers can share processing load by being a part of a consumer group. If a consumer unexpectedly fails, Kafka can detect the failure and rebalance the partitions amongst the remaining consumers in the consumer group. The consumer failures can be hard failures (e.g., SIGKILL) or soft failures (e.g., expired session timeouts), and they can be detected either when consumers fail to send heartbeats or when they fail to send poll() calls. The consumer liveness is maintained with a heartbeat, now in a background thread since KIP-62, and the configuration parameter session.timeout.ms dictates the timeout used to detect failed heartbeats. Increase the session timeout to take into account potential network delays and to avoid soft failures. Soft failures occur most commonly in two scenarios: when a batch of messages returned by poll() takes too long to process or when a JVM GC pause takes too long. If you have a poll() loop that spends too much time processing messages, you can address this either by increasing the upper bound on the amount of time that a consumer can be idle before fetching more records with max.poll.interval.ms or by reducing the maximum size of batches returned with the configuration parameter max.poll.records. Although higher session timeouts increase the time to detect and recover from a consumer failure, relatively speaking, incidents of failed clients are less likely than network issues.

Finally, when rebalancing workloads by moving tasks between event streaming application instances, you can reduce the time it takes to restore task processing state before the application instance resumes processing. In Kafka Streams, state restoration is usually done by replaying the corresponding changelog topic to reconstruct the state store. The application can replicate local state stores to minimize 
changelog-based restoration time by setting the configuration parameter num.standby.replicas. Thus, when a stream task is initialized or reinitialized on the application instance, its state store is restored to the most recent snapshot accordingly: 
If a local state store does not exist, i.e., num.standby.replicas=0, then the changelog is replayed from the earliest offset.
If a local state store does exist, i.e., num.standby.replicas is greater than 0, then the changelog is replayed from the previously checkpointed offset. This method takes less time because it is applying a smaller portion of the changelog.

Summary of Configurations for Optimizing Availability 
Consumer:
• session.timeout.ms: increase (default 10000)
Streams:
StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG: 1 or more (default 0)
Streams applications have embedded producers and consumers, so also check
those configuration recommendations

Communication

How/When/What does Admin Communicate

How/When/What does Producers Communicate

A Kafka client communicates with the Kafka brokers via the network for writing (or reading) events.

Kafka uses a binary protocol over TCP. The protocol defines all apis as request response message pairs. All messages are size delimited and are made up of the following primitive types.

The client initiates a socket connection and then writes a sequence of request messages and reads back the corresponding response message. No handshake is required on connection or disconnection. TCP is happier if you maintain persistent connections used for many requests to amortize the cost of the TCP handshake, but beyond this penalty connecting is pretty cheap.

The client will likely need to maintain a connection to multiple brokers, as data is partitioned and the clients will need to talk to the server that has their data.

The server guarantees that on a single TCP connection, requests will be processed in the order they are sent and responses will return in that order as well. The broker's request processing allows only a single in-flight request per connection in order to guarantee this ordering. Note that clients can (and ideally should) use non-blocking IO to implement request pipelining and achieve higher throughput. i.e., clients can send requests even while awaiting responses for preceding requests since the outstanding requests will be buffered in the underlying OS socket buffer. All requests are initiated by the client, and result in a corresponding response message from the server except where noted.

Metadata requests is the only request that can be addressed to any broker in the cluster.

Producers/ consumers request metadata from one of the brokers ( as each one of them caches it) and that is how they know who is the leader for a partition.

You spin up some brokers, let's say 3, then create some topic foo with 2 partitions, replication factor 2. Quite simple example, yet could be a real case for someone.
You create a producer with metadata.broker.list (or bootstrap.servers in new producer) configured to these brokers. Worth mentioning, you don't necessarily have to specify all the brokers in your cluster, in fact you can specify only 1 of them and it will still work. I'll explain this in a bit too.
You send a message to topic foo using your producer.
The producer looks up its local metadata cache to see what brokers are leaders for each partition of topic foo and how many partitions does your foo topic have. As this is the first send to the producer, local cache contains nothing.
Producer sends a TopicMetadataRequest to each broker in metadata.broker.list sequentially until first successful response. That's why I mentioned 1 broker in that list would work as long as it's alive.
Returned TopicMetadataResponse will contain the information about requested topics, in your case it's foo and brokers in the cluster. Basically, this response contains the following:
list of brokers in the cluster, where each broker has an ID, host and port. This list may not contain the entire list of brokers in the cluster, but should contain at least the list of brokers that are responsible for servicing the subject topic.
list of topic metadata, where each entry has topic name, number of partitions, leader broker ID for each partition and ISR broker IDs for each partition.
Based on TopicMetadataResponse your producer builds up its local cache and now knows exactly that the request for topic foo partition 0 should go to broker X.
Based on number of partitions in a topic, producer partitions your message and accumulates it with the knowledge that it should be sent as a part of batch to some broker.
When the batch is full or linger.ms timeout passes, your producer flushes the batch to the broker. By "flushes" I mean "opens a new connection to a broker or reuses an existing one, and sends the ProduceRequest".

The producer does not need to open unnecessary connections to all brokers, as the topic you are producing to may not be serviced by some brokers, and your cluster could be quite large. Imagine a 1000 broker cluster with lots of topics, but one of topics has just one partition - you only need that one connection, not 1000.

Metadata cache is updated in 2 cases:
If producer fails to communicate with broker for any reason - this includes the case when the broker is not reachable at all and when broker responds with an error (like "I'm not leader for this partition anymore, go away")
If no failures happen, the client still refreshes metadata every metadata.max.age.ms (https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java#L42-L43) to discover new brokers and partitions itself

With the new consumer API in Kafka 0.9 neither of the clients need to know about Zookeeper anymore and this seems to be the future for Kafka clients. Apart from getting rid of dependencies and flaky connection libraries to ZK, it also makes evolution of the client protocol easier since this is now completely managed by Kafka.

Communication Errors

The server will reject requests with a version it does not support, and will always respond to the client with exactly the protocol format it expects based on the version it included in its request. The intended upgrade path is that new features would first be rolled out on the server (with the older clients not making use of them) and then as newer clients are deployed these new features would gradually be taken advantage of.

all consumers and producers operating on that partition must connect to the leader.

(Page 8).

How/When/What does Consumers Communicate

Consumers send out health checks so that the broker knows if the consumers are healthy. Brokers' health is controlled by Controller, which is a Kafka service that runs on every broker in a Kafka cluster, but only one can be active (elected) at any point in time.

Kafka consumers use a pull model to consume data. This means that a consumer periodically sends a request to a Kafka broker in order to fetch data from it. This is called a FetchRequest. The consumer application then receives a number of records in a response, processes them and asks for more in the form of another request.

Consumers typically perform multiple fetches in parallel internally and accumulate data. The user receives the records in their application via the KafkaConsumer#poll() method call.

Just before exiting the poll() call, the consumer sends out another round of FetchRequests so that new records can be fetched while the current ones are being processed by the application

Consumers store their progress inside a Kafka topic called __consumer_offsets. Since we only ever need the latest state for a given consumer, and not its past states, this remains a compacted topic. Its retention period is seven days by default and can be configured by the offsets.retention.minutes setting.

The special broker which manages the group on the broker side is called the coordinator. The coordinator handles heartbeats and assigns the leader. Every group has a coordinator that organizes the startup of a consumer group and assist whenever a consumer leaves the group.

The leader consumer is elected by the coordinator. Its job is to assign partitions to every consumer in the group at startup or whenever a consumer leaves or joins the group. The leader holds the assignment strategy, it is decoupled from the broker. That means consumers can reconfigure the partition assignment strategy without restarting the brokers.

rebalance is a mechanism that attempts to evenly distribute (balance) the workload across all available members of a consumer group.

With the cooperative approach, consumers don’t automatically give up ownership of all topic partitions at the start of the rebalance. Instead, all members encode their current assignment and forward the information to the group leader. The group leader then determines which partitions need to change ownership—instead of producing an entirely new assignment from scratch.

Now a second rebalance is issued, but this time, only the topic partitions that need to change ownership are involved. It could be revoking topic partitions that are no longer assigned or adding new topic partitions. For the topic partitions that are in both the new and old assignment, nothing has to change, which means continued processing for topic partitions that aren’t moving.

The group rebalance protocol is a simple two-step protocol:

Collect membership, subscriptions and additional metadata for every consumer in the group
Decide on a new partition assignment and propagate it to every consumer in the group
During the first step, when a rebalance is in progress, every consumer does its part to join the group by sending a JoinGroup request to the group coordinator broker. This request holds information like the topics the consumer is interested in, its member ID and custom-defined metadata. The member ID is a unique identifier given to a consumer by the coordinator upon initially joining the group.

When the first JoinGroup request is received in the broker, the configured max.poll.interval.ms starts ticking . This is the window in which consumers have a chance to rejoin the group and also commit their offsets. During this window, a consumer’s heartbeats receive a REBALANCE_IN_PROGRESS error, informing the consumer to take action in committing its offsets and rejoining the group.

The way consumers maintain membership in a consumer group and ownership of the partitions assigned to them is by sending heartbeats to a Kafka broker designated as the group coordinator (this broker can be different for different consumer groups). As long as the consumer is sending heartbeats at regular intervals, it is assumed to be alive, well, and processing messages from its partitions. Heartbeats are sent when the consumer polls (i.e., retrieves records) and when it commits records it has consumed.

(Page 67).

When a consumer wants to join a group, it sends a JoinGroup request to the group coordinator. The first consumer to join the group becomes the group leader. The leader receives a list of all consumers in the group from the group coordinator (this will include all consumers that sent a heartbeat recently and which are therefore considered alive) and is responsible for assigning a subset of partitions to each consumer. It uses an implementation of Parti tionAssignor to decide which partitions should be handled by which consumer.

(Page 68).

After deciding on the partition assignment, the consumer leader sends the list of assignments to the GroupCoordinator, which sends this informa‐ tion to all the consumers. Each consumer only sees his own assign‐ ment—the leader is the only client process that has the full list of consumers in the group and their assignments. This process repeats every time a rebalance happens.

(Page 68).

Once we create a consumer, the next step is to subscribe to one or more topics. The subcribe() method takes a list of topics as a parameter, so it’s pretty simple to use:

(Page 69).

How/When/What does Brokers Communicate

Brokers know who is the leader for the given partition thanks to zk and one of them. To be more precise, one of them decides who will be a leader. That broker is called controller. The first broker that connects to zookeeper becomes a controller and his role is to decide which broker will be a leader and which ones will be replicas and to inform them about it. Controller itself is not excluded from this process. It is a broker like any other with this special responsibilities of choosing leaders and replicas

zk indeed detects when a broker dies/ leaves but it doesn't reelect leader. It is controller responsibility. When one of the brokers leaves a cluster, controller gets information from zk and it starts reassignment

brokers do communicate with each other ( replicas are reading the messages from leaders, controller is informing other brokers about changes), but they do not exchange metadata among themselves - they write metadata to a zookeeper

When setting up a Kafka broker, there are three essential broker configurations that you need to define. The broker needs to be able to know and tell other system elements who it is (broker.id). It needs to know where to write topic data (log.dirs). And it needs to connect to ZooKeeper (zookeeper.connect) in order to join its Kafka system as a productive member.

Within a cluster of brokers, one broker will also function as the cluster controller (elected automatically from the live members of the cluster). The controller is responsible for administrative operations, including assigning partitions to brokers and monitoring for broker failures.

(Page 8).

Every Kafka broker must have an integer identifier, which is set using the broker.id configuration. By default, this integer is set to 0, but it can be any value. The most important thing is that the integer must be unique within a single Kafka cluster. The selection of this number is arbitrary,

(Page 22).

The Kafka broker must decompress all message batches, however, in order to validate the checksum of the individual messages and assign offsets. It then needs to recompress the message batch in order to store it on disk. This is where the majority of Kafka’s requirement for processing power comes from.

(Page 30).

There are only two requirements in the broker configuration to allow multiple Kafka brokers to join a single cluster. The first is that all brokers must have the same config‐ uration for the zookeeper.connect parameter. This specifies the Zookeeper ensemble and path where the cluster stores metadata. The second requirement is that all brok‐ ers in the cluster must have a unique value for the broker.id parameter. If two brok‐ ers attempt to join the same cluster with the same broker.id, the second broker will log an error and fail to start.

(Page 32).

How/When/What does Zookeeper Communicate

ZK is a replicated fault tolerant storage system that the brokers use to save information and state for the cluster.

A Zookeeper cluster is called an ensemble. Due to the algorithm used, it is recom‐ mended that ensembles contain an odd number of servers (e.g., 3, 5, etc.) as a major‐ ity of ensemble members (a quorum) must be working in order for Zookeeper to respond to requests. This means that in a three-node ensemble, you can run with one node missing. With a five-node ensemble, you can run with two nodes missing.

(Page 19).

Consider running Zookeeper in a five-node ensemble. In order to make configuration changes to the ensemble, including swapping a node, you will need to reload nodes one at a time. If your ensemble cannot tolerate more than one node being down, doing mainte‐ nance work introduces additional risk. It is also not recommended to run more than seven nodes, as performance can start to degrade due to the nature of the consensus protocol.

(Page 19).

How/When/What does Schema Registry Communicate

Schema Registry lives outside of and separately from your Kafka brokers. Your producers and consumers still talk to Kafka to publish and read data (messages) to topics. Concurrently, they can also talk to Schema Registry to send and retrieve schemas that describe the data models for the messages.

How/When/What does REST Proxy Communicate

REST Proxy has two separate connections, one to a client and another to the broker. They are separate and distinct, and each of these connections requires a completely separate and different security configuration. You must configure security twice: once for the client-REST Proxy connection, and again for the REST Proxy-broker connection.

When a connection is initialized, there is a basic work request to REST Proxy. Such requests are plain text/JSON, and contain metadata and content to be stored in Kafka, or requests to retrieve data from Kafka. REST Proxy receives these requests, translates them into Kafka requests, and satisfies them with the Kafka broker.

No authorization actually occurs on REST Proxy (unlike in Schema Registry) so you must set broker ACLs to enforce any restrictions.

How/When/What does Streams Communicate

How/When/What does Connect Communicate

Connect allows you to use the REST API of any workers to create/update/delete a connector, however under the covers, it forwards the request to the "leader" worker.

Kafka is a distributed system, and one of the key things distributed systems need to do is deal with failures and disruptions—not just anticipate failures, but fully embrace them.

Unstarred

Dunk (exclude from study)

Apache Kafka Survival Guide

Review Questions

Todos

Sensemaker