The kicker is that Kafka can be rock solid in terms of handling massive throughput and reliability when the wheels are well greased, but there are a lot of largely undocumented lessons to learn along the way RE: configuration and certain surprising behavior that can arise at scale (such as https://issues.apache.org/jira/browse/KAFKA-2063, which our team ran into maybe a year ago & is only being fixed now).

Symptoms of these issues can cause additional knock-on effects with respect to things like leader election (we wound up with a "zombie leader" in our cluster that caused all sorts of bizarre problems) and graceful shutdowns.

Add to that the fact the software is still very much under active development (sporadic partition replica drops after an upgrade from 0.8.1 to 0.8.2; we had to apply some small but crucial patches from Uber's fork) & that it needs a certain level of operational maturity to monitor it all ... it's easy to get nervous about what the next "surprise" will be.

Having said all that, I'd use Kafka again in a heartbeat for those high volume use cases where reliability matters. Not sure I'd advise others without similar operational experience to do the same for anything mission critical, though -- unless you like stress. That stress is why Confluent is in business. :)

Kafka and Confluent Platform are very much still works in progress. I had to patch Kafka Connect HDFS connector because a fix I needed was left out of the last release. Be prepared to do something similar with any of Kafka's components.

I am wondering if working on solving the actual problems with Kafka would have been the better route. I've never used Kafka and i find ZeroMQ great, but reading that their logging solution does drop log messages is a huge no-go for operations. How can you claim to run a serious business and say "babies will die" when you can't be sure to be able to find problems?

Because, when will you lose logs? Not in normal operation, but when weird things happen. When networking has a hiccup. When Load on the system is too high, so most likely when many people are using your service. Exactly when shit hits the fan. And you just made the decision that it's ok to drop log messages in such cases? That's not good.

I think you should either dive into Kafka/Zookeeper and fix your problems or switch to another logging solution. You should probably just drop that non-sense "streaming and real-time logs" requirement and live with a log delay of a few seconds and build something really stable instead of building something inherently unstable. Honestly, just collecting syslogs on the core vm and sending them to a central server would have been the better solution. Better then looking into fancy real-time, streaming logs on a sunday night because the system is having a breakdown and you can't even be sure that you are not missing essential logs.

How do you end up with a system design that can generate logs so fast that you can't keep up? It seems to be that some fundamental element of capacity planning was missed long ago and we're trying to fix the symptoms and not the cause.

If I have a geographically distributed system, I'm going to have bandwidth and latency issues if I try this, sure. But why do I care? A request to the SF data center shouldn't involve the Munich data center. That is, if I care about response times, and if I don't care about that, then why do I care about instantaneous log availability?

I think sometimes we get so bored with the problems we have that we invent new things to get upset about. Or management does, which is always worse.

If you don't make the choice consciously, say by assuming that you can buffer to disk and avoid the problem, at some point you'll fill up your disks and your system will block: you'll have unknowingly picked the "block" option. If you decide to rotate logs and delete old rotations when too many logs are present, then you're picking the "drop" option...

ZMQ silently drops messages when subscribers fail or not listening or when buffers fill up, but as they describe later on "access to historical logs", it's much easier to set up separate process/es for just that.

It seems that when shit hits the fan for this reason ZMQ really is a more reliable choice because it's more flexible.

Also, as he describes in the article, historical logs are scoped out and it is "likely" they they will develop something for those logs in the future. Again it looks like the plan is to use ZMQ and a subscriber to put those logs into logstash. That doesn't solve the problem i mentioned at all. ZMQ may still drop the logs! So, as far as i understand they don't have a plan for reliable logging. Even if they would, they'd have one reliable solution and an unreliable solution. The unreliable ZMQ based approach is probably neat and leads to fancy realtime dashboard stuff, but since it's not a reliable source of information it's not a good solution for operating a system where "babies will die".

And our log messages are ridiculously big at times (15k to as big as 50k).

Our pipe never has problems. What fails for us is Elastic Search. In fact at one point in the past we did 100k messages/s when embarrassingly had debug turned on in production and RabbitMQ did not fail but Elastic Search and sadly Flume did as well (I tried to get rid of flume with a custom Rust AMQP to Elastic Search client but at the time had some bugs with the libraries.. Maybe I will recheck out Mozilla Heka someday).

There is this sort of beating of the developer chest with a lot of tech companies.. that hey listen we are ultra important and we are dealing with ridiculously traffic and we need ultra high performance. Please tell/show me these numbers.... Or maybe stop logging crap you don't need to log.

Or maybe I'm wrong and we should log absolutely everything and Auth0 made the right choice given their needs (lets assume they have millions of messages a second), I still think I could make a sharded RabbitMQ go pretty far.

This goes with other technology as well. You don't need to pick hot glamorous NoSQL when Postgresql or MySQL and a tiny bit of engineering will get the job done just fine particularly when mature solutions give you such many things free out of the box (RabbitMQ gives you a ton of stuff like a cool admin UI and routing that you would have to build in ZeroMQ).

BTW I didn't mean to denigrate Elastic Search (I assume that is why I'm getting downvoted.... a comment would help). We just haven't had the chance to upgrade it and properly configure it.

In fact Elastic has been pretty darn speedy as of lately particularly since we purge some of the data after 6 months (we still have permanent filesystem storage of logs of course).

Were you using Zen Discovery at the time?

I haven't kept up with ES development in the last year so maybe they fixed this, but a flaky network can cause a cluster using Zen to freak out a lot.

How many MB/s are you indexing?

I'd be really interested to hear how you can achieve such a thoughput with rabbitmq

I do know we use multiple queues and even exchanges (and I did not know about the one core to queue).

A simple googling shows though folks have achieved far greater throughput[1] than 100k (and by the way this wasn't sustained.. it was spikes).

[1]: https://blog.pivotal.io/pivotal/products/rabbitmq-hits-one-m...

If you don't need amqps out there are more modern, better supported projects.

I was surprised by the contrasting sense of importance of delivery guarantees in the article. At the start, losing a message was akin to the death of a child. At the end, shrug. Now every single machine failure (or even ømq process restart) failure will lose you log messages stored in memory :(.

Glad to hear you found a solution that worked for you though! Would love to hear about difficulties you had with the new system, in particular adding brokers.

I've never heard anyone say managing a production Kafka cluster was easy or simple. Well, anyone who has had to actually maintain such clusters hasn't said it anyway.

True, but it appears to me that availability problems and dropped log messages often have the same root cause - network issues.

So whenever they do have availability issues (and dying babies) they won't be able to investigate properly because log messages are being lost as well.

That's obviously a very general observation. It may well be that in their architecture availability issues are mostly caused by something unrelated to networking (e.g. the database).

So the question then becomes what are the failures modes of their logging setup in terms of misbehaving clients? I don't know how kafka handles misbehaving clients. I suspect it would lead to global effects and slowdown of the entire cluster because of 1 or 2 misbehaving clients whereas in the current set up local misbehavior will be localized to the nearest aggregator dropping messages. Simple memory usage and other kinds of monitoring can then be used to find these issues and then mitigate them accordingly.

This is still a heck of lot simpler setup than using kafka and worrying about all sorts of weird distributed system failure modes. I'm sure kafka got them started initially but continuing to use it is like using a sledgehammer to kill a fly. For the use case they have this setup is the correct one and migrating to kafka if it becomes necessary will be possible. So in my view this is proper engineering. They've made all the right trade-offs instead of just chasing fads and trends.

Yes, they absolutely could do that, but they apparently don't. And maybe that's because they would lose a lot of the simplicity they won by ditching Kafka.

Anyway, I didn't want to defend Kafka specifically. The one time I considered it, I ended up not using it because it seemed too heavy weight for my use case in terms of memory usage and complexity.

Running Zk and Kafka on the same nodes is likely not the best thing.

Though for my part I still don't understand why Zookeeper wasn't built as a library to add distributed strongly consistent coordination to software that needs/benefits from it rather than being an external service that needs to be connected to, and thus introduces a gnarly mess of new failure modes that make Zookeeper client behavior extremely critical and often fragile. Something that's more like a "libpaxos/libraft" (e.g. serf for Go-lang or riak_ensemble for Erlang) seems a lot more valuable. /shrug

If we have K1,Z1 -- K2,Z2 -- K3,Z3 -- and one node goes down, you've now taken down both a broker and a Zk node. Remember, the brokers don't care about connecting to any Zk node, they want the leader. So you aren't gaining any more fault tolerant by co-locating them.

If there's a network partition between the leader Zk node and other nodes, the local Kafka broker won't actually be able to do much because the Zk cluster will elect a new leader, on another node, so again, you aren't gaining anything.

Moreover, you're now tying the scalability of Kafka with Zk. Zk doesn't scale linearly, so there's only so many nodes you may have in a cluster. Kafka, on the other hand, scales linearly. So if you're colocating them and you have to bump up Kafka, do you still start up Zk for those nodes (but they don't actually join the cluster)? You're now special casing and adding more edge cases.

But if the Auth0 runs their entire operations on AWS, maybe Kinesis would have been a more natural transition.

To anyone reading this with Kafka experience, do you have any tips/advice when it comes to maintaining a Kafka service?

Use configuration management such as chef to allow you to quickly build new nodes and to roll out changes accross the cluster. You will need to make tweaks. The chef Kafka cookbook which is the top result on Google has means of coordinating restarts of brokers accross the cluster. Use consul as a locking mechanism for this. You could use zookeeper, but consul works well for auto DNS registration and auto discovery.

Use the yahoo Kafka-manager app to manage the cluster and to see what is going on.

Don't use the Kafka default of storing data in /tmp/. Your OS will periodically clean it.

That seems like MySQL level of bad defaults.

Thanks for the tips.

If you want highly available ZK, your choices are 3, 5, 7... nodes, for which you can have 1, 2, or 3 nodes offline at any one time.

If you have one node fully down on a 3 node cluster, and there is even a tiny network blip or partition (as often happens in cloud environments) then you are down.

Kafka gives you features that certain systems cannot live without, like on disk persistence (saved my life couple of times) and topics. Filtering messages on the client side like ZeroMQ does it not an option in many cases, just think about security. I think Kafka has a long way to go before it can be used as a general message queue (many features are not there yet like visibility timeout for example) but if you can manage Zookeeper and have means to work with it (somebody understands it and knows its quirks) it can provide a reliable platform for distributing a large number of messages with low latency and high throughput, just like it does at LinkedIN.

  import socket
  import time
  cs = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  cs.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
  cs.setsockopt(SOL_SOCKET, SO_BROADCAST, 1)
  while True:
      cs.sendto('Node ID', ('255.255.255.255', 54545))
      time.sleep(4)

  import socket
  s=socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  s.bind(('',54545))
  m=s.recvfrom(1024)
  print m[0]

Lots of ways to skin this cat...

As for the SPOF: my logic was that if the DB is dead, there's no point having the ZeroMQ components come up anyhow. The jobs will keep on trying to come up, but until Ops brings the database back up, they can't do any damage.

It's amazing how small use case differences can mean very big differences in the effectiveness of various strategies. The at least once / at most once / no guarantee either way but soft real time, use cases, can mean radically different toolchains. This is the real lesson of distributed streaming.

I think it's much less complex than running an entire database node just for this, which btw will also require you constantly to poll, and will require you to bring in an (often heavyweight) client library into each node too, as opposed to standard-library sockets which if you're running multinode you're almost certainly already importing. If you're looking for "simple" distributed computing my sense is that that has yet to be invented.

Also what "c++ concepts" does it enforce and how? And why is that a bad thing?

In practice you build services which are internally composed and scaled using threads and processes, and which talk to each other over plain (unencrypted) protocols, e.g. TCP between boxes, IPC between processes on the same box. And when you need to talk to the outside world you build bridges that can speak any protocol you like, such as HTTP, or ZeroMQ's encrypted TCP protocol (CurveZMQ). From the developer's point of view, it's mostly transparent. Obviously you need security infrastructure, e.g. key/certificate management.

You can build entire ZeroMQ applications using only encrypted TCP and they will run on public infrastructure. You can develop and test the same apps on a laptop. You might run the same code using IPC on a laptop, and CurveZMQ over the Internet.

So though ZeroMQ definitely uses TCP as one of its transport protocols, it's not a replacement for it.

Which means that doing things like asking for the IP address of a peer make no sense. What's the IP address of a thread, or another process on the same box? If you need identity information, you should not use the transport protocol for that.

If deployed using the Netflix co-processes both are very durable.

Now about Kafka vs ZeroMQ: you want Kafka if you cannot tolerate the loss of even a single message. The append-only log with committed reader positions is a perfect fit for that.

It is? Can you provide any evidence supporting this claim?

Mesos is mostly used to deploy stateless services.

From what I can see it doesn't really support database deployments except for ArangoDB and Cassandra.

I used MQTT but only as a message bus.

Nothing magical/weird about it, just depends on whether or not you've got a nail to hammer with your MQ-hammer.

It's a great read and describes most scenarios well and easy to understand.

There are about a dozen message systems out there that will handle much more than Kafka with minimal or no operational overhead while supporting everything they need.

0MQ doesn't sound like it is the right solution either, but yeah... often you pick the wrong tool and learn something in the process.

I knew a big part of the reliability problems we were having was related to the distributed state that needed to be kept synchronized. I wanted to move to something simpler that did not rely on any durable, distributed state, while supporting the messaging patters we required. ZeroMQ fit the bill.

While there were other implementations with similar properties, there is no reasonable way to compare them up front given that what makes the real difference at the end of the day is the behavior of the system at 2am one day after a prolonged stress run. As a startup one does not have resources to conduct an up front analysis of that sort. You just take a bet. If it does not pan out, you pivot. This is exactly what we have done with the move from Kafka to ZeroMQ in the first place.

Now that we've been using ZeroMQ for over a year and have been perfectly happy, there is no incentive to look elsewhere.

There was some drama when the maintainer quit briefly before rejoining. Since then the gitter channel has been more active than I remember it being before. The mailing list is quiet it is true. Somebody just released a Rust version, and version 1.0.0 of nanomsg was indeed released.

You can see commit history here: https://github.com/nanomsg/nanomsg/commits/master/src

In the end the whole point of ZeroMQ was to build new protocols and APIs for decentralized messaging. My real disappointment with nano was that it made zero effort to build on existing work (mainly, ZMTP) and instead just started again, as if thousands of people hadn't spent years figuring out what a decentralized messaging protocol might look like.

It was worse than that, in fact. Nano launched itself on a wave of negativity. It makes good press, and poor everything else. Such hate for one's own history and knowledge base isn't healthy. If a messaging product isn't aiming at interoperability as a primary goal, it is worthless.

I'm not a fan of the "IETF or bust" approach either. That just doesn't work if you're a decentralized community. We needed and still need lightweight processes for RFC development. We use such a process (Digistan's COSS) in our RFCs. It wasn't random. I built Digistan and COSS over years after seeing AMQP swallowed up and destroyed by a committee. Why isn't nano using a process like COSS?

Without interoperable protocols, all we have is a bunch of software projects. And they die. And then all this is for nothing and the proprietary systems will rule the world and our dreams of making distributed software cheap again will die as well.

And this makes me angry: nano had the chance to push this forwards, and threw it away like old trash. What a stupid, petty waste of opportunity and goodwill.

Personally I started using zeromq, but was a bit disturbed by the experience of the saltstack guys with timeout problems and memory leaks. Probably fixed now, I should think. But more than that, the LGPL and MPL 2.0 licences aren't great for a commercial project that might in future need to be distributed to a different entity. So the license alone is a reason one might welcome greater diversity in available choices.

ZeroMQ "memory leaks and timeouts" wasn't the issue. Do you have a reference for that?

Salt's issues with ZMTP were performance; they wanted to multicast securely, and ZMTP does not support that. I'd complain once again that making their own custom protocol (actually, it seems to be largely just CurveCP) rather than working with us to extend ZMTP was poor form. Also, Salt wanted a pure Python stack.

Did that ever happen? I still like the idea behind nanomsg.

Which is all kind of a shame since it would have been so nice to see a new C engine in the community. We have new engines in C#, Erlang, Java, yet the old core project is still that somewhat clunky C++ engine.