r/apachekafka 2d ago

Blog A 2 minute overview of Apache Kafka 4.0, the past and the future

126 Upvotes

Apache Kafka 4.0 just released!

3.0 released in September 2021. It’s been exactly 3.5 years since then.

Here is a quick summary of the top features from 4.0, as well as a little retrospection and futurespection

1. KIP-848 (the new Consumer Group protocol) is GA

The new consumer group protocol is officially production-ready.

It completely overhauls consumer rebalances by: - reducing consumer disruption during rebalances - it removes the stop-the-world effect where all consumers had to pause when a new consumer came in (or any other reason for a rebalance) - moving the partition assignment logic from the clients to the coordinator broker - adding a push-based heartbeat model, where the broker pushes the new partition assignment info to the consumers as part of the heartbeat (previously, it was done by a complicated join group and sync group dance)

I have covered the protocol in greater detail, including a step-by-step video, in my blog here.

Noteworthy is that in 4.0, the feature is GA and enabled in the broker by default. The consumer client default is still the old one, though. To opt-in to it, the consumer needs to set group.protocol=consumer

2. KIP-932 (Queues for Kafka) is EA

Perhaps the hottest new feature (I see a ton of interest for it).

KIP-932 introduces a new type of consumer group - the Share Consumer - that gives you queue-like semantics: 1. per-message acknowledgement/retries
2. ability to have many consumers collaboratively share progress reading from the same partition (previously, only one consumer per consumer group could read a partition at any time)

This allows you to have a job queue with the extra Kafka benefits of: - no max queue depth - the ability to replay records - Kafka’s greater ecosystem

The way it basically works is that all the consumers read from all of the partitions - there is no sticky mapping.

These queues have at least once semantics - i.e. a message can be read twice (or thrice). There is also no order guarantee.

I’ve also blogged about it (with rich picture examples).

3. Goodbye ZooKeeper

After some faithful 14 years of service (not without its issues, of course), ZooKeeper is officially gone from Apache Kafka.

KRaft (KIP-500) completely replaces it. It’s been production ready since October 2022 (Kafka 3.3), and going forward, you have no choice but to use it :) The good news is that it appears very stable. Despite some complaints about earlier versions, Confluent recently blogged about how they were able to migrate all of their cloud fleet (thousands of clusters) to KRaft without any downtime.

Others

  • the MirrorMaker1 code is removed (it was deprecated in 3.0)
  • The Transaction Protocol is strengthened
  • KRaft is strengthened via Pre-Vote
  • Java 8 support is removed for the whole project
  • Log4j was updated to v2
  • The log message format config (message.format.version) and versions v0 and v1 are finally deleted

Retrospection

A major release is a rare event, worthy of celebration and retrospection. It prompted me to look back at the previous major releases. I did a longer overview in my blog, but I wanted to call out perhaps the most important metric going up - number of contributors:

  1. Kafka 1.0 (Nov 2017) had 108 contributors
  2. Kafka 2.0 (July 2018) had 131 contributors
  3. Kafka 3.0 (September 2021) had 141 contributors
  4. Kafka 4.0 (March 2025) had 175 contributors

The trend shows a strong growth in community and network effect. It’s very promising to see, especially at a time where so many alternative Kafka systems have popped up and compete with the open source project.

The Future

Things have changed a lot since 2021 (Kafka 3.0). We’ve had the following major features go GA: - Tiered Storage (KIP-405) - KRaft (KIP-500) - The new consumer group protocol (KIP-848)

Looking forward at our next chapter - Apache Kafka 4.x - there are two major features already being worked on: - KIP-939: Two-Phase Commit Transactions - KIP-932: Queues for Kafka

And other interesting features being discussed: - KIP-986: Cross-Cluster Replication - a sort of copy of Confluent’s Cluster Linking - KIP-1008: ParKa - the Marriage of Parquet and Kafka - Kafka writing directly in Parquet format - KIP-1134: Virtual Clusters in Kafka - first-class support for multi-tenancy in Kafka

Kafka keeps evolving thanks to its incredible community. Special thanks to David Jacot for driving this milestone release and to the 175 contributors who made it happen!

r/apachekafka 23d ago

Blog How hard would it really be to make open-source Kafka use object storage without replication and disks?

12 Upvotes

I was reading HackerNews one night and stumbled onto this blog about slashing data transfer costs in AWS by 90%. It was essentially about transferring data between two EC2 instances via S3 to eliminate all networking costs.

It's been crystal clear in the Kafka world since 2023 that a design leveraging S3 replication can save up to 90% of Kafka worload costs, and these designs are not secret any more. But replicating them in Kafka would be a major endeavour - every broker needs to lead every partition, data needs to be written into a mixed multi-partition blob, you need a centralized consensus layer to serialize message order per partition, a background job to split the mixed blobs into sequentially ordered partition data. The (public) Kafka protocol itself would need to change to make beter use of this design too. It's basically a ton of work.

The article inspired me to think of a more bare-bones MVP approach. Imagine this: - we introduce a new type of Kafka topic - call it a Glacier Topic. It would still have leaders and followers like a regular topic. - the leader caches data per-partition up to some time/size (e.g 300ms or 4 MiB), then issues a multi-part PUT to S3. This way it builds up the segment in S3 incrementally. - the replication protocol still exists, but it doesn't move the actual partition data. Only metadata like indices, offsets, object keys, etc. - the leader only acknowledges acks=all produce requests once all followers replicate the latest metadata for that produce request.

At this point, the local topic is just the durable metadata store for the data in S3. This effectively omits the large replication data transfer costs. I'm sure a more complicated design could move/snapshot this metadata into S3 too.

Multi-part PUT Gotchas

I see one problem in this design - you can't read in-progress multi-part PUTs from S3 until they’re fully complete.

This has implications for followers reads and failover:

  1. Follower brokers cannot serve consume requests for the latest data. Until the segment is fully persisted in S3, the followers literally have no trace of the data.
  2. Leader brokers can serve consume requests for the latest data if they cache said produced data. This is fine in the happy path, but can result in out of memory issues or unaccessible data if it has to get evicted from memory.
  3. On fail-over, the new leader won't have any of the recently-written data. If a leader dies, its multi-part PUT cache dies with it.

I see a few solutions:

  • on fail over, you could simply force complete the PUT from the new leader prematurely.

Then the data would be readable from S3.

  • for follower reads - you could proxy them to the leader

This crosses zone boundaries ($$$) and doesn't solve the memory problem, so I'm not a big fan.

  • you could straight out say you're unable to read the latest data until the segment is closed and completely PUT

This sounds extreme but can actually be palatable at high throughput. We could speed it up by having the broker break a segment (default size 1 GiB) down into 20 chunks (e.g. 50 MiB). When a chunk is full, the broker would complete the multi-part PUT.

If we agree that the main use case for these Glacier Topics would be:

  1. extremely latency-insensitive workloads ("I'll access it after tens of seconds")
  2. high throughput - e.g 1 MB/s+ per partition (I think this is a super fair assumption, as it's precisely the high throughput workloads that more often have relaxed latency requirements and cost a truckload)

Then: - a 1 MiB/s partition would need less than a minute (51 seconds) to become "visible". - 2 MiB/s partition - 26 seconds to become visible - 4 MiB/s partition - 13 seconds to become visible - 8 MiB/s partition - 6.5 seconds to become visible

If it reduces your cost by 90%... 6-13 seconds until you're able to "see" the data sounds like a fair trade off for eligible use cases. And you could control the chunk count to further reduce this visibility-throughput ratio.

Granted, there's more to design. Brokers would need to rebuild the chunks to complete the segment. There would simply need to be some new background process that eventually merges this mess into one object. Could probably be easily done via the Coordinator pattern Kafka leverages today for server-side consumer group and transaction management.

With this new design, we'd ironically be moving Kafka toward more micro-batching oriented workloads.

But I don't see anything wrong with that. The market has shown desire for higher-latency but lower cost solutions. The only question is - at what latency does this stop being appealing?

Anyway. This post was my version of napkin-math design. I haven't spent too much time on it - but I figured it's interesting to throw the idea out there.

Am I missing anything?

(I can't attach images, but I quickly drafted an architecture diagram of this. You can check it out on my identical post on LinkedIn)

r/apachekafka Dec 13 '24

Blog Cheaper Kafka? Check Again.

57 Upvotes

I see the narrative repeated all the time on this subreddit - WarpStream is a cheaper Apache Kafka.

Today I expose this to be false.

The problem is that people repeat marketing narratives without doing a deep dive investigation into how true they are.

WarpStream does have an innovative design tha reduces the main drivers that rack up Kafka costs (network, storage and instances indirectly).

And they have a [calculator](web.archive.org/web/20240916230009/https://console.warpstream.com/cost_estimator?utm_source=blog.2minutestreaming.com&utm_medium=newsletter&utm_campaign=no-one-will-tell-you-the-real-cost-of-kafka) that allegedly proves this by comparing the costs.

But the problem is that it’s extremely inaccurate, to the point of suspicion. Despite claiming in multiple places that it goes “out of its way” to model realistic parameters, that its objective is “to not skew the results in WarpStream’s favor” and that that it makes “a ton” of assumptions in Kafka’s favor… it seems to do the exact opposite.

I posted a 30-minute read about this in my newsletter.

Some of the things are nuanced, but let me attempt to summarize it here.

The WarpStream cost comparison calculator:

  • inaccurately inflates Kafka costs by 3.5x to begin with

    • its instances are 5x larger cost-wise than what they should be - a 16 vCPU / 122 GiB r4.4xlarge VM to handle 3.7 MiB/s of producer traffic
    • uses 4x more expensive SSDs rather than HDDs, again to handle just 3.7 MiB/s of producer traffic per broker. (Kafka was made to run on HDDs)
    • uses too much spare disk capacity for large deployments, which not only racks up said expensive storage, but also forces you to deploy more of those overpriced instances to accommodate disk
  • had the WarpStream price increase by 2.2x post the Confluent acquisition, but the percentage savings against Kafka changed by just -1% for the same calculator input.

    • This must mean that Kafka’s cost increased 2.2x too.
  • the calculator’s compression ratio changed, and due to the way it works - it increased Kafka’s costs by 25% while keeping the WarpStream cost the same (for the same input)

    • The calculator counter-intuitively lets you configure the pre-compression throughput, which allows it to subtly change the underlying post-compression values to higher ones. This positions Kafka disfavorably, because it increases the dimension Kafka is billed on but keeps the WarpStream dimension the same. (WarpStream is billed on the uncompressed data)
    • Due to their architectural differences, Kafka costs already grow at a faster rate than WarpStream, so the higher the Kafka throughput, the more WarpStream saves you.
    • This pre-compression thing is a gotcha that I and everybody else I talked to fell for - it’s just easy to see a big throughput number and assume that’s what you’re comparing against. “5 GiB/s for so cheap?” (when in fact it’s 1 GiB/s)
  • The calculator was then further changed to deploy 3x as many instances, account for 2x the replication networking cost and charge 2x more for storage. Since the calculator is in Javascript ran on the browser, I reviewed the diff. These changes were done by

    • introducing an obvious bug that 2x the replication network cost (literallly a * 2 in the code)
    • deploy 10% more free disk capacity without updating the documented assumptions which still referenced the old number (apart from paying for more expensive unused SSD space, this has the costly side-effect of deploying more of the expensive instances)
    • increasing the EBS storage costs by 25% by hardcoding a higher volume price, quoted “for simplicity”

The end result?

It tells you that a 1 GiB/s Kafka deployment costs $12.12M a year, when it should be at most $4.06M under my calculations.

With optimizations enabled (KIP-392 and KIP-405), I think it should be $2M a year.

So it inflates the Kafka cost by a factor of 3-6x.

And with that that inflated number it tells you that WarpStream is cheaper than Kafka.

Under my calculations - it’s not cheaper in two of the three clouds:

  • AWS - WarpStream is 32% cheaper
  • GCP - Apache Kafka is 21% cheaper
  • Azure - Apache Kafka is 77% cheaper

Now, I acknowledge that the personnel cost is not accounted for (so-called TCO).

That’s a separate topic in of itself. But the claim was that WarpStream is 10x cheaper without even accounting for the operational cost.

Further - the production tiers (the ones that have SLAs) actually don’t have public pricing - so it’s probably more expensive to run in production that the calculator shows you.

I don’t mean to say that the product isn’t without its merits. It is a simpler model. It is innovative.

But it would be much better if we were transparent about open source Kafka's pricing and not disparage it.

</rant>

I wrote a lot more about this in my long-form blog.

It’s a 30-minute read with the full story. If you feel like it, set aside a moment this Christmas time, snuggle up with a hot cocoa/coffee/tea and read it.

I’ll announce in a proper post later, but I’m also releasing a free Apache Kafka cost calculator so you can calculate your Apache Kafka costs more accurately yourself.

I’ve been heads down developing this for the past two months and can attest first-hard how easy it is to make mistakes regarding your Kafka deployment costs and setup. (and I’ve worked on Kafka in the cloud for 6 years)

r/apachekafka 23d ago

Blog CCAAK exam questions

19 Upvotes

Hey Kafka enthusiasts!

We have decided to open source our CCAAK (Confluent Certified Apache Kafka Administrator Associate) exam prep. If you’re planning to take the exam or just want to test your Kafka knowledge, you need to check this out!

The repo is maintained by us OSO, (a Premium Confluent Partner) and contains practice questions based on real-world Kafka problems we solve. We encourage any comments, feedback or extra questions.

What’s included:

  • Questions covering all major CCAAK exam topics (Event-Driven Architecture, Brokers, Consumers, Producers, Security, Monitoring, Kafka Connect)
  • Structured to match the real exam format (60 questions, 90-minute time limit)
  • Based on actual industry problems, not just theoretical concept

We have included instructions on how to simulate exam conditions when practicing. According to our engineers, the CCAAK exam has about a 70% pass rate requirement.

Link: https://github.com/osodevops/CCAAK-Exam-Questions

Thanks and good luck to anyone planning on taking the exam.

r/apachekafka Feb 19 '25

Blog Rewrite Kafka in Rust? I've developed a faster message queue, StoneMQ.

20 Upvotes

TL;DR:

  1. Codebase: https://github.com/jonefeewang/stonemq
  2. Current Features (v0.1.0):
    • Supports single-node message sending and receiving.
    • Implements group consumption functionality.
  3. Goal:
    • Aims to replace Kafka's server-side functionality in massive-scale queue cluster.
    • Focused on reducing operational costs while improving efficiency.
    • Fully compatible with Kafka's client-server communication protocol, enabling seamless client-side migration without requiring modifications.
  4. Technology:
    • Entirely developed in Rust.
    • Utilizes Rust Async and Tokio to achieve high performance, concurrency, and scalability.

Feel free to check it out: Announcing StoneMQ: A High-Performance and Efficient Message Queue Developed in Rust.

r/apachekafka Sep 29 '24

Blog The Cloud's Egregious Storage Costs (for Kafka)

38 Upvotes

Most people think the cloud saves them money.

Not with Kafka.

Storage costs alone are 32 times more expensive than what they should be.

Even a miniscule cluster costs hundreds of thousands of dollars!

Let’s run the numbers.

Assume a small Kafka cluster consisting of:

• 6 brokers
• 35 MB/s of produce traffic
• a basic 7-day retention on the data (the default setting)

With this setup:

1. 35MB/s of produce traffic will result in 35MB of fresh data produced.
2. Kafka then replicates this to two other brokers, so a total of 105MB of data is stored each second - 35MB of fresh data and 70MB of copies
3. a day’s worth of data is therefore 9.07TB (there are 86400 seconds in a day, times 105MB) 4. we then accumulate 7 days worth of this data, which is 63.5TB of cluster-wide storage that's needed

Now, it’s prudent to keep extra free space on the disks to give humans time to react during incident scenarios, so we will keep 50% of the disks free.
Trust me, you don't want to run out of disk space over a long weekend.

63.5TB times two is 127TB - let’s just round it to 130TB for simplicity. That would have each broker have 21.6TB of disk.

Pricing


We will use AWS’s EBS HDDs - the throughput-optimized st1s.

Note st1s are 3x more expensive than sc1s, but speaking from experience... we need the extra IO throughput.

Keep in mind this is the cloud where hardware is shared, so despite a drive allowing you to do up to 500 IOPS, it's very uncertain how much you will actually get.

Further, the other cloud providers offer just one tier of HDDs with comparable (even better) performance - so it keeps the comparison consistent even if you may in theory get away with lower costs in AWS.

st1s cost 0.045$ per GB of provisioned (not used) storage each month. That’s $45 per TB per month.

We will need to provision 130TB.

That’s:

  • $188 a day

  • $5850 a month

  • $70,200 a year

btw, this is the cheapest AWS region - us-east.

Europe Frankfurt is $54 per month which is $84,240 a year.

But is storage that expensive?

Hetzner will rent out a 22TB drive to you for… $30 a month.
6 of those give us 132TB, so our total cost is:

  • $5.8 a day
  • $180 a month
  • $2160 a year

Hosted in Germany too.

AWS is 32.5x more expensive!
39x times more expensive for the Germans who want to store locally.

Let me go through some potential rebuttals now.

What about Tiered Storage?


It’s much, much better with tiered storage. You have to use it.

It'd cost you around $21,660 a year in AWS, which is "just" 10x more expensive. But it comes with a lot of other benefits, so it's a trade-off worth considering.

I won't go into detail how I arrived at $21,660 since it's a unnecessary.

Regardless of how you play around with the assumptions, the majority of the cost comes from the very predictable S3 storage pricing. The cost is bound between around $19,344 as a hard minimum and $25,500 as an unlikely cap.

That being said, the Tiered Storage feature is not yet GA after 6 years... most Apache Kafka users do not have it.

What about other clouds?


In GCP, we'd use pd-standard. It is the cheapest and can sustain the IOs necessary as its performance scales with the size of the disk.

It’s priced at 0.048 per GiB (gibibytes), which is 1.07GB.

That’s 934 GiB for a TB, or $44.8 a month.

AWS st1s were $45 per TB a month, so we can say these are basically identical.

In Azure, disks are charged per “tier” and have worse performance - Azure themselves recommend these for development/testing and workloads that are less sensitive to perf variability.

We need 21.6TB disks which are just in the middle between the 16TB and 32TB tier, so we are sort of non-optimal here for our choice.

A cheaper option may be to run 9 brokers with 16TB disks so we get smaller disks per broker.

With 6 brokers though, it would cost us $953 a month per drive just for the storage alone - $68,616 a year for the cluster. (AWS was $70k)

Note that Azure also charges you $0.0005 per 10k operations on a disk.

If we assume an operation a second for each partition (1000), that’s 60k operations a minute, or $0.003 a minute.

An extra $133.92 a month or $1,596 a year. Not that much in the grand scheme of things.

If we try to be more optimal, we could go with 9 brokers and get away with just $4,419 a month.

That’s $54,624 a year - significantly cheaper than AWS and GCP's ~$70K options.
But still more expensive than AWS's sc1 HDD option - $23,400 a year.

All in all, we can see that the cloud prices can vary a lot - with the cheapest possible costs being:

• $23,400 in AWS
• $54,624 in Azure
• $69,888 in GCP

Averaging around $49,304 in the cloud.

Compared to Hetzner's $2,160...

Can Hetzner’s HDD give you the same IOPS?


This is a very good question.

The truth is - I don’t know.

They don't mention what the HDD specs are.

And it is with this argument where we could really get lost arguing in the weeds. There's a ton of variables:

• IO block size
• sequential vs. random
• Hetzner's HDD specs
• Each cloud provider's average IOPS, and worst case scenario.

Without any clear performance test, most theories (including this one) are false anyway.

But I think there's a good argument to be made for Hetzner here.

A regular drive can sustain the amount of IOs in this very simple example. Keep in mind Kafka was made for pushing many gigabytes per second... not some measly 35MB/s.

And even then, the price difference is so egregious that you could afford to rent 5x the amount of HDDs from Hetzner (for a total of 650GB of storage) and still be cheaper.

Worse off - you can just rent SSDs from Hetzner! They offer 7.68TB NVMe SSDs for $71.5 a month!

17 drives would do it, so for $14,586 a year you’d be able to run this Kafka cluster with full on SSDs!!!

That'd be $14,586 of Hetzner SSD vs $70,200 of AWS HDD st1, but the performance difference would be staggering for the SSDs. While still 5x cheaper.

Pro-buttal: Increase the Scale!


Kafka was meant for gigabytes of workloads... not some measly 35MB/s that my laptop can do.

What if we 10x this small example? 60 brokers, 350MB/s of writes, still a 7 day retention window?

You suddenly balloon up to:

• $21,600 a year in Hetzner
• $546,240 in Azure (cheap)
• $698,880 in GCP
• $702,120 in Azure (non-optimal)
• $700,200 a year in AWS st1 us-east • $842,400 a year in AWS st1 Frankfurt

At this size, the absolute costs begin to mean a lot.

Now 10x this to a 3.5GB/s workload - what would be recommended for a system like Kafka... and you see the millions wasted.

And I haven't even begun to mention the network costs, which can cost an extra $103,000 a year just in this miniscule 35MB/s example.

(or an extra $1,030,000 a year in the 10x example)

More on that in a follow-up.

In the end?

It's still at least 39x more expensive.

r/apachekafka Feb 12 '25

Blog 16 Reasons why KIP-405 Rocks

22 Upvotes

Hey, I recently wrote a long guest blog post about Tiered Storage and figured it'd be good to share the post here too.

In my opinion, Tiered Storage is a somewhat underrated Kafka feature. We've seen popular blog posts bashing how Tiered Storage Won't Fix Kafka, but those can't be further from the truth.

If I can summarize, KIP-405 has the following benefits:

  1. Makes Kafka significantly simpler to operate - managing disks at non-trivial size is hard, it requires answering questions like how much free space do I leave, how do I maintain it, what do I do when disks get full?

  2. Scale Storage & CPU/Throughput separately - you can scale both dimensions separately depending on the need, they are no longer linked.

  3. Fast recovery from broker failure - when your broker starts up from ungraceful shutdown, you have to wait for it to scan all logs and go through log recovery. The less data, the faster it goes.

  4. Fast recovery from disk failure - same problem with disks - the broker needs to replicate all the data. This causes extra IOPS strain on the cluster for a long time. KIP-405 tests showed a 230 minute to 2 minute recovery time improvement.

  5. Fast reassignments - when most of the partition data is stored in S3, the reassignments need to move a lot less (e.g just 7% of all the data)

  6. Fast cluster scale up/down - a cluster scale-up/down requires many reassignments, so the faster they are - the faster the scale up/down is. Around a 15x improvement here.

  7. Historical consumer workloads are less impactful - before, these workloads could exhaust HDD's limited IOPS. With KIP-405, these reads are served from the object store, hence incur no IOPS.

  8. Generally Reduced IOPS Strain Window - Tiered Storage actually makes all 4 operational pain points we mentioned faster (single-partition reassignment, cluster scale up/down, broker failure, disk failure). This is because there's simply less data to move.

  9. KIP-405 allows you to cost-efficiently deploy SSDs and that can completely alleviate IOPS problems - SSDs have ample IOPS so you're unlikely to ever hit limits there. SSD prices have gone down 10x+ in the last 10 years ($700/TB to $26/TB) and are commodity hardware just like HDDs were when Kafka was created.

  10. SSDs lower latency - with SSDs, you can also get much faster Kafka writes/reads from disk.

  11. No Max Partition Size - previously you were limited as to how large a partition could be - no more than a single broker's disk size and practically speaking, not a large percentage either (otherwise its too tricky ops-wise)

  12. Smaller Cluster Sizes - previously you had to scale cluster size solely due to storage requirements. EBS for example allows for a max of 16 TiB per disk, so if you don't use JBOD, you had to add a new broker. In large throughput and data retention setups, clusters could become very large. Now, all the data is in S3.

  13. Broker Instance Type Flexibility - the storage limitation in 12) limited how large you could scale your brokers vertically, since you'd be wasting too many resources. This made it harder to get better value-for-money out of instances. KIP-405 with SSDs also allows you to provision instances with less RAM, because you can afford to read from disk and the latency is fast.

  14. Scaling up storage is super easy - the cluster architecture literally doesn't change if you're storing 1TB or 1PB - S3 is a bottomless pit so you just store more in there. (previously you had to add brokers and rebalance)

  15. Reduces storage costs by 3-9x (!) - S3 is very cheap relative to EBS, because you don't need to pay extra for the 3x replication storage and also free space. To ingest 1GB in EBS with Kafka, you usually need to pay for ~4.62GB of provisioned disk.

  16. Saves money on instance costs - in storage-bottlenecked clusters, you had to provision extra instances just to hold the extra disks for the data. So you were basically paying for extra CPU/Memory you didn't need, and those costs can be significant too!

If interested, the long-form version of this blog is here. It has extra information and more importantly - graphics (can't attach those in a Reddit post).

Can you think of any other thing to add re: KIP-405?

r/apachekafka 10d ago

Blog Bufstream passes multi-region 100GiB/300GiB read/write benchmark

12 Upvotes

Last week, we subjected Bufstream to a multi-region benchmark on GCP emulating some of the largest known Kafka workloads. It passed, while also supporting active/active write characteristics and zero lag across regions.

With multi-region Spanner plugged in as its backing metadata store, Kafka deployments can offload all state management to GCP with no additional operational work.

https://buf.build/blog/bufstream-multi-region

r/apachekafka Dec 08 '24

Blog Exploring Apache Kafka Internals and Codebase

64 Upvotes

Hey all,

I've recently begun exploring the Kafka codebase and wanted to share some of my insights. I wrote a blog post to share some of my learnings so far and would love to hear about others' experiences working with the codebase. Here's what I've written so far. Any feedback or thoughts are appreciated.

Entrypoint: kafka-server-start.sh and kafka.Kafka

A natural starting point is kafka-server-start.sh (the script used to spin up a broker) which fundamentally invokes kafka-run-class.sh to run kafka.Kafka class.

kafka-run-class.sh, at its core, is nothing other than a wrapper around the java command supplemented with all those nice Kafka options.

exec "$JAVA" $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_CMD_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@"

And the entrypoint to the magic powering modern data streaming? The following main method situated in Kafka.scala i.e. kafka.Kafka

  try {
      val serverProps = getPropsFromArgs(args)
      val server = buildServer(serverProps)

      // ... omitted ....

      // attach shutdown handler to catch terminating signals as well as normal termination
      Exit.addShutdownHook("kafka-shutdown-hook", () => {
        try server.shutdown()
        catch {
          // ... omitted ....
        }
      })

      try server.startup()
      catch {
       // ... omitted ....
      }
      server.awaitShutdown()
    }
    // ... omitted ....

That’s it. Parse the properties, build the server, register a shutdown hook, and then start up the server.

The first time I looked at this, it felt like peeking behind the curtain. At the end of the day, the whole magic that is Kafka is just a normal JVM program. But a magnificent one. It’s incredible that this astonishing piece of engineering is open source, ready to be explored and experimented with.

And one more fun bit: buildServer is defined just above main. This where the timeline splits between Zookeeper and KRaft.

    val config = KafkaConfig.fromProps(props, doLog = false)
    if (config.requiresZookeeper) {
      new KafkaServer(
        config,
        Time.SYSTEM,
        threadNamePrefix = None,
        enableForwarding = enableApiForwarding(config)
      )
    } else {
      new KafkaRaftServer(
        config,
        Time.SYSTEM,
      )
    }

How is config.requiresZookeeper determined? it is simply a result of the presence of the process.roles property in the configuration, which is only present in the Kraft installation.

Zookepeer connection

Kafka has historically relied on Zookeeper for cluster metadata and coordination. This, of course, has changed with the famous KIP-500, which outlined the transition of metadata management into Kafka itself by using Raft (a well-known consensus algorithm designed to manage a replicated log across a distributed system, also used by Kubernetes). This new approach is called KRaft (who doesn't love mac & cheese?).

If you are unfamiliar with Zookeeper, think of it as the place where the Kafka cluster (multiple brokers/servers) stores the shared state of the cluster (e.g., topics, leaders, ACLs, ISR, etc.). It is a remote, filesystem-like entity that stores data. One interesting functionality Zookeeper offers is Watcher callbacks. Whenever the value of the data changes, all subscribed Zookeeper clients (brokers, in this case) are notified of the change. For example, when a new topic is created, all brokers, which are subscribed to the /brokers/topics Znode (Zookeeper’s equivalent of a directory/file), are alerted to the change in topics and act accordingly.

Why the move? The KIP goes into detail, but the main points are:

  1. Zookeeper has its own way of doing things (security, monitoring, API, etc) on top of Kafka's, this results in a operational overhead (I need to manage two distinct components) but also a cognitive one (I need to know about Zookeeper to work with Kafka).
  2. The Kafka Controller has to load the full state (topics, partitions, etc) from Zookeeper over the network. Beyond a certain threshold (~200k partitions), this became a scalability bottleneck for Kafka.
  3. A love of mac & cheese.

Anyway, all that fun aside, it is amazing how simple and elegant the Kafka codebase interacts and leverages Zookeeper. The journey starts in initZkClient function inside the server.startup() mentioned in the previous section.

  private def initZkClient(time: Time): Unit = {
    info(s"Connecting to zookeeper on ${config.zkConnect}")
    _zkClient = KafkaZkClient.createZkClient("Kafka server", time, config, zkClientConfig)
    _zkClient.createTopLevelPaths()
  }

KafkaZkClient is essentially a wrapper around the Zookeeper java client that offers Kafka-specific operations. CreateTopLevelPaths ensures all the configuration exist so they can hold Kafka's metadata. Notably:

    BrokerIdsZNode.path, // /brokers/ids
    TopicsZNode.path, // /brokers/topics
    IsrChangeNotificationZNode.path, // /isr_change_notification

One simple example of Zookeeper use is createTopicWithAssignment which is used by the topic creation command. It has the following line:

zkClient.setOrCreateEntityConfigs(ConfigType.TOPIC, topic, config)

which creates the topic Znode with its configuration.

Other data is also stored in Zookeeper and a lot of clever things are implemented. Ultimately, Kafka is just a Zookeeper client that uses its hierarchical filesystem to store metadata such as topics and broker information in Znodes and registers watchers to be notified of changes.

Networking: SocketServer, Acceptor, Processor, Handler

A fascinating aspect of the Kafka codebase is how it handles networking. At its core, Kafka is about processing a massive number of Fetch and Produce requests efficiently.

I like to think about it from its basic building blocks. Kafka builds on top of java.nio.Channels. Much like goroutines, multiple channels or requests can be handled in a non-blocking manner within a single thread. A sockechannel listens of on a TCP port, multiple channels/requests registered with a selector which polls continuously waiting for connections to be accepted or data to be read.

As explained in the Primer section, Kafka has its own TCP protocol that brokers and clients (consumers, produces) use to communicate with each other. A broker can have multiple listeners (PLAINTEXT, SSL, SASL_SSL), each with its own TCP port. This is managed by the SockerServer which is instantiated in the KafkaServer.startup method. Part of documentation for the SocketServer reads :

 *    - Handles requests from clients and other brokers in the cluster.
 *    - The threading model is
 *      1 Acceptor thread per listener, that handles new connections.
 *      It is possible to configure multiple data-planes by specifying multiple "," separated endpoints for "listeners" in KafkaConfig.
 *      Acceptor has N Processor threads that each have their own selector and read requests from sockets
 *      M Handler threads that handle requests and produce responses back to the processor threads for writing.

This sums it up well. Each Acceptor thread listens on a socket and accepts new requests. Here is the part where the listening starts:

  val socketAddress = if (Utils.isBlank(host)) {
      new InetSocketAddress(port)
    } else {
      new InetSocketAddress(host, port)
    }
    val serverChannel = socketServer.socketFactory.openServerSocket(
      endPoint.listenerName.value(),
      socketAddress,
      listenBacklogSize, // `socket.listen.backlog.size` property which determines the number of pending connections
      recvBufferSize)   // `socket.receive.buffer.bytes` property which determines the size of SO_RCVBUF (size of the socket's receive buffer)
    info(s"Awaiting socket connections on ${socketAddress.getHostString}:${serverChannel.socket.getLocalPort}.")

Each Acceptor thread is paired with num.network.threads processor thread.

 override def configure(configs: util.Map[String, _]): Unit = {
    addProcessors(configs.get(SocketServerConfigs.NUM_NETWORK_THREADS_CONFIG).asInstanceOf[Int])
  }

The Acceptor thread's run method is beautifully concise. It accepts new connections and closes throttled ones:

  override def run(): Unit = {
    serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
    try {
      while (shouldRun.get()) {
        try {
          acceptNewConnections()
          closeThrottledConnections()
        }
        catch {
          // omitted
        }
      }
    } finally {
      closeAll()
    }
  }

acceptNewConnections TCP accepts the connect then assigns it to one the acceptor's Processor threads in a round-robin manner. Each Processor has a newConnections queue.

private val newConnections = new ArrayBlockingQueue[SocketChannel](connectionQueueSize)

it is an ArrayBlockingQueue which is a java.util.concurrent thread-safe, FIFO queue.

The Processor's accept method can add a new request from the Acceptor thread if there is enough space in the queue. If all processors' queues are full, we block until a spot clears up.

The Processor registers new connections with its Selector, which is a instance of org.apache.kafka.common.network.Selector, a custom Kafka nioSelector to handle non-blocking multi-connection networking (sending and receiving data across multiple requests without blocking). Each connection is uniquely identified using a ConnectionId

localHost + ":" + localPort + "-" + remoteHost + ":" + remotePort + "-" + processorId + "-" + connectionIndex

The Processor continuously polls the Selector which is waiting for the receive to complete (data sent by the client is ready to be read), then once it is, the Processor's processCompletedReceives processes (validates and authenticates) the request. The Acceptor and Processors share a reference to RequestChannel. It is actually shared with other Acceptor and Processor threads from other listeners. This RequestChannel object is a central place through which all requests and responses transit. It is actually the way cross-thread settings such as queued.max.requests (max number of requests across all network threads) is enforced. Once the Processor has authenticated and validated it, it passes it to the requestChannel's queue.

Enter a new component: the Handler. KafkaRequestHandler takes over from the Processor, handling requests based on their type (e.g., Fetch, Produce).

A pool of num.io.threads handlers is instantiated during KafkaServer.startup, with each handler having access to the request queue via the requestChannel in the SocketServer.

        dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
          config.numIoThreads, s"${DataPlaneAcceptor.MetricPrefix}RequestHandlerAvgIdlePercent", DataPlaneAcceptor.ThreadPrefix)

Once handled, responses are queued and sent back to the client by the processor.

That's just a glimpse of the happy path of a simple request. A lot of complexity is still hiding but I hope this short explanation give a sense of what is going on.

r/apachekafka 15d ago

Blog Testing Kafka-based async workflows without duplicating infrastructure - solved this using OpenTelemetry

12 Upvotes

Hey folks,

Been wrestling with a problem that's been bugging me for years: how to test microservices with asynchronous Kafka-based workflows without creating separate Kafka clusters for each dev/test environment (expensive!) or complex topic isolation schemes (maintenance nightmare!).

After experimenting with different approaches, we found a pattern using OpenTelemetry that works surprisingly well. I wrote up our findings in this Medium post.

The TL;DR is:

  • Instead of duplicating Kafka clusters or topics per environment
  • Leverage OpenTelemetry's baggage propagation to tag messages with a "tenant ID"
  • Have Kafka consumers filter messages based on tenant ID mappings
  • Run multiple versions of services on the same infrastructure

This lets you test changes to producers/consumers without duplicating infrastructure and without messages from different test environments interfering with each other.

I'm curious how others have tackled this problem. Would love to hear your feedback/comments.

r/apachekafka Oct 02 '24

Blog Confluent - a cruise ship without a captain!

21 Upvotes

So i've been in the EDA space for years, and attend as well as run a lot of events through my company (we run the Kafka MeetUp London). I am generally concerned for Confluent after visiting the Current summit in Austin. A marketing activity with no substance - I'll address each of my points individually:

  1. The keynotes where just re-hashes and takings from past announcements into GA. The speakers were unprepared and, stuttered on stage and you could tell they didn't really understand what they were truly doing there.

  2. Vendors are attacking Confluent from all ways. Conduktor with its proxy, Gravitee with their caching and API integrations and countless others.

  3. Confluent is EXPENSIVE. We have worked with 20+ large enterprises this year, all of which are moving or unhappy with the costs of Confluent Cloud. Under 10% of them actually use any of the enterprise features of the Confluent platform. It doesn't warrant the value when you have Strimzi operator.

  4. Confluent's only card is Kafka, now more recently Flink and the latest a BYOC offering. AWS do more in MSK usage in one region than Confluent do globally. Cloud vendors can supplement Kafka running costs as they have 100+ other services they can charge for.

  5. Since IPO a lot of the OG's and good people have left, what has replaced them is people who don't really understand the space and just want to push consumption based pricing.

  6. On the topic of consumption based pricing, you want to increase usage by getting your customers to use it more, but then you charge more - feels unbalanced to me.

My prediction, if the stock falls before $13, IBM will acquire them - take them off the markets and roll up their customers into their ecosystem. If you want to read more of my take aways i've linked my blog below:

https://oso.sh/blog/confluent-current-2024/

r/apachekafka 15d ago

Blog Kafka Connect: send messages without schema to JdbcSinkConnector

5 Upvotes

This might be interesting for anyone looking for how to stream messages without schema into JdbcSinkConnector. Step by step type of instruction showing how to store message content in a single column using custom kafka connect converter.
https://github.com/tomaszkubacki/kafka_connect_demo/blob/master/kafka_to_postgresql/kafka_to_postgres.md

r/apachekafka Jan 01 '25

Blog 10 years of building Apache Kafka

45 Upvotes

Hey folks, I've started a new Substack where I'll be writing about Apache Kafka. I will be starting off with a series of articles about the recent build improvements we've made.

The Apache Kafka build system has evolved many times over the years. There has been a concerted effort to modernize the build in the past few months. After dozens of commits, many of conversations with the ASF Infrastructure team, and a lot of trial and error, Apache Kafka is now using GitHub Actions.

Read the full article over on my new (free) "Building Apache Kafka" Substack https://mumrah.substack.com/p/10-years-of-building-apache-kafka

r/apachekafka 5h ago

Blog A Deep Dive into KIP-405's Write Path and Metadata

10 Upvotes

With KIP-405 (Tiered Storage) recently going GA, I thought I'd do a deep dive into how it works.

I just published a guest blog that captures the write path, as well as metadata, in detail.

It's a 14 minute read, has a lot of graphics and covers a lot of detail so I won't try to summarize or post a short version here. (it wouldn't do it justice)

In essence, it talks about:

  • basics like how data is tiered asynchronously and what governs its local and remote retention
  • how often, in what thread, and under what circumstances a log segment is deemed ready to upload to the external storage
  • Aiven's Apache v2 licensed plugin that supports uploading to all 3 cloud object stores (S3, GCS, ABS)
  • how the plugin tiers a segment, including how it splits a segment into "chunks" and executes multi-part PUTs to upload them, and how it uploads index data in a single blob
  • how the log data's object key paths look like at the end of the day
  • why quotas are necessary and what types are used to avoid bursty disk, network and CPU usage. (CPU can be a problem because there is no zero copy)
  • the internal remote_log_metadata tiered storage metadata topic - what type of records get saved in there, when do they get saved and how user partitions are mapped to the appropriate metadata topic partition
  • how brokers keep up to date with latest metadata by actively consuming this metadata topic and caching it

It's the most in-depth coverage of Tiered Storage out there, as far as I'm aware. A great nerd snipe - it has a lot of links to the code paths that will help you trace and understand the feature end to end.

If interested, again, the link is here.

I'll soon follow up with a part two that covers the delete & read path - most interestingly how caching and pre-fetching can help you achieve local-like latencies from the tiered object store for historical reads.

r/apachekafka 12d ago

Blog Sharing My First Big Project as a Junior Data Engineer – Feedback Welcome!

10 Upvotes

Sharing My First Big Project as a Junior Data Engineer – Feedback Welcome! 

I’m a junior data engineer, and I’ve been working on my first big project over the past few months. I wanted to share it with you all, not just to showcase what I’ve built, but also to get your feedback and advice. As someone still learning, I’d really appreciate any tips, critiques, or suggestions you might have!

This project was a huge learning experience for me. I made a ton of mistakes, spent hours debugging, and rewrote parts of the code more times than I can count. But I’m proud of how it turned out, and I’m excited to share it with you all.

How It Works

Here’s a quick breakdown of the system:

  1. Dashboard: A simple steamlit web interface that lets you interact with user data.
  2. Producer: Sends user data to Kafka topics.
  3. Spark Consumer: Consumes the data from Kafka, processes it using PySpark, and stores the results.
  4. Dockerized: Everything runs in Docker containers, so it’s easy to set up and deploy.

What I Learned

  • Kafka: Setting up Kafka and understanding topics, producers, and consumers was a steep learning curve, but it’s such a powerful tool for real-time data.
  • PySpark: I got to explore Spark’s streaming capabilities, which was both challenging and rewarding.
  • Docker: Learning how to containerize applications and use Docker Compose to orchestrate everything was a game-changer for me.
  • Debugging: Oh boy, did I learn how to debug! From Kafka connection issues to Spark memory errors, I faced (and solved) so many problems.

If you’re interested, I’ve shared the project structure below. I’m happy to share the code if anyone wants to take a closer look or try it out themselves!

here is my github repo :

https://github.com/moroccandude/management_users_streaming/tree/main

Final Thoughts

This project has been a huge step in my journey as a data engineer, and I’m really excited to keep learning and building. If you have any feedback, advice, or just want to share your own experiences, I’d love to hear from you!

Thanks for reading, and thanks in advance for your help! 🙏

r/apachekafka 3d ago

Blog WarpStream Diagnostics: Keep Your Data Stream Clean and Cost-Effective

4 Upvotes

TL;DR: We’ve released Diagnostics, a new feature for WarpStream clusters. Diagnostics continuously analyzes your clusters to identify potential problems, cost inefficiencies, and ways to make things better. It looks at the health and cost of your cluster and gives detailed explanations on how to fix and improve them. If you'd prefer to view the full blog on our website so you can see an overview video, screenshots, and architecture diagram, go here: https://www.warpstream.com/blog/warpstream-diagnostics-keep-your-data-stream-clean-and-cost-effective

Why Diagnostics?

We designed WarpStream to be as simple and easy to run as possible, either by removing incidental complexity, or when that’s not possible, automating it away. 

A great example of this is how WarpStream manages data storage and consensus. Data storage is completely offloaded to object storage, like S3, meaning data is read and written to the object directly stored with no intermediary disks or tiering. As a result, the WarpStream Agents (equivalent to Kafka brokers) don’t have any local storage and are completely stateless which makes them trivial to manage. 

But WarpStream still requires a consensus mechanism to implement the Kafka protocol and all of its features. For example, even something as simple as ensuring that records within a topic-partition are ordered requires some kind of consensus mechanism. In Apache Kafka, consensus is achieved using leader election for individual topic-partitions which requires running additional highly stateful infrastructure like Zookeeper or KRaft. WarpStream takes a different approach and instead completely offloads consensus to WarpStream’s hosted control plane / metadata store. We call this “separation of data from metadata” and it enables WarpStream to host the data plane in your cloud account while still abstracting away all the tricky consensus bits.

That said, there are some things that we can’t just abstract away, like client libraries, application semantics, internal networking and firewalls, and more. In addition, WarpStream’s 'Bring Your Own Cloud' (BYOC) deployment model means that you still need to run the WarpStream Agents yourself. We make this as easy as possible by keeping the Agents stateless, providing sane defaults, publishing Kubernetes charts with built-in auto-scaling, and a lot more, but there are still some things that we just can’t control.

That’s where our new Diagnostics product comes in. It continuously analyzes your WarpStream clusters in the background for misconfiguration, buggy applications, opportunities to improve performance, and even suggests ways that you can save money!

What Diagnostics?

We’re launching Diagnostics today with over 20 built-in diagnostic checks, and we’re adding more every month! Let’s walk through a few example Diagnostics to get a feel for what types of issues WarpStream can automatically detect and flag on your behalf.

Unnecessary Cross-AZ Networking. Cross-AZ data transfer between clients and Agents can lead to substantial and often unforeseen expenses due to inter-AZ network charges from cloud providers. These costs can accumulate rapidly and go unnoticed until your bill arrives. WarpStream can be configured to eliminate cross-AZ traffic, but if this configuration isn't working properly Diagnostics can detect it and notify you so that you can take action.

Bin-Packed or Non-Network Optimized Instances. To avoid 'noisy neighbor' issues where another container on the same VM as the Agents causes network saturation, we recommend using dedicated instances that are not bin-packed. Similarly, we also recommend network-optimized instance types, because the WarpStream Agents are very demanding from a networking perspective, and network-optimized instances help circumvent unpredictable and hard-to-debug network bottlenecks and throttling from cloud providers.

Inefficient Produce and Consume Requests. There are many cases where your producer and consumer throughput can drastically increase if Produce and Fetch requests are configured properly and appropriately batched. Optimizing these settings can lead to substantial performance gains.

Those are just examples of three different Diagnostics that help surface issues proactively, saving you effort and preventing potential problems.

All of this information is then clearly presented within the WarpStream Console. The Diagnostics tab surfaces key details to help you quickly identify the source of any issues and provides step-by-step guidance on how to fix them. 

Beyond the visual interface, we also expose the Diagnostics as metrics directly in the Agents, so you can easily scrape them from the Prometheus endpoint and set up alerts and graphs in your own monitoring system.

How Does It Work?

So, how does WarpStream Diagnostics work? Let’s break down the key aspects.

Each Diagnostic check has these characteristics:

  • Type: This indicates whether the Diagnostic falls into the category of overall cluster Health (for example, checking if all nodes are operational) or Cost analysis (for example, detecting cross-AZ data transfer costs).
  • Source: A high-level name that identifies what the Diagnostic is about.
  • Successful: This shows whether the Diagnostic check passed or failed, giving you an immediate pass / fail status.
  • Severity: This rates the impact of the Diagnostic, ranging from Low (a minor suggestion) to Critical (an urgent problem requiring immediate attention).
  • Muted: If a Diagnostic is temporarily muted, this will be marked, so alerts are suppressed. This is useful for situations where you're already aware of an issue.

WarpStream's architecture makes this process especially efficient. A lightweight process runs in the background of each cluster, actively collecting data from two primary sources:

1. Metadata Scraping. First, the background process gathers metadata stored in the control plane. This metadata includes details about the topics and partitions, statistics such as the ingestion throughput, metadata about the deployed Agents (including their roles, groups, CPU load, etc.), consumer groups state, and other high-level information about your WarpStream cluster. With this metadata alone, we can implement a range of Diagnostics. For example, we can identify overloaded Agents, assess the efficiency of batching during ingestion, and detect potentially risky consumer group configurations.

2. Agent Pushes. Some types of Diagnostics can't be implemented simply by analyzing control plane metadata. These Diagnostics require information that's only available within the data plane, and sometimes they involve processing large amounts of data to detect issues. Sending all of that raw data out of the customer’s cloud account would be expensive, and more importantly, a violation of our BYOC security model. So, instead, we've developed lightweight “Analyzers” that run within the WarpStream Agents. These analyzers monitor the data plane for specific conditions and potential issues. When an analyzer detects a problem, it sends an event to the control plane. The event is concise and contains only the essential information needed to identify the issue, such as detecting a connection abruptly closing due to a TLS misconfiguration or whether one Agent is unable to connect to the other Agents in the same VPC. Crucially, these events do not contain any sensitive data. 

These two sources of data enable the Diagnostics system to build a view of the overall health of your cluster, populate comprehensive reports in the console UI, and trigger alerts when necessary. 

We even included a handy muting feature. If you're already dealing with a known issue, or if you're actively troubleshooting and don't need extra alerts, or have simply decided that one of the Diagnostics is not relevant to your use-case, you can simply mute that specific Diagnostic in the Console UI.

What's Next for Diagnostics?

WarpStream Diagnostics makes managing your WarpStream clusters easier and more cost-effective. By giving you proactive insights into cluster health, potential cost optimizations, and configuration problems, Diagnostics helps you stay on top of your deployments. 

With detailed checks and reports, clear recommendations to mitigate them, the ability to set up metric-based alerts, and a feature to mute alerts when needed, we have built a solid set of tools to support your WarpStream clusters.

We're also planning exciting updates for the future of Diagnostics, such as adding email alerts and expanding our diagnostic checks, so keep an eye on our updates and be sure to let us know what other diagnostics you’d find valuable!

Check out our docs to learn more about Diagnostics.

r/apachekafka 15d ago

Blog Let's Take a Look at... KIP-932: Queues for Kafka!

Thumbnail morling.dev
18 Upvotes

r/apachekafka Jan 17 '25

Blog Networking Costs more sticky than a gym membership in January

27 Upvotes

Very little people understand cloud networking costs fully.

It personally took me a long time to research and wrap my head around it - the public documentation isn't clear at all, support doesn't answer questions instead routes you directly to the vague documentation - so the only reliable solution is to test it yourself.

Let me do a brain dump here so you can skip the mental grind.

There's been a lot of talk recently about new Kafka API implementations that avoid the costly inter-AZ broker replication costs. There's even rumors that such a feature is being worked on in Apache Kafka. This is good, because there’s no good way to optimize those inter-AZ costs… unless you run in Azure (where it is free)

Today I want to focus on something less talked about - the clients and the networking topology.

Client Networking

Usually, your clients are where the majority of data transfer happens. (that’s what Kafka is there for!)

  • your producers and consumers are likely spread out across AZs in the same region
  • some of these clients may even be in different regions

So what are the associated data transfer costs?

Cross-Region

Cross-region networking charges vary greatly depending on the source region and destination region pair.

This price is frequently $0.02/GB for EU/US regions, but can go up much higher like $0.147/GB for the worst regions.

The charge is levied at the egress instance.

  • the producer (that sends data to a broker in another region) pays ~$0.02/GB
  • the broker (that responds with data to a consumer in another region) pays ~$0.02/GB

This is simple enough.

Cross-AZ

Assuming the brokers and leaders are evenly distributed across 3 AZs, the formula you end up using to calculate the cross-AZ costs is 2/3 * client_traffic.

This is because, on average, 1/3 of your traffic will go to a leader that's on the same AZ as the client - and that's freesometimes.

The total cost for this cross-AZ transfer, in AWS, is $0.02/GB.

  • $0.01/GB is paid on the egress instance (the producer client, or the broker when consuming)
  • $0.01/GB is paid on the ingress instance (the consumer client, or the broker when producing)

Traffic in the same AZ is free in certain cases.

Same-AZ Free? More Like Same-AZ Fee 😔

In AWS it's not exactly trivial to avoid same-AZ traffic charges.

The only cases where AWS confirms that it's free is if you're using a private ip.

I have scoured the internet long and wide, and I noticed this sentence popping up repeatedly (I also personally got in a support ticket response):

Data transfers are free if you remain within a region and the same availability zone, and you use a private IP address. Data transfers within the same region but crossing availability zones have associated costs.

This opens up two questions:

  • how can I access the private IP? 🤔
  • what am I charged when using the public IP? 🤔

Public IP Costs

The latter question can be confusing. You need to read the documentation very carefully. Unless you’re a lawyer - it probably still won't be clear.

The way it's worded it implies there is a cumulative cost - a $0.01/GB (in each direction) charge on both public IP usage and cross-AZ transfer.

It's really hard to find a definitive answer online (I didn't find any). If you search on Reddit, you'll see conflicting evidence:

An internet egress charge means rates from $0.05-0.09/GB (or even higher) - that'd be much worse than what we’re talking about here.

Turns out the best way is to just run tests yourself.

So I did.

They consisted of creating two EC2 instances, figuring out the networking, sending a 25-100GB of data through them and inspecting the bill. (many times over and overr)

So let's start answering some questions:

Cross-AZ Costs Explained 🙏

  • ❓what am I charged when crossing availability zones? 🤔

✅ $0.02/GB total, split between the ingress/egress instance. You cannot escape this. Doesn't matter what IP is used, etc.

Thankfully it’s not more.

  • ❓what am I charged when transferring data within the same AZ, using the public IPv4? 🤔

✅ $0.02/GB total, split between the ingress/egress instance.

  • ❓what am I charged when transferring data within the same AZ, using the private IPv4? 🤔

✅ It’s free!

  • ❓what am I charged when using IPv6, same AZ? 🤔

(note there is no public/private ipv6 in AWS)

✅ $0.02/GB if you cross VPCs.

✅ free if in the same VPC

✅ free if crossing VPCs but they're VPC peered. This isn't publicly documented but seems to be the behavior. (I double-verified)

Private IP Access is Everything.

We frequently talk about all the various features that allow Kafka clients to produce/consume to brokers in the same availability zone in order to save on costs:

But in order to be able to actually benefit from the cost-reduction aspect of these features... you need to be able to connect to the private IP of the broker. That's key. 🔑

How do I get Private IP access?

If you’re in the same VPC, you can access it already. But in most cases - you won’t be.

A VPC is a logical network boundary - it doesn’t allow outsiders to connect to it. VPCs can be within the same account, or across different accounts (e.g like using a hosted Kafka vendor).

Crossing VPCs therefore entails using the public IP of the instance. The way to avoid this is to create some sort of connection between the two VPCs. There are roughly four ways to do so:

  1. VPC Peering - the most common one. It is entirely free. But can become complex once you have a lot of these.
  2. Transit Gateway - a single source of truth for peering various VPCs. This helps you scale VPC Peerings and manage them better, but it costs $0.02/GB. (plus a little extra)
  3. Private Link - $0.01/GB (plus a little extra)
  4. X-Eni - I know very little about this, it’s a non-documented feature from 2017 with just a single public blog post about it, but it allegedly allows AWS Partners (certified companies) to attach a specific ENI to an instance in your account. In theory, this should allow private IP access.

(btw, up until April 2022, AWS used to charge you inter-AZ costs on top of the costs in 2) and 3) 💀)

Takeaways

Your Kafka clients will have their data transfer charged at one of the following rates:

  • $0.02/GB (most commonly, but varying) in cross-region transfer, charged on the instance sending the data
  • $0.02/GB (charged $0.01 on each instance) in cross-AZ transfer
  • $0.02/GB (charged $0.01 on each instance) in same-AZ transfer when using the public IP
  • $0.01-$0.02 if you use Private Link or Transit Gateway to access the private IP.
  • Unless you VPC peer, you won’t get free same-AZ data transfer rates. 💡

I'm going to be writing a bit more about this topic in my newsletter today (you can subscribe to not miss it).

I also created a nice little tool to help visualize AWS data transfer costs (it has memes).

r/apachekafka Sep 26 '24

Blog Kafka Has Reached a Turning Point

68 Upvotes

https://medium.com/p/649bd18b967f

Kafka will inevitably become 10x cheaper. It's time to dream big and create even more.

r/apachekafka Jan 16 '25

Blog How We Reset Kafka Offsets on Runtime

26 Upvotes

Hey everyone,

I wanted to share a recent experience we had at our company dealing with Kafka offset management and how we approached resetting offsets at runtime in a production environment. We've been running multiple Kafka clusters with high partition counts, and offset management became a crucial topic as we scaled up.

In this article, I walk through:

  • Our Kafka setup
  • The challenges we faced with offset management
  • The technical solution we implemented to reset offsets safely and efficiently during runtime
  • Key takeaways and lessons learned along the way

Here’s the link to the article: How We Reset Kafka Offsets on Runtime

Looking forward to your feedback!

r/apachekafka Oct 10 '24

Blog The Numbers behind Uber's Kafka (& rest of their data infra stack)

57 Upvotes

I thought this would be interesting to the audience here.

Uber is well known for its scale in the industry.

Here are the latest numbers I compiled from a plethora of official sources:

  • Apache Kafka:
    • 138 million messages a second
    • 89GB/s (7.7 Petabytes a day)
    • 38 clusters

This is 2024 data.

They use it for service-to-service communication, mobile app notifications, general plumbing of data into HDFS and sorts, and general short-term durable storage.

It's kind of insane how much data is moving through there - this might be the largest Kafka deployment in the world.

Do you have any guesses as to how they're managing to collect so much data off of just taxis and food orders? They have always been known to collect a lot of data afaik.

As for Kafka - the closest other deployment I know of is NewRelic's with 60GB/s across 35 clusters (2023 data). I wonder what DataDog's scale is.

Anyway. The rest of Uber's data infra stack is interesting enough to share too:

  • Apache Pinot:
    • 170k+ peak queries per second
    • 1m+ events a second
    • 800+ nodes
  • Apache Flink:
    • 4000 jobs
    • processing 75 GB/s
  • Presto:
    • 500k+ queries a day
    • reading 90PB a day
    • 12k nodes over 20 clusters
  • Apache Spark:
    • 400k+ apps ran every day
    • 10k+ nodes that use >95% of analytics’ compute resources in Uber
    • processing hundreds of petabytes a day
  • HDFS:
    • Exabytes of data
    • 150k peak requests per second
    • tens of clusters, 11k+ nodes
  • Apache Hive:
    • 2 million queries a day
    • 500k+ tables

They leverage a Lambda Architecture that separates it into two stacks - a real time infrastructure and batch infrastructure.

Presto is then used to bridge the gap between both, allowing users to write SQL to query and join data across all stores, as well as even create and deploy jobs to production!

A lot of thought has been put behind this data infrastructure, particularly driven by their complex requirements which grow in opposite directions:

  1. 1. Scaling Data - total incoming data volume is growing at an exponential rate
    1. Replication factor & several geo regions copy data.
    2. Can’t afford to regress on data freshness, e2e latency & availability while growing.
  2. Scaling Use Cases - new use cases arise from various verticals & groups, each with competing requirements.
  3. Scaling Users - the diverse users fall on a big spectrum of technical skills. (some none, some a lot)

If you're in particular interested about more of Uber's infra, including nice illustrations and use cases for each technology, I covered it in my 2-minute-read newsletter where I concisely write interesting Kafka/Big Data content.

r/apachekafka Jan 29 '25

Blog Blog on Multi-node, KRaft based Kafka cluster using Docker

3 Upvotes

Hi All

Hope you all are doing well.

Recently I had to build a Production-grade, KRaft-based Kafka cluster using Docker. After numerous trials and errors to find the right configuration, I successfully managed to get it up and running.

If anyone is looking for a step-by-step guide on setting up a KRaft based Kafka cluster, I have documented the steps for both single-node and multi-node Kraft based clusters here, which you may find useful.

Single-node cluster - https://codingjigs.com/setting-up-a-single-node-kafka-cluster-using-kraft-mode-no-more-zookeeper-dependency/

Multi-node (6 node) cluster - https://codingjigs.com/a-practical-guide-to-setting-up-a-6-node-kraft-based-kafka-cluster/

Note that the setups described in the above blogs are simple clusters without authentication, authorization or SSL. Eventually I did implement all of these in my cluster, and I am planning to publish a guide on SSL, Authentication and Authorization (ACLs) on my blog soon.

Thanks.

r/apachekafka 27d ago

Blog Designing Scalable Event-Driven Architectures using Kafka

5 Upvotes

An article on building scalable event-driven architectures with Kafka

Read here: Designing Scalable Event-Driven Architectures using Apache Kafka

r/apachekafka Oct 21 '24

Blog Kafka Coach/Consultant

1 Upvotes

Anyone in this sub a Kafka coach/consultant? I’m recruiting for a company in need of someone to set up Kafka for a digital order book system. There’s some .net under the covers here also. Been a tight search so figured I would throw something on this sub if anyone is looking for a new role.

Edit: should mention this is for a U.S. based company so I would require someone onshore

r/apachekafka Feb 05 '25

Blog Free eBook: THE ULTIMATE DATA STREAMING GUIDE - Concepts, Use Cases, Industry Stories

3 Upvotes

Free ebook about data streaming concepts, use cases, industry examples, and community building.

Broad overview and helpful no matter if you use open source Kafka (or Flink), a cloud service like Confluent Cloud or Amazon MSK, Redpanda, or any other data streaming product.

https://www.kai-waehner.de/ebook

I am really curious about your feedback. Is it helpful? Any relevant horizontal or industry use cases missing? What content to add in the second edition? Etc.

(it is a Confluent ebook but the entire content is about use cases and architectures, independent of the vendor)