This is the most impactful Apache Kafka mentor you’ve never heard of: Chia-Ping Tsai. In just 18 months, he bootstrapped a large Taiwanese open source community boasting: • 5000 participants • 15,000 Slack messages/month • 10 meetings/week • 20+ Apache committers

The 4 key reasons Kafka scales to insane throughput: - immutability (append-only) - active-passive model (single leader) - sharding (partitioning) - distributed system (horizontal scaling)

Something amazing is coming to Apache Kafka… Consumer Groups v2! If you’ve ever used consumer groups in production at any non-trivial scale, you probably know all the problems with it: - ⛔️ Group-wide synchronization barrier acts as a cap on scalability A single misbehaving consumer can disturb the whole group. Even if you have cooperative rebalancing and static membership enabled, you will still have rebalances happen. 🤷‍♂️ It’s a fact of life I've heard - death, taxes & consumer group rebalances. And the problem is that even with cooperative rebalancing (which helped a lot!), you’re bound on waiting for the slowest member of the group to complete the rebalance(s). 🐌 The problem is that no consumer can commit offsets while a rebalance is in progress. ❌ Another subtle thing is that with cooperative rebalancing, a rebalance will take longer than usual. Why? Because consumers are allowed to process partitions during rebalances. They will call the poll() method more infrequently -- they're busy processing records after all. Thus, the overall rebalance time will increase. This makes it pretty hard to scale to 1000s of members. - 🤯 Complexity There’s a reason you’re reading this! The current protocol is pretty complex and hard to understand. It's used for a bunch of stuff, including metadata propagation in Kafka Streams. This compexity results in more: - 🐛 bugs The harder to reason about and the more moving parts you have - the greater chance for bugs. There have been quite a few in the protocol. And due to the fact that a lot of the protocol’s logic lives on the clients, that results in: - 🐌 slow fixes Bugs require client-side fixes, which are slow to be adopted. If you run a Kafka ops team, you know how hard it is to get all of your clients' teams to upgrade! If you're using a cloud service, you need to wait for a new Kafka release to go out. Can't have the cloud provider handle it for you behind the scenes! - 🔍 hard to debug Debugging is harder because you need client logs. In the cloud, that's hard to do again. On-prem, it requires reading through a lot of logs and collecting a lot of files. - ⚙️ very extendable There’s a reusable embedded protocol within the rebalance protocol, where clients are free to attach raw bytes that only they can then parse themselves. It's challenging to build compatible software for this cross-client protocol, as well as near-impossible to inspect from the broker side. - 😢 inconsistent metadata Clients are responsible for triggering rebalances based on the metadata, but different clients can have different views of the metadata. - 😵‍💫 interoperability Different implementations of the clients (i.e. anything besides the Java client) may have bugs. The complex logic needs to be re-implemented quite a few times. This usually means more bugs and slower time to ship features in your favorite client. A combination nobody likes. ... So what should an open source community do? Move the logic to the broker! Then? Simplify it. The new protocol is very elegant - it streamlines all of the regular Kafka consumer logic inside a new heartbeat API. It has the broker decide what partition assignments the consumers should have, and totally omits the notion of a Group Leader client. Another major change is that the notion of a group-wide rebalance is removed now. The rebalance is more fine-grained now. 👌 When you think about it, a rebalance is simply a reassignment of some partitions from some consumers to others. 💡 Why does the whole group need to stop and know about this? It had to before because the logic was on the client. It doesn’t now. 🎂 The new protocol is fine-grained in its assignments. It maintains per-member epochs, as well as separate epochs for the general group membership and the global assignment. The goal is simple - get all of those epoch numbers to be the same. The order is the following: 1. the group-wide epoch is bumped. 2. the target assignment epoch is bumped. 3. individual consumers catch up to the epoch via the heartbeat request, individually. (fine-grained) In general, what you have is a simple state machine inside the Group Coordinator broker that’s running a constant reconciliation loop. 💥 Because every member converges to the target state independently, the coordinator is free to simplify that convergence member by member. 👍 It has the logic to resolve dependencies between members - the act of: 1. revoking one member’s partition. 2. confirming the success of that. 3. bumping that member's epoch before assigning that partition to another consumer. Here is an example visual of what happens when two members join a consumer group one by one:

Kafka Consumer Group Rebalance Visualized Its a dance consisting of two requests: 💃 1. JoinGroup 2. SyncGroup And once established, consumers regularly heartbeat to the group coordinator to signal they’re alive. The whole group goes through the two steps sequentially. ❤️ Phase 0 - Heartbeats / Steady State Let’s assume our group is already formed. In the steady state, every consumer is consuming peacefully and periodically sending heartbeats to the group coordinator in order to signal that it is alive and well. There are 6 reasons why a group can be forced to rebalance: • a consumer joins the group (sends a JoinGroup request). 👋 • a consumer shuts down gracefully - it leaves the group via a LeaveGroup request. ✌️ • max_poll_interval_ms passing between the poll calls. ⏱ • a consumer dies - its heartbeat requests time out after the configured session_timeout_ms. 💀 • the enforceRebalance API is called. 🏌️ • a new partition is added to a topic that the group is subscribed to. ✨ Let’s imagine our group needs to rebalance because a new member is joining. 👋 When the group is in a rebalance state, the coordinator begins to respond to the heartbeats with an error indicating that a rebalance is in progress. It’s through this response that it communicates the need for a rebalance to each consumer. When the consumer realizes it’s time to rebalance, it begins the dance. 💃 💃 Phase 1 - JoinGroup Every consumer first sends a JoinGroup to signal their intention of joining the group. This is done even they're already in the group - in that case, they simply reaffirm their membership and signal their intent to participate in the rebalance. The very first JoinGroup request the Coordinator receives is special: ⏱ - after it, the max poll interval timeout starts ticking... 👑 - the consumer that sent this first request is crowned the Group Leader! It will later be responsible for assigning partitions. The JoinGroup request sent by each consumer contains the protocol type and the extendable protocol-specific metadata. This is an embedded protocol within the rebalance protocol. For regular Kafka consumers - that’s most importantly the partition assignment that the consumer is currently assigned (if already in the group). The Group Coordinator broker responds to these JoinGroup requests when either: 🟣 - all known members have sent a request. 🟣 - max_poll_interval_ms passes. Once either of these is fulfilled, the Coordinator responds to all the JoinGroup requests it has received. The JoinGroup response sent to the Group Leader client is special - it contains all of the other JoinGroup request information sent by the other members. 🧳 In our example - the partition assignment of every consumer. 👍 The Group Leader then uses this information to craft a partition assignment for every consumer. 💃 Phase 2 - SyncGroup After a consumer processes the JoinGroup response, it immediately follows up with a SyncGroup request. The Group Coordinator will respond to these requests with the computed assignment for each consumer respectively. But. It has to wait and hold the requests in purgatory until it receives the assignment information. It is the Group Leader’s SyncGroup request that contains all the data (the computed partition assignment) that should be propagated to each member. 👑 When the broker receives the Group Leader’s SyncGroup request, it immediately persists this new group state to disk by appending a new record to the log of the `__consumer_offsets` partition it leads. 💿 Recall that each group is assigned a specific consumer offsets partition, and the leader of that partition is the Group Coordinator for that particular consumer group. Group. I thought I’d say that word again. 😁 Then, the Coordinator starts responding to the requests - propagating the particular consumer's assignment in its specific SyncGroup response. From a consumer’s point of view - the rebalance is finished once it receives a SyncGroup response. That frees it to resume consumption from the partitions it is assigned. The End🎬 Note that this post explained the eager rebalance protocol, which is the default in the consumer. The incremental cooperative rebalance protocol is a bit more interesting - and we will cover it with a video shortly too. ✌️

Read carefully: Understanding Kafka's consumer group rebalance requires focus! The protocol is very intricate. It was intentionally designed with clear separation of concerns: 🟣 the broker knows about group membership & subscribed topics 🟣 the consumers know about partition assignment Notably, the clients themselves choose how to distribute the load among themselves - they pass this information through an embedded protocol within the general rebalance. Design-wise, this makes sense because every application is different and it knows best how to split its work appropriately. 🍱 ✌️ The rebalance consists of two steps: 1. Collect or reaffirm group membership information. 2. Compute and propagate partition assignment. For the pictured example, imagine the following 3/3/3 scenario: We have 3 consumers consuming from a topic with 3 partitions, and 3 new partitions were created in the same topic. 3️⃣ 🎬 Phase 0 - Initiate the Rebalance? How do the clients know that a rebalance is taking place? There can be different triggers for a rebalance - here it is a notification of new partitions, but it can be new consumer clients coming online or leaving the group, etc. In any case - the same path is used. 📸 Not pictured here: every consumer sends lightweight heartbeat requests to the Group Coordinator broker. When the broker realizes that a rebalance should happen, it starts responding to the heartbeats with a special response that informs the clients. The consumers then start the rebalance dance. 🎬 Phase 1 - JoinGroup A JoinGroup request is sent to the Group Coordinator by every consumer. This informs the Group Coordinator broker of what consumer clients are part of the group. The very first JoinGroup request the Coordinator receives is special: ⏱ - after it, the max poll interval timeout starts ticking... 👑 - the consumer that sent this first request is crowned the Group Leader! The JoinGroup request sent by each consumer contains the protocol type and the protocol-specific metadata. This is the embedded protocol. For regular Kafka consumers - that’s usually the partition assignment that the consumer is assigned, among other things like rack id, consumer instance member id, and other things specific to the configured partition assignor. For Kafka Connect - there's other information like the URL and offset of the worker. The broker is agnostic to this information - it doesn’t parse it, it simply acts as a simple proxy and re-sends the same bytes. The Group Coordinator broker responds when either: 🟣 - all known members have sent a request 🟣 - max.poll.interval\.ms passes Once this is fulfilled, the Coordinator starts responding to all the JoinGroup requests it received so far. Note that this acts as a sort of synchronization barrier for the protocol. At this point of response, we know that no member of the group is consuming from any partition: 🟣 - if the member sent a request, it willfully stopped consumption. (note there is a different type of rebalance - the incremental cooperative one - that changes details here. we will cover this later ✌️) 🟣 - if the timeout passed, the member isn't part of the consumer group anymore. The JoinGroup response sent to the Group Leader client is special - it has all of the other JoinGroup request information sent by the other members. ⭐️🧳 The GroupLeader then uses this information to craft a partition assignment for every consumer, and the 2nd phase starts: 🎬 Phase 2 - SyncGroup A SyncGroup request is sent by each consumer immediately after they process the JoinGroup response. The Group Leader’s SyncGroup request sends all the data (the computed partition assignment) that should be sent to each member. 💿 When the broker receives that request, it immediately persists this new group state to disk by appending a new record to the `__consumer_offsets` partition it leads. Then, the Coordinator starts responding to the requests - propagating the particular consumer's assignment in its specific SyncGroup response. 🏁 The rebalance is then over -- For more concise #ApacheKafka content: 1. As they say in church - hit the 🔔 2. Check the 1-a-week 2-minute-read #Kafka newsletter in my profile (while hitting the 🔔) 3. 💜 like & RT to increment the Twitter algorithm counter?

What people think Kafka compaction is: - unique keys 🔑 What it actually is: - a config to denote how you delete old data and preserve existing 🤔 It takes a while to get used to them, so bear with me if some parts are unclear. A compacted topic is one where you have a more granular retention strategy, essentially saying “retain at least one record per key”. This allows you to eventually delete duplicate key records since your application probably doesn’t need them. Notably, it unlocks the ability to preserve at least one record per key, regardless of how old that record is. How does Kafka do this? It’s involved. The log is split into two abstract sections - the “cleaned” part (also called Log Tail) and the “dirty” part (also called Log Head). ✨ - Tail / Clean - the part that has been compacted. You have a guarantee that all records inside this section are unique per key. 💩 - Head / Dirty - the part that has not been compacted yet, meaning it contains duplicate keys. 🚨 Common Gotcha: Compaction does NOT guarantee any sort of unique key semantics in Kafka. Your consumer can read from any part of the log (dirty or clean), hence you may read duplicate keys. Compaction is above all a retention strategy - a config to denote how you preserve existing and delete old data. Nothing else. The compaction is done by the so-called Log Cleaner component. 🚨 Another gotcha: you can disable the log cleaner component via log_cleaner_enable=false. This is a guaranteed shot in the foot, as the consumer offsets internal topic (used by every consumer group) is compacted. If disabled, basically no compaction will happen. I don’t know why we allow disabling it… 🦶🔫 You can control how often the compaction happens by the min.cleanable.dirty.ratio config. If you set it to 0.3, you’re saying that you want to start compacting whenever more than 30% of your cleanable data is dirty. What's "cleanable" data? Well... you best follow me to find out in the next post :) 🔔

Zero copies. One copy. Two copies… Which one is better? Zero, of course. Zero-copy is an optimization that has to do with how we copy data in the OS. The problem is simple: Many applications, Kafka included, have the job of basically: 🔸1. reading files from the disk 🔸2. serving that data over the network as a response. When we get into the internals of it all, we see that the actual bytes get copied a bunch of times: 1. disk → OS read buffer 2. OS read buffer → application buffer 3. application buffer → socket buffer 4. socket buffer → NIC buffer Four copies! That being said, some of these copies are more efficient than others because they use the DMA engine. DMA stands for Direct Memory Access. It’s a feature in computers memory controllers which allows certain hardware (sound cards, graphics cards, network cards, etc.) to access the memory (RAM) without the CPU’s involvement. Additionally, we have a bunch of user <->kernel mode switches which aren’t free either: 1. user -> kernel mode 2. kernel -> user mode 3. user -> kernel mode 4. kernel -> user mode after response is written out So in total - 4 context switches and 4 copies of the data. Do we need all of this? 💡 Step 1: Disk -> OS Read Buffer. The OS read buffer may seem unnecessary here - we could just copy directly into the application. In a single isolated operation, the OS read buffer does not make sense - but throughout the continuous functioning of the system it shines. Why? Caching. Readahead caching is when the OS caches more parts of a file that you actually wanted to read. This is why linear reads on HDDs can be so efficient - the OS typically reads more than requested amount, and the rest is left in the cache. On the next read, you read from memory and not the disk, resulting in a big performance gain. 🤖 Translated into Kafka - the next fetch request will directly hit the cache and avoid accessing the disk. So let's keep that buffer. 🤝 As for the first context switch associated with it, it is fundamentally unavoidable. We will always need to go to kernel mode to read from the disk. 💡 Step 2: OS Read Buffer -> Application Buffer. If your application is not doing anything with the data, why do we need to copy the bytes there? Exactly. We don’t. This is what zero-copy is. Kafka stores its data in the same binary format it responds to requests with. Zero-copy, misleadingly named, is the act of NOT copying data back to the application but rather directly copying it into the NIC buffer. This avoids copies 2) and 3), and avoids the two context switches associated with them! This is a very efficient optimization, especially when you're reading more than what the OS read buffer can take. 🙈Imagine the worst case we had - reading multiples of what the disk read cache can store. This meant that: 💥1. the extra cached copy in the OS read buffer wouldn't get used on further reads (entirely useless copy) 💥2. we had to do steps 1) and 2) in a loop until the wohle data is read by the application. The new flow, thanks in part to scatter-gather operations, is such that: 1. the DMA engine reads from disk into the OS read buffer. 2. the CPU copies very small amounts - file descriptors (references) into the OS socket buffer. 3. scatter-gather operations have the DMA engine use the socket buffer references to read directly from the OS read buffer into the NIC buffer. Misleadingly named - it's not quite *zero* copies, but it's much less than we started with! Four context switches and four copies down to just two of each! It's a great optimization that has been benchmarked to be up to 65% more efficient in certain cases. I am not certain of its exact efficiency gains in Kafka in particular, as it depends a lot on the application that's being benchmarked.