Kafka at the low end: how bad can it get?
broot.ca
broot.ca
for Ignition?
Mostly people that ruminate on naive choices like REST/HTTP2/MQTT will have zero clue how the problems of multiple distributed telemetry sources scale. These kids are generally at another firm by the time their designs hit the service capacity of a few hundred concurrent streams per node, and their fragile reverse-proxy load-balancer CISCO rhetoric starts to catch fire.
Note, I've seen AMQP nodes hit well over 14000 concurrent users per IP without issue, as RabbitMQ/OTP acts like a traffic shock-absorber at the cost of latency. Some engineers get pissy when they can't hammer these systems back into the monad laden state-machines they were trained on, but those people tend to get fired eventually.
Note SCADA systems were mostly designed by engineers, and are about as robust as a vehicular bridge built by a JavaScript programmer.
Anecdotally, I think of Java as being a deprecated student language (one reason to avoid Kafka in new stacks), but it is still a solid choice in many use-cases. Sounds like you might be too smart to work with any team. =3
If you are inferring someone writing software for several decades might share, than one might want to at least reconsider civility over ones ego. Best of luck =3
"It works! so don't worry about spending a day or two exploring..." should be the takeaway insight about Erlang/RabbitMQ. Have a wonderful day. =3
Honestly from reading this it seems like you’re the one who is too smart to work with any team.
* easy to use * more efficient and lower latency than the big resources needed for Kafka
The efficiency really matters at scale and low latency yes but the simplicity of deployment and use is also a huge win.
How does redpanda compare under those constraints?
As for Redpanda vs Kafka in multi AZ setups and latency, the big win in Redpanda is tail latencies are kept low (we have a variety of techniques to do this). Here's some numbers here: https://www.redpanda.com/blog/kafka-kraft-vs-redpanda-perfor...
Multi AZ latency is mostly single digit millisecond (ref: https://www.bitsand.cloud/posts/cross-az-latencies/) and the JVM can easily take just as long during GC, which can drive up those tail latencies.
The fsync thing is complete FUD by RedPanda. They later introduce write caching[1] and call it an innovation[2]. I notice you also work for them.
Nevertheless, those that are super concerned with safety usually run with an RF of 5 (e.g banks). And you can configure Kafka to fsync as often as you want[3]
1 - https://www.redpanda.com/blog/write-caching-performance-benc... 2 - https://www.linkedin.com/posts/timfox_oh-you-cant-make-this-... 3 - https://kafka.apache.org/documentation/#brokerconfigs_log.fl...
It's just my polite way of saying it's safe enough for most use cases and that you're wrong.
Low volume data can be some of the most valuable data on the planet. Think SEC reporting (EDGAR), law changes (Federal Register), court judgements (PACER), new cybersecurity vulnerabilities (CVEs), etc. Missing one record can be detrimental if its the one record that matters.
Does everyone need durability by default? Probably not, but Redpanda users get it for free because there is a product philosophy of default-safe behavior that aligns with user expectations - most folks don't even know how this stuff works, why not protect them when possible?
The fsync thing is complete FUD by RedPanda.
You want durability? Pay the `fsync()` cost. Otherwise recognize that acknowledgement and durability are decoupled and that the data is sitting in unsafe volatile memory for a bit.
They later introduce write caching[1] and call it an innovation[2].
There are legitimate cases where customers don't care about durability and want the fastest possible system. We heard from these folks and responded with a feature they can selectively opt-in for that behavior _knowing the risks_. Again the idea is to be safer by default, and allow folks to opt-in to more risky behaviors.
those that are super concerned with safety usually run with an RF of 5 (e.g banks)
Going above RF=3 does not guarantee "more nines" since you need more independent server racks, independent power supplies or UPSs, etc, otherwise you're just pigeonholing yourself. This greatly drives up costs. Disks and durability is just cheaper and simpler. Worst case you pull the drives and pull the data off them, not fun and not easy, but possible unlike in-memory copies.
And you can configure Kafka to fsync as often as you want[3]
Absolutely! But nobody changes the default which is the issue - expectations of new users are not aligned with actual behavior. Same thing happened during the early MongoDB days. Either there needs to be better documentation/education to have people understand what the durability guarantees actually are, or change the defaults.
I agree that there needs to be better documentation.
I just don't agree that losing 3 replicas each living in a different DC at once is a realistic concern. The ones that would truly be concerned about this issue would do one of two things - run RF>3 (yes, it costs more) or set up some disaster recovery strategy (e.g run in multiple regions, yes that costs more.)
Because truth be told - losing 3 AZs at once is a disaster. And even if you durably persisted to disk - all 3 disks may have become corrupt anyway.
https://adamdrake.com/command-line-tools-can-be-235x-faster-...
Let’s be real: teams come to the infra team asking for a queue system. They give their requirements, and you—like a responsible engineer—suggest a more capable queue to handle their needs more efficiently. But no, they want Kafka. Kafka, Kafka, Kafka. Fine. You (meaning an entire team) set up Kafka clusters across three environments, define SLIs, enforce SLOs, make sure everything is production-grade.
Then you look at the actual traffic: 300kb/s in production. And right next to it? A RabbitMQ instance happily chugging along at 200kb/s.
You sit there, questioning every decision that led you to this moment. But infra isn’t the decision-maker. Sometimes, adding unnecessary complexity just makes everyone happier. And no, it’s not just resume-padding… probably.
Have I used (not necessarily decided on) Kafka in every single company/project for the last 8-9 years? Yes.
Was it the optimal choice for all of those? No.
Was it downright wrong or just added for weird reasons? Also no, not even a single time - it's just kinda ubiquitous.
If anybody knows of a simpler way to accomplish this, please do let me know.
Data volume was low though.
Then choose a different partitioning strategy. Often key based partitioning can solve this issue. Worst case scenario, you use a custom partitioning strategy.
Additionally , why can’t you match the number of consumers in consumer group to number of partitions?
The KIP mentioned seems interesting though. Kafka folks trying to make a play towards replacing all of the distributed messaging systems out there. But does seem a bit complex on the consumer side, and probably a few foot guns here for newbies to Kafka. [1]
[1] https://cwiki.apache.org/confluence/plugins/servlet/mobile?c...
Matching the number of consumers can still produce an uneven result too, and OP clarifies that even if the worst case he laid out doesn't happen - in practice there still are idle workers. For the same reason, he likely doesn't want to have 16 workers at all times
The way I describe Kafka is, "an event has transpired... sometimes you care, and choose to take an action based on that event"
The way I describe RabbitMQ is, "there's a new ticket in the lineup... it needs to be grabbed for action or left in the lineup... or discarded"
Definitely not perfect analogies. But they get the point across that Kafka is designed to be reactive and message queues/job queues are meant to be more imperative.
https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A...
Especially for low levels of load, that doesn't require that the dispatcher and consumer are written in the same language.
RabbitMQ or AWS SQS are probably good choices.
RabbitMQ is good for 1 producer -> 1 consumer with ack/nack
Right?
RabbitMQ/Erlang on OTP is probably one of the most solid solutions I've deployed over the years (low service cycle demands.) Highly recommended with the AMQP SSL credential certs, and GUID approach to application layer load-balancing. Cut our operational costs around 37 times lower than traditional load-balancer approaches. =3
On the other hand, if you know what you need to do and it’s supported by it, NATS is IME the way to go (particularly JetStream).
For hosted, SQS or Google Cloud Tasks. Google's approach is push-based (as opposed to pull-based) and is far and above easier to use than any other queueing system.
raw dogging gcloud? Terraform? or something more manageable?
I've been curious for one of my smaller projects, but I am worried about adopting more GCPisms.
gcloud beta resource-config bulk-export --resource-format=terraformNow that I'm mostly on AWS... I still use the same system. I have a thin little project that deploys to GAE and has a queue.yaml file. It sets up the cloud tasks queues. They hit my EB endpoints just like they used to hit my GAE endpoints.
As a bonus, my thin little GAE app also has a cron.yaml that it proxies to my AWS app. Appengine's cron is also better than Amazon's overcomplicated eventbridge system.
It's great.
eg we built a system at my last company to process 150 million objects / hour, and we modeled this using a postgres-backed queue with multiple processes pulling from the queue.
we observed that, whenever there were a lot of locked rows (ie lots of work being done), Postgres would correctly SKIP these rows, but having to iterate over and skip that many locked rows did have a noticeable impact on CPU utilization.
we worked around this by partitioning the queue, indexing on partition, and assigning each worker process a partition to pull from upon startup. this reduced the # of locked rows that postgres would have to skip over because our queries would contain a `WHERE partition=X` clause.
i had some great graphs on how long `SELECT FOR UPDATE ... SKIP LOCKED` takes as the number of locked rows in the queue increases, and how this partiton work around reduced the time to execute the SKIP LOCKED query, but unfortunately they are in the hands of my previous employer :(
Is not a low volume unless this could be done in batches of hundreds.
(or you can just use SQS or google cloud tasks, which work out of the box)
https://softwaremill.com/mqperf/
Maybe you should write a letter?
But I think the biggest reason I hit that number so easily was the ridiculous ease of batching. Starting with a query to select one task at a time, "converting" it select multiple tasks instead is ... a change of a single integer literal. FOR UPDATE SKIP LOCKED works the same regardless of whether your LIMIT is 1 or 1000.
Redesigned to create batches on the fly and then `SELECT FOR UPDATE batch SKIP LOCKED LIMIT 1` instead of `SELECT FOR UPDATE object SKIP LOCKED LIMIT 1000`. And just like that, 1000x reduction in load. Postgres is awesome.
----
The application is for processing updates to objects. Using a dedicated task queue for this is guaranteed to be worse. The objects are picked straight from their tables, based on the values of a few columns. Using a task queue would require reading these tables anyway, but then writing them out to the queue, and then invalidating / dropping the queue should any of the objects' properties update. FOR UPDATE SKIP LOCKED allows simply reading from the table ... and that's it.
A single application-layer thread doing batches of batch creation (heh). Not instant, but fast enough. I did have to add 'batchmaker is done' onto the 'no batch left' condition for worker exit.
> ... that pushes the locking from selecting queue entries to ...
To selecting batches. A batch is immutable once created. If work has to be restarted to handle new/updated objects, all batches are wiped and the batchmaker (and workers, anyway) start over.
Is the concept of having different solutions for different scales not familiar to you?
https://mikehadlow.blogspot.com/2012/04/database-as-queue-an...
The main complaint seems to be that it's not optimal...but then, the frame of the discussion was "Until you hit scale", so IMHO convenience and simpler infra trumps having the absolute most efficient tool at that stage.
These weren't his last words, but Jim Gray had this to say about this so-called "antipattern".
Queues Are Databases (1995)
Message-oriented-middleware (MOM) has become an small industry. MOM offers queued transaction processing as an advance over pure client-server transaction processing. This note makes four points: Queued transaction processing is less general than direct transaction processing. Queued systems are built on top of direct systems. You cannot build a direct system atop a queued system. It is difficult to build direct, conversational, or distributed transactions atop a queued system. Queues are interesting databases with interesting concurrency control. It is best to build these mechanisms into a standard database system so other applications can use these interesting features. Queue systems need DBMS functionality. Queues need security, configuration, performance monitoring, recovery, and reorganization utilities. Database systems already have these features. A full-function MOM system duplicates these database features. Queue managers are simple TP-monitors managing server pools driven by queues. Database systems are encompassing many server pool features as they evolve to TP-lite systems.
I'm not saying that configuring and deploying databases is easy, but it's probably going to happen anyway. Deploying and configuring Kafka is a huge headache: bad documentation, no testing tools, no way to really understand performance in the light of durability guarantees (which are also obscured by the poor quality documentation). It's just an honestly bad product (from the infra perspective): poor UX, poor design... and worst of all, it's kind of useless from the developer standpoint. Not 100% useless, but whatever it offers can be replaced by other existing tools with a tiny bit of work.
SELECT
*
FROM jobs
WHERE partition_key = (
SELECT partition_key
FROM jobs
ORDER BY partition_key
LIMIT 1
SKIP LOCKED
)
ORDER BY submitted_at
FOR UPDATE SKIP LOCKED;Also if events for a partition gets processed quick would the last partition get an equal chance?
Also, ServiceControl, ServiceInsight and ServicePulse are inventions of developers who are clearly WinAdmins who don't know what modern DevOps is. If you want to use that, you are bad and should feel bad.
(Sorry, I have absolute rage around this topic)
EDIT: If you insist, use MassTransit (https://masstransit.io/)
It's on the friendlier end of the spectrum among the tooling I help manage, at least compared to Microsoft crap.
Either way I'm feeling quite validated by your rage, so thanks for sharing. I feel like we could be good friends.
* Redis streams
* Redis lists (this is what Celery uses when Redis backend is configured)
* RabbitMQ
* ZeroMQ
Also give a shoutout to Beanstalkd (https://beanstalkd.github.io/)
The quickstart to getting everything running locally all documented here: https://docs.redpanda.com/current/get-started/quick-start/#d...
Disclaimer: I work at Redpanda
We have been using it in this application for half a decade now with no serious issues. I don't understand why it doesn't get more popular attention.
I don't see any evidence of StreamNative abandoning Pulsar at this point. I do see a compatibility layer for the Kafka protocol. That's fine.
I put it in quotes because I'm a massive fan of Pulsar and addressing the shortcomings of Kafka. However, with regards to some choices at a former workplace: The broader existing support/integration ecosystem along with Confluent's commercial capabilities won out with regards to technology choices and I was forced to acquiesce.
A bit like Betamax vs VHS, albeit that one pre-dates me significantly.
The main point of the article is that low message volumes mean you can get unlucky and end up with idle workers when there is still work to be done
The problem the author describes is 100% true and if you are scaled with enaugh workers this can turn out really bad.
While not beeing the only issue we faced (others are more environment/project-language specific) we got to a point where we decided to switch from kafka to rabbitmq.
You can over partition and make at-least-once workers happen (if you have a good Kafka client), or you use an http gateway and give up safe at-least-once. Hopefully this will make it easier to build an at-least-once style gateway that's easier to work with across a variety of languages. I know many have tried in the past but not dropping messages is hard to do right.
Uber wrote a Kafka push gateway years ago, when it was considerably harder to do well: https://www.uber.com/blog/kafka-async-queuing-with-consumer-...
> Note: when Queues for Kafka (KIP-932) becomes a thing, a lot of these concerns go away. I look forward to it!
Ultimately, the system was fast enough that the telco company emailed us and asked to slow down our requests because their API was not keeping up.
In short: we had two Apache Camel based apps: one to look at the database for paid content schedule, and queue up the messages (phone number and content). Then, another for triggering the telco company API.
Per Wiktionary, Kafkaesque: [1]
1. "Marked by a senseless, disorienting, often menacing complexity."
2. "Marked by surreal distortion and often a sense of looming danger."
3. "In the manner of something written by Franz Kafka." (like the software language was written by Franz Kafka)
Example: Metamorphosis Intro: "One morning, when Gregor Samsa woke from troubled dreams, he found himself transformed in his bed into a horrible vermin. He lay on his armour-like back, and if he lifted his head a little he could see his brown belly, slightly domed and divided by arches into stiff sections. The bedding was hardly able to cover it and seemed ready to slide off any moment. His many legs, pitifully thin compared with the size of the rest of him, waved about helplessly as he looked." [2]
[1] Wiktionary, Kafkaesque: https://en.wiktionary.org/wiki/Kafkaesque
[2] Gutenberg, Metamorphosis: https://www.gutenberg.org/cache/epub/5200/pg5200.txt
Seems like a good name for a high-volume distributed log that deletes based on retention, not after consumption.
With that (and sharding based on that ID/value) - all your consumers/workers will get equal amount of messages/tasks.
Both post and seemingly general theme of comments here is trashing choice of Kafka for low volume.
Interestingly both are ignoring other valid reasons/requirements making Kafka perfectly good choice despite low volume - e.g.:
- multiple different consumers/workers consuming same messages at their own pace
- needing to rewind/replay messages
- guarantee that all messages related to specific user (think bank transactions in book example of CQRS) will be handled by one pod/consumer, and in consistent order
- needing to chain async processing
And I'm probably forgetting bunch of other use cases.
And yes, even with good sharding - if you have some tasks/work being small/quick while others being big/long can still lead to non-optimal situations where small/quick is waiting for bigger one to be done.
However - if you have other valid reasons to use Kafka, and it's just this mix of small and big tasks that's making you hesitant... IMHO it's still worth trying Kafka.
Between using bigger buckets (so instead of 1 fetch more items/messages and handle work async/threads/etc), and Kafka automatically redistributing shards/partitions if some workers are slow ... You might be surprised it just works.
And sure - you might need to create more than one topic (e.g. light, medium, heavy) so your light work doesn't need to wait for heavier one.
Finally - I still didn't see anyone mention actual real deal breakers for Kafka.
From the top of my head I recall a big one is no guarantee of item/message being processed only once - even without you manually rewinding/reprocessing it.
It's possible/common to have situations where worker picks up a message from Kafka, processes (wrote/materialized/updated) it and when it's about to commit the kafka offset (effectively mark it as really done) it realizes Kafka already re-partitioned shards and now another pod owns particular partition.
So if you can't model items/messages or the rest of system in a way that can handle such things ... Say with versioning you might be able to just ignore/skip work if you know underlying materialized data/storage already incorporates it, or maybe whole thing is fine with INSERT ON DUPLICATE KEY UPDATE) - then Kafka is probably not the right solution.
If you want to continue processing other/newer items/messages (and usually you do), you need to commit Kafka topic offset - leaving you to figure out what to do with failed item/message.
One simple thing is just re-inserting it again into the same topic (at the end). If it was temps transient error that could be enough
Instead of same topic, you can also insert it into another failedX Kafka topic (and have topic processed by cron like scheduled task).
And if you need things like progressive backing off before attempting reprocessing - you liekly want to push failed items into something else.
While it could be another tasks system/setup where you can specify how many reprocessing attempts to make, how much time to wait before next attempt ...etc. Often it's enough to have a simple DB/table.
You say: > What that post describes (all work going to one/few workers) in practice doesn't really happen if you properly randomize (e.g. just use random UUID) ID of the item/task when inserting it into Kafka.
I would love to be wrong about this, but I don't _think_ this changes things. When you have few enough messages, you can still get unlucky and randomly choose the "wrong" partitions. To me, it's a fundamental probability thing - if you roll the dice enough times, it all evens out (high enough message volume), but this article is about what happens when you _don't_ roll the dice enough times.
.25^20 is not a "somewhat unlucky sequence of events"
Fair enough. I agree .25^20 is basically infinitesimal, and even with a smaller exponent (like .25^3) the odds are not great, so I appreciate you calling this out.
Flipping this around, though, if you have 4 workers total and 3 are busy with jobs (1 idle), your next job has only a 25% chance of hitting the idle worker. This is what I see the most in practice; there is a backlog, and not all workers are busy even though there is a backlog.
You control/configure numberOfShards - and its usually set to something order of magnitude bigger than your expected number of workers (to be precise - that's number of docker pods or hardware boxes/servers) - e.g. 32, 64 or 128.
So in practice - Kafka assigns multiple shards to each of your "workers" (if you have more workers than shards then some workers don't do any work).
And while each of your workers is limited to one thread for consuming Kafka messages. Each worker can still process multiple messages at the same time - in different async/threads.
While you could also use Kafka like that - and it might even work for your use case, as long as you configure option (sorry forgot the name) that makes Kafka redistribute shards because particular workers/consumers are too slow.
AFAIK the usual way is for each worker to get more than one message/item at a time, and do the actual item/work in/through separate thread/work pool (or another async mechanism).
Kafka then keeps track of which messages were picked up by each worker/consumer, and how big is the gap between that and committed offset (marked as done).
It gets a bit more tricky if you: - can't afford to process some messages/work again (well at extreme end it might actually be a show stopper for using Kafka) - need to have automatic retry on error/fail, how quickly/slowly you want to retry, how many times to retry...etc. - can you afford to temporarily "lose" some pending (picked up from Kafka but offset not marked as done) items for random things (worker OOMKILLED, solar flare hit network cable ...)
We've actually solved some of these with simply having another (set of) worker(s) that consume same topic with a delay (imagine cron job that runs every 5 minutes). And doing things in case there's no record of task being done, putting it into same topic again for retry ...etc.