9 ms·

Eliminating Task Processing Outages by Replacing RabbitMQ with Apache Kafka

98 points by sciurus 6y ago

move-on-by 6y ago
I enjoyed the read, but it glossed over some really important details: > Failovers took more than 20 minutes to complete and would often get stuck requiring manual intervention. Messages were often lost in the process as well. Data loss while RabbitMQ is down is certainly a problem. > The Kafka-consumer process is responsible for fetching messages from Kafka, and placing them on a local queue that is read by the task-execution processes. This is a bit confusing, so now instead of a centralized outage, grinding everything to a halt, the risk is distributed data loss as local queues go down? As long as they stay down, it should be limited data loss to the size of the queues times the number of consumers, but if an instance is flagging it could eat through a whole lot of messages before someone notices the problem. I guess it’s really dependent on the types of messages being processed and if they are idempotent enough to be replayed without consequence.
- secondcoming 6y ago
  Yes, this makes little to me. It seems that a slow task-executor can still cause the local queue to fill up. Each task-executor should have its own queue.
  - rad_gruchalski 6y ago
    The difference being that in Kafka the topic partition is isolated and the ingest is separate from consumption. Slow consumer does not slow down ingest which means that your task executor is simply that - processing is slower but no data is lost because ingest is still operational. Kafka is not a queue. It's a distributed log. The difference is significant because in a queue, the message can be processed by a single worker within the single unit of time - message is picked, locked, nobody else can see it, after processing, message is either acked or nacked. In a log, multiple consumers can process the same message in parallel. If a message is bad, it is usually skipped by simply advancing the offset to the next offset. The log has the advantage that the reason for slow processing is easy to identify and isolate - it's the slow consumer. It's also very easy to monitor the consumption lag - simply look at how fast the maximum offset grows vs what's the current consumed offset (high / low watermark). If the difference grows, your consumer is too slow.
    - XorNot 6y ago
      This seems very much like saying that if you shoot the hostage in a hostage situation, you no longer have a hostage situation. Like...the whole point of having messages work that way is that you're not having to bake event processing and duplicate transaction handling and compensating transactions and all the other message handling logic into literally every process which comes after your message queue. It's a non-solution - you're just leaking the problem out to another part of the stack.
      - Joeri 6y ago
        It isn’t actually that complicated. The kafka client abstracts away the complexity of tracking offsets. You fetch batches of messages through the kafka client in a loop, process them one by one, and commit the batch’s offset if everything went well. If something goed wrong, after restarting the consumer the kafka client will make sure it gets offered the same batch again. So you’re in an at least once scenario where as long as you process messages in an idempotent way and skip the bad ones it is basically foolproof.
        rad_gruchalski 6y ago
        Indeed. Plus it automatically handles consumer group rebelancing and partition leader changes. From the user perspective, one only needs to ensure having enough consumers in a consumer group. At Klarrio, we have a rather small cluster in production but we can easily push 2.4Gbps with <100ms end to end latency for dozens of consumers. It’s working really well.
shanemhansen 6y ago
In my experience Kafka is alot of work. If you need the guarantees it provides or the ecosystem it unlocks then by all means, use it. If you can relax your constraints, then something like NSQ can be a great option that I heartily recommend.
alexfromapex 6y ago
I think they make some fair points but I also don’t think they had a firm enough grasp on how RabbitMQ works
- Fizzadar 6y ago
  Came here to say this, in particular: > This problem was never root caused, though we suspect an issue in the Celery workers themselves and not RabbitMQ. So the problem wasn't Rabbit itself, but the usage/lack of understanding.
- thu2111 6y ago
  Could you elaborate? It's a great writeup. But I'm personally curious about them hitting the limits of vertical scalability. They said they used the biggest node 'available to them' without clarifying how big that was. If the issue was RabbitMQ hitting resource limits then upgrading its hardware (or swapping it out for some other similarly featureful broker but which is faster) might have fixed some of the problems too, albeit not things like insufficient observability. I also wonder to what extent RabbitMQ suffers from being written in Erlang. Switching to a featureful broker with sharding and replication that isn't Kafka e.g. Artemis might have allowed them to avoid rewriting code to not use helpful features. This performance blog suggests RabbitMQ may top out at ~4000 messages/sec when making things durable, which isn't especially good performance: https://softwaremill.com/mqperf/ https://softwaremill.com/mqperf/
  - conradfr 6y ago
    As someone noted in the comments it's because on RabbitMQ one queue uses only one CPU core by design and the whole benchmark is based on using one queue.
  - sumtechguy 6y ago
    Queue style systems usually start running into limits of the disk subsystem. Kafka is distributed log style. What that means is 'sharding' is basically built into kafka while MQ you need to build it in one layer up. Kafka can run into the same issues if you pick a poor hash for your partition. The messages per second really depends on the size of the messages too. For example in one of my tests I was hitting well over 10k-20k messages per second if the message size was 10 bytes (if my memory is working today). It dropped off dramatically as size grew. The limit was directly determined by disk and cpu. Switching to a different stack would mitigate some of that. But in the end if you expect scale (millions of messages an hour) you need a distributed system in there somewhere eventually. Would I personally reach for kafka out of the box? Not so much. I personally think it is quite a monster to get setup correctly. If you need dual datacenter up time it also becomes very expensive as you need to buy cloudera or confluent. Or have a plan to manage it yourself which will be 'interesting'.
Apreche 6y ago
What I find really interesting is that their decision making process. At least the way the article is written suggests they thought of five possible solutions. However, they decided to go all-in on a single solution without actually doing any significant investigation or actually trying the others. Option 1: Redis broker. Did they ever setup a test environment with a Redis broker to see how it compares to a RabbitMQ broker? Seems like they didn’t. Option 2: Same as option 1. Did they actually try to setup a kafka broker for celery to test it out and see how it goes? Option 3: Multiple brokers. They could have attempted a very simple setup starting with just two RabbitMQs and dividing the tasks roughly in half between them. This would allow scaling of RabbitMQ horizontally. At least try it and see. Option 4: Upgrade versions. Did they even try setting up an environment with upgraded versions. They say there’s no guarantee it fixes their observed bugs. How about trying it out and doing some more observation? Option 5: We decided to go all-in on this one without even trying the other 4 and doing any sort of comparison with actual tests or benchmarks. Maybe this was the best solution for them, and they chose wisely, but how do they know? The big thing that gets me is they talk about the observed issue of celery workers that stop processing tasks as well as the limited observability of celery workers and RabbitMQ. These aren’t black boxes. They are open source. You can debug them yourselves, report the bugs, fix the bugs and submit patches, add features, fork it yourself if necessary, etc. The fact that they don’t know if new versions will fix their observed bugs makes it clear they never identified those bugs. What’s the point of running on an open source stack if you’re going to treat things like a black box anyway?
- rad_gruchalski 6y ago
  > The fact that they don’t know if new versions will fix their observed bugs makes it clear they never identified those bugs. What’s the point of running on an open source stack if you’re going to treat things like a black box anyway? This will probably not be a popular opinion here but the business of DoorDash is not to fix open source software. I get your point. I use a lot of OSS in my business and we do contribute back. But not every business has the opportunity to observe the stack for days or weeks while the real clients suffer due to outages. Each client who is affected by an outage is most likely churning immediately and you're not going to win them back easily. In essence, the migration to Kafka will only benefit them long term and they have ultimately made the right business choice.
bydlocoder 6y ago
I wonder how they handle retries with backoff? It's something that Redis backend is pretty good at with sorted maps but kinda hard to implement in Kafka.
jsmeaton 6y ago
I sympathise - we’ve also experienced a bunch of issues with celery and rabbitmq and observing what’s actually happening is awful. I actually tweeted out some similar issues a week ago: https://twitter.com/jarshwah/status/1310820638655877120 https://twitter.com/jarshwah/status/1310820638655877120 There are ways to scale rabbitmq past 1 node without giving up too much throughput. - use multiple queues as each queue is single threaded - use sharding plugins which present a single queue externally but use multiple queues across a cluster internally - use lazy queues to get messages straight to disk, lower throughput but higher message count It doesn’t really sound like the other options were seriously considered and it was more about moving away from celery. That’s fine - just be honest about it.
- jojo2000 6y ago
  +1. I also think that their problem was more celery-related that rabbitmq related, but that's just a guess. We also evaluated kafka vs. rabbitmq, and chose rabbitmq. For people hitting performance degradation with rabbitmq this is a good reference [0] RabbitMQ should be used as-is without a middleman as much as possible, to make the best possible use of all of its concepts. [0] https://www.cloudamqp.com/blog/2018-01-19-part4-rabbitmq-13-common-errors.html https://www.cloudamqp.com/blog/2018-01-19-part4-rabbitmq-13-...
- zbentley 6y ago
  Per your tweet, yes, running without publish confirmations is a major unpleasant surprise that folks eventually run into. However, that's less Celery's fault and more RabbitMQ's: pub confirms are a Rabbit-specific extension to their protocol, and are left off in most example implementations. MongoDB, also, chooses horrible default settings; however, I wouldn't blame a Mongo-backed product entirely for being hackable. Additionally (I mentioned this in a sibling thread), Kafka's publish behavior is even worse. The vast majority of Kafka publishers' default settings don't even send the data to the broker without waiting for a response (like RabbitMQ without publish confirms); they send the data to a local buffer which is asynchronously flushed periodically/volumetrically. If your process crashes, you lose the buffer. Unpleasantly, if you force a buffer flush on every publish, Kafka's ingest rate goes from 10-100x RabbitMQ's to 0.01-0.1x RabbitMQ's.
michaelgiba 6y ago
I’m just surprised a single RabbitMQ node powered all of door dash for as long as it did
- fernandotakai 6y ago
  honestly, a single rabbitmq node on a really big machine can handle a ton of messages without problems. now that rabbitmq clustering is actually working (~7y ago it was basically impossible to setup clustering without having huge headaches), you can easily setup 3~5 machines in cluster and never have a single issue.
tannhaeuser 6y ago
I understand DoorDash/Uber to have Kafka in-house experience making this migration a no-brainer, but another option is to use RabbitMQ's AMQP protocol option and migrate to another AMQP-supporting product/project such as Apache's. Advantage of standardizing on protocols rather than products/projects and all. I wonder if anybody has done that.
- Plugawy 6y ago
  AFAIK there's no RabbitMQ-compatible broker, unless you use AMQP 1.0 in RabbitMQ right now. Their AMQP 0.9 implementation has a lot of RMQ-specific stuff and there's no equivalent.
  - mtmmtm 6y ago
    Yes there is (supporting AMQP 0.9.1) See here: https://joram.ow2.io/index.html https://joram.ow2.io/index.html
    - Plugawy 6y ago
      TIL, will check it out - thanks!
zbentley 6y ago
This is fascinating. Doordash seems to have encountered many of the same problems we (Klaviyo) did using a RabbitMQ/Celery stack, but arrived at totally different answers. Klaviyo doubled down on making the tools work for us, rather than jumping to a new and untested (for us) broker/job framework technology and arrived at some pretty nice solutions: Problems with observability? We added lots of custom StatsD and text logging instrumentation (Celery "signal" middleware), so that we could get e.g. accurate "how long do tasks like this spend waiting in the queue?" answers. Other than the inherent limitation of RabbitMQ being a strict queue (unlike Kafka, you can't "peek" at things in the middle of a topic without consuming that topic--we could do something hyper-complicated with exchanges and deliberate duplication of messages to address this limitation, but that doesn't sound worth it to me at all), observability of the brokers themselves seems pretty good. Coupled with Sentry reporting issues that occur during task processing, and some custom decorators to retry/delay/store tasks affected by common classes of systems issues, our visibility tends to be better than I've seen in any other asynchronous queue/message-bus system I've worked on. Sneaky task disappearances turned out to be mostly bugs in the way we were starting/stopping workers, related to signal handling and rabbitmq "redelivery". By really understanding the Celery worker start/stop lifecycle and coupling that with how Systemd chooses to kill processes, we were able to reduce those to zero. Celery also had a couple of "bugs" (questionable behavior choices) in this area which were resolved in 4.4. Celery ETA/countown task induced RabbitMQ load turned out to be because Celery made the questionable decision to queue ETA tasks on every single (eventual executor destination) worker node. We customized the celery task-dispatching code to route all ETA tasks to a set of workers which only buffer tasks, and re-deliver them to the actual work queues when their time is up. As a result, the domain of cases in which RabbitMQ had to "take back" (unack -> ready) large amounts of tasks went from "every time every worker restarts (and we deploy to them a lot!)" to "every time a very specific, single-purpose worker crashes", which reduced issues with that system to zero. A lack of scale-out in RabbitMQ was addressed by adding additional separate brokers (and therefore celery "apps") along two axes: sometimes we peel off specific task workloads to their own broker, and in other cases we run "fleets" of identical brokers that tasks round-robin across, with a (currently manual, moving in the direction of automatic) circuit breaker to take a broker out of rotation if it ever has issues. We wrote a whole blog post[1] about scaling that specific set of RabbitMQs and workers. Totally agree with Doordash that rabbit's "HA" generally isn't, and that scale-out needs to happen across brokers. Connection churn-related broker instability was addressed partially by scaling the number of brokers, but also on the consumer side, by carefully tuning per-worker concurrency to minimize connection counts while doing as much work as possible on a given piece of hardware, and by disabling the Celery remote control channel (pidbox queues). While that means that nice tools like e.g. Flower aren't as useful to us, it also means that the cost to a RabbitMQ broker of losing a whole bunch of consumer connections is much lower. When it comes to the "harakiri" churn of publisher connections discussed in the article, we haven't encountered connection issues due to our publisher tier. Doordash's web tier is almost certainly bigger than ours, but I'd make a deeply uneducated guess that we're at most an order of magnitude apart. I'd be curious to learn more about the story there, since, even at a reduced size, we regularly run 10ks of connections on a single broker with a pretty high flap-rate due to e.g. recycling webserver processes or restarting consumers due to code releases. In general, I agree with the article and yesterday's Celery 5.0 release post comments, that Celery is a quirky piece of tech, and that RabbitMQ is far from simple to run. However, I'm generally pretty pleased with Klaviyo's approach to go "through" the problem by diving deep on issues we had and fixing them in the stack we chose, rather than tossing large parts of it and re-learning the foibles of some other piece of software. At present, we run dozens of brokers and process 100ks of tasks per second at peak volume. While nobody considers our setup simple or issue-free, it's one of the most fully understood pieces of technology we run. While it's not out of the question for us to adopt it at some point in the future, Kafka was discouraging to us when hardening our RabbitMQ/Celery setup for a few reasons (though we do use it for some other pieces of our infrastructure which require it): As the Doordash folks indicated in the article, Kafka is really not well-integrated with the Celery stack at all, so building in things like front-vs-back-of-queue retries (both of which are extremely useful in different situations), deferred delivery, and the ability to rapidly change the number of consumers on a topic all take effort. Each of those problems has a solution, or at least a response, in the Kafka ecosystem, but Python task-processing frameworks which integrate those behaviors are both unfamiliar to us, and significantly younger than Celery. As with any clustered (rather than sharded) system, we lack expertise in understanding why publishes fail when Kafka is in a partially degraded state. With our existing Kafka workloads, many failure waves (consume or publish) happen without a full grasp of what's wrong/how to fix it. That's most definitely an "us" problem, and we are learning, but it's likely going to be quite awhile before we're at the comfort level that we currently have with, say, recovering message data from a data volume in the aftermath of a massive AWS-induced broker crash, or replacing RabbitMQ nodes that are experiencing elevated latency or flow control. Lastly, we were unpleasantly surprised by Kafka publishers habit of lying to their clients and saying a message was published when it was in fact buffered in-memory, pending a periodic (or volume-initiated) flush operation. Our processes crash a lot, usually when we least want them to, and having those crashes cause the data loss of an entire pre-publish Kafka buffer has been extremely unpleasant for us. When we reduced "batch.size" to 1, we discovered, to our dismay, that Kafka's vaunted "way better than RabbitMQ" publish time and volume numbers were entirely dependent on batch-wise optimizations, and that publishes were tens of times slower with batch.size=1 Kafka than they were with pub-confirms-enabled RabbitMQ (RabbitMQ also has batch-wise optimizations with publish confirms, which I'd argue have vastly better reliability semantics than Kafka, but that's another story and we're not using those...yet; ask me if interested). Again, that's partly an "us" problem (our crash rate is high, and dropped Kafka-destined batches could be recovered in other ways if we spent the time), but one that we don't have to worry about in the Celery/RabbitMQ setup. 1. https://klaviyo.tech/load-testing-our-event-pipeline-2019-42c984b90aae https://klaviyo.tech/load-testing-our-event-pipeline-2019-42... Edits: a few for clarity and removed a few things that the Doordash folks already covered and that my initial less-than-careful read didn't catch. The substance of my post didn't change.
ErlangSolutions 6y ago
We have a dedicated RabbitMQ engineering team here at Erlang Solutions. If you need assistance with scaling RabbitMQ in distributed environments, check out one of our videos https://youtu.be/MFH-GDYdxwQ https://youtu.be/MFH-GDYdxwQ or reach out, our experts would be happy to assist.
02020202 6y ago
kafka is the jehovah's witnesses of tech. not even going to bother reading it.
- kyuudou 6y ago
  On initial read it does seem... Kafkaesque, can I say edit: in this sense - https://en.wiktionary.org/wiki/Kafkaesque https://en.wiktionary.org/wiki/Kafkaesque
avinassh 6y ago
> Celery allows users to schedule tasks in the future with a countdown or ETA. Our heavy use of these countdowns resulted in noticeable load increases on the broker. Some of our outages were directly related to an increase in tasks with countdowns. Celery allows users to schedule tasks in the future with a countdown or ETA. Our heavy use of these countdowns resulted in noticeable load increases on the broker. Some of our outages were directly related to an increase in tasks with countdowns. We ultimately decided to restrict the use of countdowns in favor of another system we had in place How this problem of scheduled /delayed tasks was solved after moving to Kafka? The another system they mentioned, is something else entirely than the solution proposed in the post?
kerblang 6y ago
> The Kafka-consumer process is responsible for fetching messages from Kafka, and placing them on a local queue that is read by the task-execution processes Ah, yep, as usual you end up with a queue for your queue because Kafka isn't actually a queue in the first place. This is why I recommend Kafka for _data_, rabbit for _tasks_. The article doesn't really explain how this works, but I'm guessing that one service holds commands in memory, then sends http messages to the workers. As they noted, you end up with lost tasks on restart, and that's the big gaping hole that a lot of people would not want to fall into - your mileage may vary. I assume the kafka reader service has to also track how busy each worker is, also lost on restart.
- thebean11 6y ago
  If you have some mapping from data to units of work, can't you just use callbacks to make sure the data isn't acked until the units of work complete successfully? I use kafka at work and so far haven't really run into this as an issue. In the bit you quoted, I'd assume the local work queue has callback objects.
  - kerblang 6y ago
    Sorry for late answer: A Kafka ack is not a signal that an individual message is complete, but that the message index pointer can move forwards. So if you read 10 tasks and run them in parallel, you can't just ack a random selection of 5 as they complete. You could work out a system to ack the entire bunch when the last one finishes though. All of this assumes you're ok with more-than-once delivery, of course.
throwingtt 6y ago
Your rabbitmq connection churn should be at zero pretty much all the time. If you're seeing connection churn increase with volume you need to fix your client code.
orthecreedence 6y ago
I know this isn't a stylish opinion, but I've been using beanstalkd (https://beanstalkd.github.io/ https://beanstalkd.github.io/) in production for years now and never had a problem. Once in a while the server it's hosted on will have troubles, but the process itself is bulletproof and handles thousands of jobs/s without breaking a sweat. It has compacting binlogs, so for mission critical jobs that cannot be recreated you can save them to disk. For jobs that can be re-created, you can run an in-memory instance that can easily handle more jobs than most people here will ever need. If you want redundancy or write-scaling, you can run 2+ of them side by side, round-robin your jobs to the available instances, and have your workers read from them (RR as well). Beanstalkd is a lifesaver for job queues. It's the first thing I reach for when I need a queue. I'm excited to see how Redis' Disque turns out, but until then I'm happy with bean. I've evaluated replacing bean with something like Kafka, but it just doesn't make any sense at all. A distributed log is NOT a queue, and even using something like RabbitMQ gets tricky if you want to pull jobs instead of push them. For dedicated worker queues at small-to-medium scale, I cannot recommend beanstalkd enough. No, it doesn't have native sharding or multi-region replication, but like I said, I run millions of jobs/hour through it and never needed any of that crap.
- maxpert 6y ago
  Beanstalkd is probably the only beautiful queue I instantly fell in love with. It shares it's DNS with memcached, and makes life so much simple! I know people have the tendency to write complex abstractions and questionable choices around simple things. But sometimes simplicity and capability to reason well is the answer to everything.

98 points by sciurus 6y ago

I enjoyed the read, but it glossed over some really important details: > Failovers took more than 20 minutes to complete and would often get stuck requiring manual intervention. Messages were often lost in the process as well. Data loss while RabbitMQ is down is certainly a problem. > The Kafka-consumer process is responsible for fetching messages from Kafka, and placing them on a local queue that is read by the task-execution processes. This is a bit confusing, so now instead of a centralized outage, grinding everything to a halt, the risk is distributed data loss as local queues go down? As long as they stay down, it should be limited data loss to the size of the queues times the number of consumers, but if an instance is flagging it could eat through a whole lot of messages before someone notices the problem. I guess it’s really dependent on the types of messages being processed and if they are idempotent enough to be replayed without consequence.

secondcoming 6y ago
Yes, this makes little to me. It seems that a slow task-executor can still cause the local queue to fill up. Each task-executor should have its own queue.
- rad_gruchalski 6y ago
  The difference being that in Kafka the topic partition is isolated and the ingest is separate from consumption. Slow consumer does not slow down ingest which means that your task executor is simply that - processing is slower but no data is lost because ingest is still operational. Kafka is not a queue. It's a distributed log. The difference is significant because in a queue, the message can be processed by a single worker within the single unit of time - message is picked, locked, nobody else can see it, after processing, message is either acked or nacked. In a log, multiple consumers can process the same message in parallel. If a message is bad, it is usually skipped by simply advancing the offset to the next offset. The log has the advantage that the reason for slow processing is easy to identify and isolate - it's the slow consumer. It's also very easy to monitor the consumption lag - simply look at how fast the maximum offset grows vs what's the current consumed offset (high / low watermark). If the difference grows, your consumer is too slow.
  - XorNot 6y ago
    This seems very much like saying that if you shoot the hostage in a hostage situation, you no longer have a hostage situation. Like...the whole point of having messages work that way is that you're not having to bake event processing and duplicate transaction handling and compensating transactions and all the other message handling logic into literally every process which comes after your message queue. It's a non-solution - you're just leaking the problem out to another part of the stack.
    - Joeri 6y ago
      It isn’t actually that complicated. The kafka client abstracts away the complexity of tracking offsets. You fetch batches of messages through the kafka client in a loop, process them one by one, and commit the batch’s offset if everything went well. If something goed wrong, after restarting the consumer the kafka client will make sure it gets offered the same batch again. So you’re in an at least once scenario where as long as you process messages in an idempotent way and skip the bad ones it is basically foolproof.
      - rad_gruchalski 6y ago
        Indeed. Plus it automatically handles consumer group rebelancing and partition leader changes. From the user perspective, one only needs to ensure having enough consumers in a consumer group. At Klarrio, we have a rather small cluster in production but we can easily push 2.4Gbps with <100ms end to end latency for dozens of consumers. It’s working really well.

shanemhansen 6y ago

In my experience Kafka is alot of work. If you need the guarantees it provides or the ecosystem it unlocks then by all means, use it. If you can relax your constraints, then something like NSQ can be a great option that I heartily recommend.

alexfromapex 6y ago

I think they make some fair points but I also don’t think they had a firm enough grasp on how RabbitMQ works

Fizzadar 6y ago
Came here to say this, in particular: > This problem was never root caused, though we suspect an issue in the Celery workers themselves and not RabbitMQ. So the problem wasn't Rabbit itself, but the usage/lack of understanding.
thu2111 6y ago
Could you elaborate? It's a great writeup. But I'm personally curious about them hitting the limits of vertical scalability. They said they used the biggest node 'available to them' without clarifying how big that was. If the issue was RabbitMQ hitting resource limits then upgrading its hardware (or swapping it out for some other similarly featureful broker but which is faster) might have fixed some of the problems too, albeit not things like insufficient observability. I also wonder to what extent RabbitMQ suffers from being written in Erlang. Switching to a featureful broker with sharding and replication that isn't Kafka e.g. Artemis might have allowed them to avoid rewriting code to not use helpful features. This performance blog suggests RabbitMQ may top out at ~4000 messages/sec when making things durable, which isn't especially good performance: https://softwaremill.com/mqperf/ https://softwaremill.com/mqperf/
- conradfr 6y ago
  As someone noted in the comments it's because on RabbitMQ one queue uses only one CPU core by design and the whole benchmark is based on using one queue.
- sumtechguy 6y ago
  Queue style systems usually start running into limits of the disk subsystem. Kafka is distributed log style. What that means is 'sharding' is basically built into kafka while MQ you need to build it in one layer up. Kafka can run into the same issues if you pick a poor hash for your partition. The messages per second really depends on the size of the messages too. For example in one of my tests I was hitting well over 10k-20k messages per second if the message size was 10 bytes (if my memory is working today). It dropped off dramatically as size grew. The limit was directly determined by disk and cpu. Switching to a different stack would mitigate some of that. But in the end if you expect scale (millions of messages an hour) you need a distributed system in there somewhere eventually. Would I personally reach for kafka out of the box? Not so much. I personally think it is quite a monster to get setup correctly. If you need dual datacenter up time it also becomes very expensive as you need to buy cloudera or confluent. Or have a plan to manage it yourself which will be 'interesting'.

Apreche 6y ago

What I find really interesting is that their decision making process. At least the way the article is written suggests they thought of five possible solutions. However, they decided to go all-in on a single solution without actually doing any significant investigation or actually trying the others. Option 1: Redis broker. Did they ever setup a test environment with a Redis broker to see how it compares to a RabbitMQ broker? Seems like they didn’t. Option 2: Same as option 1. Did they actually try to setup a kafka broker for celery to test it out and see how it goes? Option 3: Multiple brokers. They could have attempted a very simple setup starting with just two RabbitMQs and dividing the tasks roughly in half between them. This would allow scaling of RabbitMQ horizontally. At least try it and see. Option 4: Upgrade versions. Did they even try setting up an environment with upgraded versions. They say there’s no guarantee it fixes their observed bugs. How about trying it out and doing some more observation? Option 5: We decided to go all-in on this one without even trying the other 4 and doing any sort of comparison with actual tests or benchmarks. Maybe this was the best solution for them, and they chose wisely, but how do they know? The big thing that gets me is they talk about the observed issue of celery workers that stop processing tasks as well as the limited observability of celery workers and RabbitMQ. These aren’t black boxes. They are open source. You can debug them yourselves, report the bugs, fix the bugs and submit patches, add features, fork it yourself if necessary, etc. The fact that they don’t know if new versions will fix their observed bugs makes it clear they never identified those bugs. What’s the point of running on an open source stack if you’re going to treat things like a black box anyway?

rad_gruchalski 6y ago
> The fact that they don’t know if new versions will fix their observed bugs makes it clear they never identified those bugs. What’s the point of running on an open source stack if you’re going to treat things like a black box anyway? This will probably not be a popular opinion here but the business of DoorDash is not to fix open source software. I get your point. I use a lot of OSS in my business and we do contribute back. But not every business has the opportunity to observe the stack for days or weeks while the real clients suffer due to outages. Each client who is affected by an outage is most likely churning immediately and you're not going to win them back easily. In essence, the migration to Kafka will only benefit them long term and they have ultimately made the right business choice.

bydlocoder 6y ago

I wonder how they handle retries with backoff? It's something that Redis backend is pretty good at with sorted maps but kinda hard to implement in Kafka.

jsmeaton 6y ago

I sympathise - we’ve also experienced a bunch of issues with celery and rabbitmq and observing what’s actually happening is awful. I actually tweeted out some similar issues a week ago: https://twitter.com/jarshwah/status/1310820638655877120 https://twitter.com/jarshwah/status/1310820638655877120 There are ways to scale rabbitmq past 1 node without giving up too much throughput. - use multiple queues as each queue is single threaded - use sharding plugins which present a single queue externally but use multiple queues across a cluster internally - use lazy queues to get messages straight to disk, lower throughput but higher message count It doesn’t really sound like the other options were seriously considered and it was more about moving away from celery. That’s fine - just be honest about it.

jojo2000 6y ago
+1. I also think that their problem was more celery-related that rabbitmq related, but that's just a guess. We also evaluated kafka vs. rabbitmq, and chose rabbitmq. For people hitting performance degradation with rabbitmq this is a good reference [0] RabbitMQ should be used as-is without a middleman as much as possible, to make the best possible use of all of its concepts. [0] https://www.cloudamqp.com/blog/2018-01-19-part4-rabbitmq-13-common-errors.html https://www.cloudamqp.com/blog/2018-01-19-part4-rabbitmq-13-...
zbentley 6y ago
Per your tweet, yes, running without publish confirmations is a major unpleasant surprise that folks eventually run into. However, that's less Celery's fault and more RabbitMQ's: pub confirms are a Rabbit-specific extension to their protocol, and are left off in most example implementations. MongoDB, also, chooses horrible default settings; however, I wouldn't blame a Mongo-backed product entirely for being hackable. Additionally (I mentioned this in a sibling thread), Kafka's publish behavior is even worse. The vast majority of Kafka publishers' default settings don't even send the data to the broker without waiting for a response (like RabbitMQ without publish confirms); they send the data to a local buffer which is asynchronously flushed periodically/volumetrically. If your process crashes, you lose the buffer. Unpleasantly, if you force a buffer flush on every publish, Kafka's ingest rate goes from 10-100x RabbitMQ's to 0.01-0.1x RabbitMQ's.

michaelgiba 6y ago

I’m just surprised a single RabbitMQ node powered all of door dash for as long as it did

fernandotakai 6y ago
honestly, a single rabbitmq node on a really big machine can handle a ton of messages without problems. now that rabbitmq clustering is actually working (~7y ago it was basically impossible to setup clustering without having huge headaches), you can easily setup 3~5 machines in cluster and never have a single issue.

tannhaeuser 6y ago

I understand DoorDash/Uber to have Kafka in-house experience making this migration a no-brainer, but another option is to use RabbitMQ's AMQP protocol option and migrate to another AMQP-supporting product/project such as Apache's. Advantage of standardizing on protocols rather than products/projects and all. I wonder if anybody has done that.

Plugawy 6y ago
AFAIK there's no RabbitMQ-compatible broker, unless you use AMQP 1.0 in RabbitMQ right now. Their AMQP 0.9 implementation has a lot of RMQ-specific stuff and there's no equivalent.
- mtmmtm 6y ago
  Yes there is (supporting AMQP 0.9.1) See here: https://joram.ow2.io/index.html https://joram.ow2.io/index.html
  - Plugawy 6y ago
    TIL, will check it out - thanks!

zbentley 6y ago

This is fascinating. Doordash seems to have encountered many of the same problems we (Klaviyo) did using a RabbitMQ/Celery stack, but arrived at totally different answers. Klaviyo doubled down on making the tools work for us, rather than jumping to a new and untested (for us) broker/job framework technology and arrived at some pretty nice solutions: Problems with observability? We added lots of custom StatsD and text logging instrumentation (Celery "signal" middleware), so that we could get e.g. accurate "how long do tasks like this spend waiting in the queue?" answers. Other than the inherent limitation of RabbitMQ being a strict queue (unlike Kafka, you can't "peek" at things in the middle of a topic without consuming that topic--we could do something hyper-complicated with exchanges and deliberate duplication of messages to address this limitation, but that doesn't sound worth it to me at all), observability of the brokers themselves seems pretty good. Coupled with Sentry reporting issues that occur during task processing, and some custom decorators to retry/delay/store tasks affected by common classes of systems issues, our visibility tends to be better than I've seen in any other asynchronous queue/message-bus system I've worked on. Sneaky task disappearances turned out to be mostly bugs in the way we were starting/stopping workers, related to signal handling and rabbitmq "redelivery". By really understanding the Celery worker start/stop lifecycle and coupling that with how Systemd chooses to kill processes, we were able to reduce those to zero. Celery also had a couple of "bugs" (questionable behavior choices) in this area which were resolved in 4.4. Celery ETA/countown task induced RabbitMQ load turned out to be because Celery made the questionable decision to queue ETA tasks on every single (eventual executor destination) worker node. We customized the celery task-dispatching code to route all ETA tasks to a set of workers which only buffer tasks, and re-deliver them to the actual work queues when their time is up. As a result, the domain of cases in which RabbitMQ had to "take back" (unack -> ready) large amounts of tasks went from "every time every worker restarts (and we deploy to them a lot!)" to "every time a very specific, single-purpose worker crashes", which reduced issues with that system to zero. A lack of scale-out in RabbitMQ was addressed by adding additional separate brokers (and therefore celery "apps") along two axes: sometimes we peel off specific task workloads to their own broker, and in other cases we run "fleets" of identical brokers that tasks round-robin across, with a (currently manual, moving in the direction of automatic) circuit breaker to take a broker out of rotation if it ever has issues. We wrote a whole blog post[1] about scaling that specific set of RabbitMQs and workers. Totally agree with Doordash that rabbit's "HA" generally isn't, and that scale-out needs to happen across brokers. Connection churn-related broker instability was addressed partially by scaling the number of brokers, but also on the consumer side, by carefully tuning per-worker concurrency to minimize connection counts while doing as much work as possible on a given piece of hardware, and by disabling the Celery remote control channel (pidbox queues). While that means that nice tools like e.g. Flower aren't as useful to us, it also means that the cost to a RabbitMQ broker of losing a whole bunch of consumer connections is much lower. When it comes to the "harakiri" churn of publisher connections discussed in the article, we haven't encountered connection issues due to our publisher tier. Doordash's web tier is almost certainly bigger than ours, but I'd make a deeply uneducated guess that we're at most an order of magnitude apart. I'd be curious to learn more about the story there, since, even at a reduced size, we regularly run 10ks of connections on a single broker with a pretty high flap-rate due to e.g. recycling webserver processes or restarting consumers due to code releases. In general, I agree with the article and yesterday's Celery 5.0 release post comments, that Celery is a quirky piece of tech, and that RabbitMQ is far from simple to run. However, I'm generally pretty pleased with Klaviyo's approach to go "through" the problem by diving deep on issues we had and fixing them in the stack we chose, rather than tossing large parts of it and re-learning the foibles of some other piece of software. At present, we run dozens of brokers and process 100ks of tasks per second at peak volume. While nobody considers our setup simple or issue-free, it's one of the most fully understood pieces of technology we run. While it's not out of the question for us to adopt it at some point in the future, Kafka was discouraging to us when hardening our RabbitMQ/Celery setup for a few reasons (though we do use it for some other pieces of our infrastructure which require it): As the Doordash folks indicated in the article, Kafka is really not well-integrated with the Celery stack at all, so building in things like front-vs-back-of-queue retries (both of which are extremely useful in different situations), deferred delivery, and the ability to rapidly change the number of consumers on a topic all take effort. Each of those problems has a solution, or at least a response, in the Kafka ecosystem, but Python task-processing frameworks which integrate those behaviors are both unfamiliar to us, and significantly younger than Celery. As with any clustered (rather than sharded) system, we lack expertise in understanding why publishes fail when Kafka is in a partially degraded state. With our existing Kafka workloads, many failure waves (consume or publish) happen without a full grasp of what's wrong/how to fix it. That's most definitely an "us" problem, and we are learning, but it's likely going to be quite awhile before we're at the comfort level that we currently have with, say, recovering message data from a data volume in the aftermath of a massive AWS-induced broker crash, or replacing RabbitMQ nodes that are experiencing elevated latency or flow control. Lastly, we were unpleasantly surprised by Kafka publishers habit of lying to their clients and saying a message was published when it was in fact buffered in-memory, pending a periodic (or volume-initiated) flush operation. Our processes crash a lot, usually when we least want them to, and having those crashes cause the data loss of an entire pre-publish Kafka buffer has been extremely unpleasant for us. When we reduced "batch.size" to 1, we discovered, to our dismay, that Kafka's vaunted "way better than RabbitMQ" publish time and volume numbers were entirely dependent on batch-wise optimizations, and that publishes were tens of times slower with batch.size=1 Kafka than they were with pub-confirms-enabled RabbitMQ (RabbitMQ also has batch-wise optimizations with publish confirms, which I'd argue have vastly better reliability semantics than Kafka, but that's another story and we're not using those...yet; ask me if interested). Again, that's partly an "us" problem (our crash rate is high, and dropped Kafka-destined batches could be recovered in other ways if we spent the time), but one that we don't have to worry about in the Celery/RabbitMQ setup. 1. https://klaviyo.tech/load-testing-our-event-pipeline-2019-42c984b90aae https://klaviyo.tech/load-testing-our-event-pipeline-2019-42... Edits: a few for clarity and removed a few things that the Doordash folks already covered and that my initial less-than-careful read didn't catch. The substance of my post didn't change.

ErlangSolutions 6y ago

We have a dedicated RabbitMQ engineering team here at Erlang Solutions. If you need assistance with scaling RabbitMQ in distributed environments, check out one of our videos https://youtu.be/MFH-GDYdxwQ https://youtu.be/MFH-GDYdxwQ or reach out, our experts would be happy to assist.

02020202 6y ago

kafka is the jehovah's witnesses of tech. not even going to bother reading it.

kyuudou 6y ago
On initial read it does seem... Kafkaesque, can I say edit: in this sense - https://en.wiktionary.org/wiki/Kafkaesque https://en.wiktionary.org/wiki/Kafkaesque

avinassh 6y ago

> Celery allows users to schedule tasks in the future with a countdown or ETA. Our heavy use of these countdowns resulted in noticeable load increases on the broker. Some of our outages were directly related to an increase in tasks with countdowns. Celery allows users to schedule tasks in the future with a countdown or ETA. Our heavy use of these countdowns resulted in noticeable load increases on the broker. Some of our outages were directly related to an increase in tasks with countdowns. We ultimately decided to restrict the use of countdowns in favor of another system we had in place How this problem of scheduled /delayed tasks was solved after moving to Kafka? The another system they mentioned, is something else entirely than the solution proposed in the post?

kerblang 6y ago

> The Kafka-consumer process is responsible for fetching messages from Kafka, and placing them on a local queue that is read by the task-execution processes Ah, yep, as usual you end up with a queue for your queue because Kafka isn't actually a queue in the first place. This is why I recommend Kafka for _data_, rabbit for _tasks_. The article doesn't really explain how this works, but I'm guessing that one service holds commands in memory, then sends http messages to the workers. As they noted, you end up with lost tasks on restart, and that's the big gaping hole that a lot of people would not want to fall into - your mileage may vary. I assume the kafka reader service has to also track how busy each worker is, also lost on restart.

thebean11 6y ago
If you have some mapping from data to units of work, can't you just use callbacks to make sure the data isn't acked until the units of work complete successfully? I use kafka at work and so far haven't really run into this as an issue. In the bit you quoted, I'd assume the local work queue has callback objects.
- kerblang 6y ago
  Sorry for late answer: A Kafka ack is not a signal that an individual message is complete, but that the message index pointer can move forwards. So if you read 10 tasks and run them in parallel, you can't just ack a random selection of 5 as they complete. You could work out a system to ack the entire bunch when the last one finishes though. All of this assumes you're ok with more-than-once delivery, of course.

throwingtt 6y ago

Your rabbitmq connection churn should be at zero pretty much all the time. If you're seeing connection churn increase with volume you need to fix your client code.

orthecreedence 6y ago

I know this isn't a stylish opinion, but I've been using beanstalkd (https://beanstalkd.github.io/ https://beanstalkd.github.io/) in production for years now and never had a problem. Once in a while the server it's hosted on will have troubles, but the process itself is bulletproof and handles thousands of jobs/s without breaking a sweat. It has compacting binlogs, so for mission critical jobs that cannot be recreated you can save them to disk. For jobs that can be re-created, you can run an in-memory instance that can easily handle more jobs than most people here will ever need. If you want redundancy or write-scaling, you can run 2+ of them side by side, round-robin your jobs to the available instances, and have your workers read from them (RR as well). Beanstalkd is a lifesaver for job queues. It's the first thing I reach for when I need a queue. I'm excited to see how Redis' Disque turns out, but until then I'm happy with bean. I've evaluated replacing bean with something like Kafka, but it just doesn't make any sense at all. A distributed log is NOT a queue, and even using something like RabbitMQ gets tricky if you want to pull jobs instead of push them. For dedicated worker queues at small-to-medium scale, I cannot recommend beanstalkd enough. No, it doesn't have native sharding or multi-region replication, but like I said, I run millions of jobs/hour through it and never needed any of that crap.

maxpert 6y ago
Beanstalkd is probably the only beautiful queue I instantly fell in love with. It shares it's DNS with memcached, and makes life so much simple! I know people have the tendency to write complex abstractions and questionable choices around simple things. But sometimes simplicity and capability to reason well is the answer to everything.