boredandroid

In my opinion this is pre-cloud thinking.

It used to be that distributed systems were a big trade off. They were operationally complex, they had limited apis (NoSQL), but they scaled. The best solution used to be to build things using a non-scalable but easy to use and run system, and then re-write it later if it needed to scale (often in a big hurry).

This is just not the case any more, though. Why? Two reasons:

1. We’ve gotten much better at distributed systems so the apis aren’t nearly as limited. It’s no longer that you either choose hard-to-use things like Hadoop/NoSQL or elegant but unscalable single-server databases. You can have both good abstractions and scale.

2. The cloud makes it possible to get systems as a service so there should be way less ops than running a single node system yourself

In the case of Kafka, I’m super biased as I’m one of the original authors, but I think the abstractions Kafka gives, stream processing capabilities, connectors, etc are just way better than a lot of the traditional solutions. Using something worse until you “need” Kafka might make sense on premise, but not in the cloud.

Confluent offers a Kafka service which is fully managed so you don’t do any of the upgrades, security patches, midnight pages, etc you just use the APIs. This is super affordable for the kind of simple apps the article describes. The price varies by cloud, but e.g. on GCP it starts at $0.11/GB for reads and $0.10/GB stored. That is a lot cheaper than using a single node system and then rebuilding everything if you need to scale, but not only that, it is also lower operational overhead (effective none) and a better interface/abstraction.

I think this isn’t unique to Kafka, either. There are great managed systems that are built to scale for most of the kinds of data systems you would use—-CockroachDB, Spanner, Aurora, Snowflake, Elasticsearch, Bigquery, etc.

Basically, you can have nice things now, just like the big tech companies.

At my last startup, things would have been far less dysfunctional if the leadership team had read (and adopted) this philosophy. When we got backed by Venture funding it basically forced us into rapid scaling that we weren't prepared for and most of the engineering team quit because of mismanagement.

The point about effective managers need to stay out of the details and don't solve their team's problems can't be overstated.

This is Jay, the CEO of Confluent. No, actually quite the opposite. We took pains to ensure that the license was written so that software products like Landoop were free to embed our open source and compete with us in that way. This FAQ covers the competition in more detail: https://www.confluent.io/confluent-community-license-faq

This is Jay Kreps, I'm the CEO of Confluent. The license is actually not a EULA, that is a misunderstanding. We have added this item to our FAQ: https://www.confluent.io/confluent-community-license-faq

Confluent continues to fund the development of quite a lot of Apache 2.0 code and that is a huge part of our business and strategy. Perhaps your point is that if a company produces any non-open source code they are not an open source company?

I can address some of the critiques that seemed aimed at the blog post I wrote announcing the Confluent license change:

1. We aren't trying to get cloud providers to license our proprietary features. We run a cloud service of our software.

2. The book analogy is not very accurate. We have an FAQ here the helps clarify interpretation. The limitations it places are extraordinarily small, 99.9999% of users are completely unimpacted, it really only impacts companies wanting to offer, say, KSQL-as-a-service. https://www.confluent.io/confluent-community-license-faq

3. We aren't trying to "co-opt" the community or open terminology. We actually tried super hard both in the license and in the blog post to be honest and upfront. Whatever else you think you have to agree that Confluent's license is _exceptionally_ permissive and the software has a pretty great community of users. How do you describe a license that let's you run, modify, fork, and redistribute the code and do virtually anything other than offer a competing SaaS offering of the product?

4. Bryan Cantrill is an amazing engineer, but, well, as a lawyer, I think ours are probably better. We're quite confident in the enforceability, but it's a bit ironic because I remember this being the FUD around GPL that it was "totally unenforceable".

5. The "open source companies are all failing"-meme isn't factually correct. Many open source companies are actually doing quite well. MongoDB has gone up in value about 3x over the last year, Elastic was the breakout IPO of the year. There are a handful of other really strong businesses a year or so behind, including Confluent. An open source project is not in-and-of-itself a business model, but it is, just empirically, a big part of some of the recent successes in the infrastructure space. Probably worth noting that the reverse is true too: if you look at some of the really cool up-and-coming open source platform data technologies, a lot of them have the support of a company behind them. Of course there are plenty of sucky open source companies, but that is true of every category of startup.

6. I agree that it is silly to moralize about the behavior of the cloud providers. They are following their economic interest. The point is that this behavior does undermine the cycle of investment in some of the more promising hard tech open source projects and to try to change this dynamic.

7. This article has a bit of a tone of "Son, new things aren't possible, trust me, I tried them and have the scars to prove it". I have huge respect for Bryan, and I know that to some extent that is his schtick as a public personality, but I'm not sure that attitude is most likely to lead to improvement. I don't think the current crop of licenses was handed down from the mountain on Stone Tablets by our elders to be revered and not questioned. I think CockroachDB, Elastic, MongoDB, and Confluent are building really innovating technology platforms and building pretty cool companies to help fund that. I don't think we need dogma. And I still don't say "GNU/Linux".

We address that here: https://www.confluent.io/confluent-community-license-faq

Short answer: 1. AGPL doesn't actually solve this problem. This is why MongoDB just moved away from AGPL. 2. AGPL is quite restrictive for people who want to use the software in proprietary applications but don't want to open source their own code. This is a really large proportion of usage for us and we don't want to restrict people in that way.

Sure, I've updated the post so it doesn't imply that we comply with the OSI open source definition.

Thanks!

This is a great question. Many people think the AGPL solves this problem but it absolutely does not. This is why MongoDB, which was AGPL licensed, just changed to a custom license.

The other reason we don't want to use AGPL is that it can be quite aggressive in making you open source your own code if you want to embed our code in a proprietary application. We actually think building proprietary applications is a fine use of our software, and is one of the major use cases for this kind of infrastructure. This is addressed in more depth along with many other questions in this FAQ: https://www.confluent.io/confluent-community-license-faq

I agree that the proliferation of new licenses is annoying. But I think the reason you see changes in licensing from Elastic, MongoDB, etc is that the world has changed pretty dramatically since most of the original licenses were created. This proliferation existed early on in the days of open source and eventually died down to a set of standard things that solved the problems most people had. But the reality is the public cloud has dramatically changed the dynamic and many of the people who want to make code available with a permissive license don't have an option that is common and well understood and that protects them. I believe such a thing will likely emerge and will allow us all to standardize.

No, this has no effect on Kafka, which is part of the ASF and under the Apache 2.0 license. This just impacts things like KSQL that are part of Confluent Platform.

Hi all, I'm Jay Kreps, the CEO of Confluent and author of that blog post. I'm happy to answer any questions.

Yes the idempotence part of this feature set is very similar to TCP (the transactional consumption and updates obviously aren't). But this isn't a reimplementation at all. TCP provides deduplication within the context of a connection tied to a process. If that connection is lost or the process dies then duplicates may occur. The feature in Kafka is much stronger as the "connection" is persistent and replicated with the log so effectively the "connection" fails over if the server dies.

That is part of it but there is also a general purpose transaction feature that lets you link together updates, state journalling and consumption all in a single transaction. This enables correct stream processing on top of Kafka and is arguably the more technically sophisticated aspect.

To clarify: the transaction feature is a transaction over updates to Kafka and provides transparent, correct processing in Kafka for Kafka Streams or anything else only maintaining state/output in Kafka. The trick you described where you either transactionally or idempotently store your offsets and state updates in an external store when consuming in a general app still works, but obviously for that to work it requires there are no duplicates in the log itself, which the idempotence feature in the producer described here enables.

Dear hacker news:

1. Please read the section entitled "Is this Magical Pixie Dust I can sprinkle on my app?" Before making angry comments. Answer: for general consumer apps just consuming messages, no. However, Kafka's design, which let's the consumer control it's position in the log, combined with this feature which eliminates duplicates in the log make building end-to-end exactly once messaging using the consumer quite approachable. For stream processing using Kafka's Streams API (https://www.confluent.io/blog/introducing-kafka-streams-stre...), where you are taking input streams, maintaining state, and producing output streams, the answer is that it actually kind of is like magic pixie dust, you change a single config and get exactly-once processing semantics on arbitrary code. Obviously you still need to get the resulting data our of Kafka, but when combined with any off-the-shelf Kafka connector which maintains exactly-once you can get this for free. So for that style of app design you actually can get correct results end-to-end without needing to do any of the hard stuff.

2. Someone is going to come and say "FLP means that exactly once messaging is impossible!" or something else from a half-understood tidbit of distributed systems theory they picked up on a blog. Let me preempt that. FLP is about the impossibility of consensus in a fully asynchronous setting (e.g. no timeout-based failure detection). Of course as you know the vast majority of the systems you use in AWS or your own datacenter depend in deep ways on consensus. Kafka itself is a CP log, about as close of a direct analog to consensus as you could ask for. Obviously Kafka and all these systems are "impossible" in the same sense that if you can make the network or other latency issue bad enough you can make the system unavailable for writes. This feature doesn't change that at all, it just piggybacks on the existing consensus Kafka does. It doesn't violate any theorems in distributed systems theory: Kafka and any consensus-based system can't work in a fully asynchronous setting, Kafka was a CP system in the CAP sense prior to this feature and this feature doesn't change that guarantee.

For those who want a deeper dive into how it all works there is a longer write up on the design here: https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+E... The use for stream processing is described here: https://cwiki.apache.org/confluence/display/KAFKA/KIP-129%3A...

I'm one of the original Kafka creators and am an evil corporate shill at Confluent :-)

That was a super silly discussion and I'd be annoyed reading it too. It does not, however, represent any kind of "move away from open source" by Confluent.

The discussion was not actually about open source vs non-open source but rather around how to curate the Kafka ecosystem. Currently there are several hundred independent open source projects that act as connectors, clients, monitoring tools, processing libraries and frameworks, etc. Some of these are fantastic and active communities, some are built by companies that use Kafka, some by vendors that use Kafka in a product in some way. This has always been the structure of this ecosystem, since even before we moved Kafka into Apache. These projects are mostly Apache licensed and on github, but mostly not part of the ASF. You can see a subsample of them here: https://cwiki.apache.org/confluence/display/KAFKA/Ecosystem

The proposal from a group at Hortonworks was to duplicate one of the modules Confluent built, a REST layer, and build a new one in the main project. I don't think this is inherently a bad thing to do, but there wasn't really any concrete rationale...there was no complaint with the code in that module, nor any kind of complaint with the governance, nor any complaint with the community, etc. As a result a lot of us at Confluent kind of felt it was an odd exercise that had more to do with Hortonworks' policy of only shipping Apache projects, which in turn comes out of the weird fractious politics of the Hadoop ecosystem. The reality is that policy just doesn't fit with the Kafka ecosystem as it stands today.

We haven't broken with ZK, it just isn't part of the protocol clients use anymore.

This is important for three reasons: 1. To make clients easy to build 2. To have a single security layer for everything (so you don't have to authenticate yourself with two things) 3. To be able to enforce things like throttling and protect the system in an environment with lots of clients.

This article is pretty out of date, I think the central concerns have actually been addressed.

It's true that when we were working at LinkedIn Kafka tended to have much better Java support. Since founding Confluent (I'm one of the co-founders) we've really focused on improving the situation outside Java.

A few specific corrections:

1. We added full support for consumers with no interaction with zookeeper in the main kafka protocol. There is no longer any direct interaction with zookeeper from either the producer or consumer. We did this because we care a lot about the non-java clients.

2. Kafka has been extremely disciplined about backwards compatibility. The protocol comes with versioning and changes are always implemented in a way that supports both the old and new version and can be rolled out without downtime. In the five year history of the project we did one backwards incompatible release--the break from 0.5.x-0.7.x to 0.8.x. This was done intentionally to allow us to refactor the apis. I think this is a pretty good track record.

It's worth also addressing why Kafka clients directly access nodes in the cluster rather than requiring a proxy layer. The reason we do this is to allow very high throughput, partition aware processing. This is really required for use cases like stream processing that need to process data efficiently, especially in cases where you are reprocessing data. You can always build a proxy layer on top of direct access but not vice versa.

Confluent (where I work) is doing two things that help the non-java client ecosystem: 1. We maintain an open source REST proxy that provides decoupled access (albeit with a little overhead compared to the direct clients) 2. We have picked up work on clients. We offer and fully support a c/c++ client, a python client, and have a Go client coming soon. All of these are in feature parity with the Java clients. (More on the way).

Both of these efforts are open source and apache licensed and included in the open source Confluent Platform distribution of Kafka.

Several people asked how this compares to Kafka (I'm one of the people who created Kafka at LinkedIn). Here's my take:

I think the motivations they list are: 1. Different I/O model 2. Was started before Kafka had replication (the first release of Kafka with replication was in late 2013 I think)

The I/O model I'm less sure about, we looked at similar things for Kafka and they didn't seem worth it (basically you're doing a ton of stuff at the app level that the OS does pretty well--namely caching and buffering linear I/O), we'd have to look at actual benchmarks to know.

Here is my take on the pros and cons of the core tech.

Pros: - Seems to have better built in support for fencing/idempotence - Better geo placement?

Cons: - Lots more moving pieces. Already people are irritated that there are both Kafka nodes and ZK to set up. This system seems to split this over separate physical tiers for serving, core, storage, and zookeeper. My experience has been lot's of tiers is generally a big headache.

Neutral: - There seems to be a built in achival to HDFS. I think if the consumer is fast and efficient then you don't need to reach around your consumer api which will be high latency (since you have to wait for files to be closed out).

There is also a bunch of stuff Kafka does that I'm just not sure about how complete it is in DistributedLog: - Clients in a bunch of languages - Integration with all the major stream processing frameworks - Log compaction http://kafka.apache.org/documentation.html#compaction - Connector management http://www.confluent.io/blog/announcing-kafka-connect-buildi... - Quotas/throttling - Security/ACLs

I'm one of the Kafka developers, happy to answer any questions.

Thanks for clarifying!

I think there are two separate things: 1. Is there a principled notion of when a write is "committed" and a leadership election algorithm that works with this to ensure committed writes aren't lost as long as the failure criteria are met? This has been true since we added replication to Kafka. 2. The second is what is the recovery behavior when you have no available replicas. Previously we would aggressively elect any replica even if it had incomplete data. Now that behavior is configurable.

There is a more detailed explanation here: http://blog.empathybox.com/post/62279088548/a-few-notes-on-k...

This post is a bit confused about how Kafka replication works. Replication in Kafka is always synchronous in the sense that the cluster internally has a strong notion of which messages are committed and no uncommitted message is handed out to consumers. It is just that the client has the option of writing to Kafka without blocking while the servers commit the message.

This is described in more detail here: http://www.confluent.io/blog/distributed-consensus-reloaded-...

Here is how I think about this, there are three high level paradigms you see for processing: 1. Request/response (e.g. most UI actions, REST, etc) 2. Stream (e.g. subscribing to a Kafka topic) 3. Batch/periodic (e.g. Hadoop, DWH)

You actually need all of these at least somewhere in a company, and they each have their place.

The dividing line for request/response is that someone is waiting on the other end and that if the request fails you can just show them an error and continue on your way. So when a web service gets overwhelmed it usually just times out requests.

Consider an e-commerce site as an example: 1. Request/response Displaying a product Making a sale

2. Stream or batch Restocking Logistics and shipping Price adjustments Analytics Product catalog import Search index updates Etc

The later category is asynchronous so it can be done in a batch fashion (once an hour or day) if latency is not a concern, or in a streaming fashion if it needs to be faster.

One thing to realize is that a partitioned log is a generalization of an unpartitioned log (i.e. if you set # partitions = 1 in a partitioned log you have an unpartitioned log).

In Kafka the purpose of partitions is to provide computational parallelism not model entities in the world. So if you have 100m users you would map that into a number of partitions based on your computational parallelism (maybe 10-100 machines/processes/threads). In other words you would have a single topic partitioned by user id, not a topic per user.

If you have a centralized relational database that maps reasonably well to a single partition log (both in terms of scalability and guarantees).

For distributed databases you generally don't have a total order over all operations. What you usually have is (at best) a per partition ordering, which maps well to a partitioned log as well.

For applications that record events (logging or whatever) it is natural to think of each application thread or process as a kind of actor with a total order.

I really think there are a couple of levels of immutability that it is easy to conflate.

Specifically immutability for

1. In memory data structures...this is the contention of the functional programming people.

2. Persistent data stores. This is the lsm style of data structure that substitutes linear writes and compaction for buffered in-place mutation.

3. Distributed system internals--this is a log-centric, "state machine replication" style of data flow between nodes. This is a classic approach in distributed databases, and present in systems like PNUTs.

4. Company-wide data integration and processing around streams of immutable records between systems. This is what I have argued for (http://engineering.linkedin.com/distributed-systems/log-what...) and I think Martin is mostly talking about.

There are a lot of analogies between these but they aren't the same. Success of one of these things doesn't really imply success for any of the others. Functional programming could lose and log-structured data stores could win or vice versa. Pat Helland has made an across the board call for immutability (http://www.cidrdb.org/cidr2015/Papers/CIDR15_Paper16.pdf), but that remains a pretty strong assertion. So it is worth being specific about which level you are thinking about.

For my part I am pretty bullish about stream processing and data flow between systems being built around a log or stream of immutable records as the foundational abstraction. But whether those systems internally are built in functional languages, use lsm style data layout on disk is kind of an implementation detail. From my point of view immutability is a lot more helpful in the large than in the small--I have never found small imperative for loops particularly hard to read, but process-wide mutable state is a big pain, and undisciplined dataflow between disparate systems, caches, and applications at the company level can be a real disaster.

Anyhow in the Bay Area interested in learning more about Apache Samza should attend the meetup tonight in Mountain View: http://www.meetup.com/Bay-Area-Samza-Meetup/events/220354853...

I agree with this comment, but I think this blog post is really addressing data flow at the company-wide or datacenter scale. Surprisingly there immutable event data hasn't really had much of a home at all beyond data warehousing.

This is an excellent point, there IS a fundamental tradeoff between latency and throughput. Computers are much better a processing larger chunks of data linearly rather than small bits of data. However the question to ask is how much data do you have to batch together to get good throughput? There are diminishing returns and you stop getting much benefit after about 1MB in my experience. So this line of reasoning will not get you to additional latency of more than a few seconds, it definitely won't get you to a 24 hour batch data cycle.

Actually because we think this tradeoff is important we allow consumers from Kafka to specify how much data they want accumulated on the server before the server should complete their fetch request. The default is 1 byte, which means respond as soon as there is anything new for me, which optimizes latency. However setting this to a higher number will reduce round-trips and optimize throughput by avoiding lots of small fetches. This allows you to trade a small amount of latency for throughput. (In either case you can bound the waiting with a timeout so the latency is never worse than some maximum delay even if sufficient data hasn't arrived).

In any case when doing re-processing, as described in this article, there will always be lots of data accumulated and you will always fetch chunks of the maximum size you have configured (say 1MB). So in the reprocessing case you always get the "batch"-like throughput.