From e1f8aa740288f13532d47417cfaea460b0937080 Mon Sep 17 00:00:00 2001 From: Ankur Mhatre Date: Tue, 1 Sep 2026 23:27:08 +0530 Subject: [PATCH] Add the broker-comparison module Kafka, RabbitMQ and Pulsar measured side by side on ordering, replay and consumer scaling by driving the three client libraries directly, plus an operational-footprint measurement of each broker's own distribution. Nine tests, three brokers started without Docker, and every number in the documentation regenerated by scripts/run-all.sh. --- README.md | 16 +- broker-comparison/README.md | 80 +++++ broker-comparison/docs/01-three-models.md | 49 +++ broker-comparison/docs/02-ordering.md | 66 ++++ broker-comparison/docs/03-replay.md | 59 ++++ broker-comparison/docs/04-consumer-scaling.md | 80 +++++ .../docs/05-operational-footprint.md | 55 ++++ broker-comparison/docs/06-what-spring-adds.md | 52 +++ .../docs/07-the-decision-table.md | 59 ++++ broker-comparison/docs/output/footprint.txt | 40 +++ .../docs/output/kafka-consumer-scaling.txt | 16 + .../docs/output/kafka-ordering.txt | 30 ++ .../docs/output/kafka-replay.txt | 10 + .../docs/output/pulsar-consumer-scaling.txt | 30 ++ .../docs/output/pulsar-ordering.txt | 24 ++ .../docs/output/pulsar-replay.txt | 14 + .../docs/output/rabbit-consumer-scaling.txt | 34 ++ .../docs/output/rabbit-ordering.txt | 18 ++ .../docs/output/rabbit-replay.txt | 12 + broker-comparison/docs/output/tests.txt | 12 + broker-comparison/pom.xml | 66 ++++ broker-comparison/scripts/footprint.sh | 73 +++++ broker-comparison/scripts/pulsar-broker.sh | 25 ++ broker-comparison/scripts/rabbit-broker.sh | 31 ++ broker-comparison/scripts/run-all.sh | 37 +++ .../brokers/BrokerComparisonApplication.java | 15 + .../test/java/com/ankurm/brokers/Capture.java | 23 ++ .../ankurm/brokers/KafkaComparisonTest.java | 232 ++++++++++++++ .../ankurm/brokers/PulsarComparisonTest.java | 298 ++++++++++++++++++ .../ankurm/brokers/RabbitComparisonTest.java | 286 +++++++++++++++++ 30 files changed, 1835 insertions(+), 7 deletions(-) create mode 100644 broker-comparison/README.md create mode 100644 broker-comparison/docs/01-three-models.md create mode 100644 broker-comparison/docs/02-ordering.md create mode 100644 broker-comparison/docs/03-replay.md create mode 100644 broker-comparison/docs/04-consumer-scaling.md create mode 100644 broker-comparison/docs/05-operational-footprint.md create mode 100644 broker-comparison/docs/06-what-spring-adds.md create mode 100644 broker-comparison/docs/07-the-decision-table.md create mode 100644 broker-comparison/docs/output/footprint.txt create mode 100644 broker-comparison/docs/output/kafka-consumer-scaling.txt create mode 100644 broker-comparison/docs/output/kafka-ordering.txt create mode 100644 broker-comparison/docs/output/kafka-replay.txt create mode 100644 broker-comparison/docs/output/pulsar-consumer-scaling.txt create mode 100644 broker-comparison/docs/output/pulsar-ordering.txt create mode 100644 broker-comparison/docs/output/pulsar-replay.txt create mode 100644 broker-comparison/docs/output/rabbit-consumer-scaling.txt create mode 100644 broker-comparison/docs/output/rabbit-ordering.txt create mode 100644 broker-comparison/docs/output/rabbit-replay.txt create mode 100644 broker-comparison/docs/output/tests.txt create mode 100644 broker-comparison/pom.xml create mode 100755 broker-comparison/scripts/footprint.sh create mode 100755 broker-comparison/scripts/pulsar-broker.sh create mode 100755 broker-comparison/scripts/rabbit-broker.sh create mode 100755 broker-comparison/scripts/run-all.sh create mode 100644 broker-comparison/src/main/java/com/ankurm/brokers/BrokerComparisonApplication.java create mode 100644 broker-comparison/src/test/java/com/ankurm/brokers/Capture.java create mode 100644 broker-comparison/src/test/java/com/ankurm/brokers/KafkaComparisonTest.java create mode 100644 broker-comparison/src/test/java/com/ankurm/brokers/PulsarComparisonTest.java create mode 100644 broker-comparison/src/test/java/com/ankurm/brokers/RabbitComparisonTest.java diff --git a/README.md b/README.md index ad6412c..cfc0b9c 100644 --- a/README.md +++ b/README.md @@ -10,12 +10,14 @@ module's `scripts/run-all.sh`, never typed by hand. | [`kafka-basics/`](kafka-basics/README.md) | [Spring Boot 4.1 and Apache Kafka: Producer, Consumer and Serialisation from Scratch](https://ankurm.com/spring-boot-4-1-kafka-producer-consumer-serialisation/) | The on-ramp: what the starter gives you, the two Jackson serializer families, where a key lands and why, and which defaults are Kafka's rather than Spring's | | [`kafka-error-handling/`](kafka-error-handling/README.md) | [Kafka Error Handling with Spring Kafka 4.1: DLT, Retry Topics and Poison Pills](https://ankurm.com/spring-kafka-4-1-error-handling-dlt-retry-topics/) | What the default error handler really does, why a poison pill stops a partition, and what blocking retries cost that retry topics do not | | [`rabbitmq/`](rabbitmq/README.md) | [Spring Boot and RabbitMQ: Exchanges, Queues, Bindings and a Working Dead-Letter Queue](https://ankurm.com/spring-boot-rabbitmq-exchanges-dead-letter-queue/) | All four exchange types against a real broker, manual acknowledgement, and a dead-letter path exercised through both rejection and TTL expiry | +| [`broker-comparison/`](broker-comparison/README.md) | [Kafka vs RabbitMQ vs Pulsar for Java Teams: A Decision Framework with Benchmarks](https://ankurm.com/kafka-vs-rabbitmq-vs-pulsar-java-decision-framework/) | The three brokers measured side by side on ordering, replay, consumer scaling and operational footprint, ending in a decision table where every row has a transcript behind it | -The two brokers make an instructive pair. Kafka's consumer holds an offset and the broker -remembers nothing about individual records; RabbitMQ's broker owns the message until it is -acknowledged and can route, expire and dead-letter it on its own. Almost every difference in how -you handle failure follows from that one sentence — which is why Kafka needs a retry topic -to do what RabbitMQ does with a queue argument. +The brokers make an instructive set. Kafka's consumer holds an offset and the broker remembers +nothing about individual records; RabbitMQ's broker owns the message until it is acknowledged and +can route, expire and dead-letter it on its own; Pulsar keeps the log but lets each subscriber +choose how it is read. Almost every difference in how you handle failure follows from those three +sentences — which is why Kafka needs a retry topic to do what RabbitMQ does with a queue +argument, and why `broker-comparison` is mostly an argument about where a message lives. They are meant to be read in order. `kafka-basics` establishes that the default acknowledgement mode is `BATCH` and therefore that delivery is at-least-once; everything the error-handling @@ -29,8 +31,8 @@ Central and from Boot's own `spring-boot-dependencies` POM, rather than from rel announcements. Every module runs its broker without Docker, so the transcripts can be regenerated on any -machine with a JDK: Kafka via the in-process KRaft broker in `spring-kafka-test`, RabbitMQ via a -real broker started by the module's own scripts. Each module also ships a Testcontainers +machine with a JDK: Kafka via the in-process KRaft broker in `spring-kafka-test`, RabbitMQ and +Pulsar via real brokers started by the modules' own scripts. Each module also ships a Testcontainers configuration for the cases where the deployed image is what you need to test against. ## Licence diff --git a/broker-comparison/README.md b/broker-comparison/README.md new file mode 100644 index 0000000..e953b36 --- /dev/null +++ b/broker-comparison/README.md @@ -0,0 +1,80 @@ +# broker-comparison + +Companion project for **Kafka vs RabbitMQ vs Pulsar for Java Teams: A Decision Framework with +Benchmarks** on [ankurm.com](https://ankurm.com). + +Three brokers, three questions, one set of measurements. Ordering, replay and consumer scaling are +asked identically of all three by driving the client libraries directly, so that what is compared +is the brokers' delivery models rather than three sets of Spring defaults. What each Spring +integration adds on top is [chapter 6](docs/06-what-spring-adds.md). + +## Verified stack + +| Component | Version | Source of the number | +|---|---|---| +| JDK (tests) | 25.0.4.1+1 (Temurin) | `java -version` | +| JDK (brokers) | 21.0.12.1+1 (Temurin) | `java -version` | +| Spring Boot | 4.1.1 | `maven-metadata.xml` on Maven Central | +| kafka-clients | 4.2.1 | `spring-boot-dependencies-4.1.1.pom` | +| amqp-client | 5.30.0 | `spring-boot-dependencies-4.1.1.pom` | +| pulsar-client | 4.2.4 | `spring-boot-dependencies-4.1.1.pom` | +| Kafka broker | 4.2.1 (KRaft) | `kafka_2.13-4.2.1.tgz` | +| RabbitMQ broker | 3.10.25 | generic-unix tarball | +| Pulsar broker | 4.2.4 standalone | `apache-pulsar-4.2.4-bin.tar.gz` | + +RabbitMQ is pinned at 3.10.25 because the box that produced these transcripts has Erlang 24; +3.11 and later need Erlang 25. The AMQP 0-9-1 semantics measured here are unchanged in 4.x. + +## Quickstart + +The Kafka measurements need no external broker — `spring-kafka-test` starts a real KRaft broker +in-process. The other two need a broker each, and the scripts start both without Docker and +without root. + +```bash +mvn -Dgroups=kafka test + +ERL_ROOT=... RABBITMQ_HOME=... scripts/rabbit-broker.sh +mvn -Dgroups=rabbit test + +PULSAR_HOME=... scripts/pulsar-broker.sh +mvn -Dgroups=pulsar test + +scripts/footprint.sh kafka|rabbit|pulsar # the operational numbers +scripts/run-all.sh # regenerates every docs/output/ file +``` + +## Documentation + +| Chapter | What it settles | +|---|---| +| [01 Three models](docs/01-three-models.md) | Where a message lives, and why everything else follows | +| [02 Ordering](docs/02-ordering.md) | What survives the second consumer, in all three | +| [03 Replay](docs/03-replay.md) | Reading it twice, and the one broker that cannot | +| [04 Consumer scaling](docs/04-consumer-scaling.md) | The partition ceiling, and the client buffer that defeats fan-out in the other two | +| [05 Operational footprint](docs/05-operational-footprint.md) | Startup, memory, ports, processes, configuration surface | +| [06 What Spring adds](docs/06-what-spring-adds.md) | The three integrations, and the Boot 4 starter trap | +| [07 The decision table](docs/07-the-decision-table.md) | Every row backed by a measurement, and how to choose | + +## Captured output + +| File | What it shows | +|---|---| +| [`kafka-ordering.txt`](docs/output/kafka-ordering.txt) | Per-key order kept, global order not — and why A, B and C were the wrong keys to demonstrate it | +| [`kafka-replay.txt`](docs/output/kafka-replay.txt) | The same 12 records read three times | +| [`kafka-consumer-scaling.txt`](docs/output/kafka-consumer-scaling.txt) | 5 consumers, 3 partitions, 2 idle | +| [`rabbit-ordering.txt`](docs/output/rabbit-ordering.txt) | Message 1 finishing tenth | +| [`rabbit-replay.txt`](docs/output/rabbit-replay.txt) | 12 messages, then 0, then an empty queue | +| [`rabbit-consumer-scaling.txt`](docs/output/rabbit-consumer-scaling.txt) | 40/0/0/0/0 without `basicQos`, 8 each with it | +| [`pulsar-ordering.txt`](docs/output/pulsar-ordering.txt) | `Shared` spreading a key, `Key_Shared` pinning it | +| [`pulsar-replay.txt`](docs/output/pulsar-replay.txt) | `seek(earliest)` and a new subscription | +| [`pulsar-consumer-scaling.txt`](docs/output/pulsar-consumer-scaling.txt) | The 1000-message receiver queue defeating fan-out | +| [`footprint.txt`](docs/output/footprint.txt) | Startup, memory, ports and config surface for all three | +| [`tests.txt`](docs/output/tests.txt) | The nine tests behind all of the above | + +## What is not measured here + +Throughput. A messages-per-second number from one 2-core container says nothing about any of +these brokers, and publishing one would be worse than publishing nothing. The measurements above +are structural: they hold on any hardware, because they are properties of the delivery models +rather than of the machine. diff --git a/broker-comparison/docs/01-three-models.md b/broker-comparison/docs/01-three-models.md new file mode 100644 index 0000000..636936d --- /dev/null +++ b/broker-comparison/docs/01-three-models.md @@ -0,0 +1,49 @@ +[README](../README.md) · next: [Ordering](02-ordering.md) + +# 1. Three storage models, and everything that follows from them + +Almost every difference between these brokers follows from one sentence about where a message +lives. + +- **Kafka** is an append-only log, partitioned. The broker keeps every record for the retention + period and remembers nothing about individual consumers except an offset. Reading does not + remove anything. +- **RabbitMQ** is a router with queues. The broker owns each message until a consumer + acknowledges it, and then deletes it. It can route, expire, and dead-letter on its own. +- **Pulsar** is a log too, but the cursor and the storage are separate services: brokers are + stateless and BookKeeper holds the data. Acknowledgement moves a cursor, and a message is + deleted once every subscription has passed it — unless a retention policy says otherwise. + +Read those three sentences again before reading any comparison table, including the one in +[chapter 7](07-the-decision-table.md). They predict most of it. + +| | Kafka | RabbitMQ | Pulsar | +|---|---|---|---| +| unit of parallelism | partition | queue | subscription | +| set when | topic is created | queue is declared | consumer subscribes | +| consuming | moves an offset | deletes the message | moves a cursor | +| a second reader | new consumer group | another queue, bound in advance | new subscription | + +The last row is the one people underestimate. In Kafka and Pulsar you can add a reader that sees +history you have already processed. In RabbitMQ you cannot add one after the fact at all — the +messages are gone — so the decision to have a second consumer has to be made *before* the +messages arrive. + +## What was measured, and how + +Three questions, asked identically of all three brokers by driving the client libraries directly +rather than through three different Spring abstractions: + +| Question | Chapter | +|---|---| +| What ordering survives when you add a second consumer? | [02](02-ordering.md) | +| Can you read the same message twice? | [03](03-replay.md) | +| How far do consumers scale, and what stops them? | [04](04-consumer-scaling.md) | +| What does each one cost to run? | [05](05-operational-footprint.md) | + +Versions are the ones a Spring Boot 4.1.1 application gets: kafka-clients 4.2.1, amqp-client +5.30.0, pulsar-client 4.2.4, all read from `spring-boot-dependencies-4.1.1.pom`. The brokers are +Kafka 4.2.1, RabbitMQ 3.10.25 and Pulsar 4.2.4, each started from its own distribution by a script +in `scripts/`. + +next: [Ordering](02-ordering.md) diff --git a/broker-comparison/docs/02-ordering.md b/broker-comparison/docs/02-ordering.md new file mode 100644 index 0000000..0e4289d --- /dev/null +++ b/broker-comparison/docs/02-ordering.md @@ -0,0 +1,66 @@ +prev: [Three models](01-three-models.md) · [README](../README.md) · next: [Replay](03-replay.md) + +# 2. Ordering + +Every broker here is FIFO with one producer and one consumer. The question that matters is what +survives the second consumer, because that is the first thing you add. + +## Kafka: ordered within a partition, and a key picks the partition + +Twelve records, three keys, three partitions +([`docs/output/kafka-ordering.txt`](output/kafka-ordering.txt)): + +``` +F -> partition [2] values [2, 5, 8, 11] +A -> partition [1] values [1, 4, 7, 10] +D -> partition [0] values [3, 6, 9, 12] +``` + +Each key's values come back in the order they were produced. The global sequence does not: the +consumer drains one partition's buffer before moving to the next, so the delivered order is +2, 5, 8, 11, 1, 4, 7, 10, 3, 6, 9, 12. + +That is the whole Kafka ordering guarantee, and it is usually enough, because "in order" nearly +always means "in order per customer / per account / per device" rather than globally. + +**A detail worth stealing.** The keys in that transcript are D, A and F, not A, B and C. With +three partitions, murmur2 sends A, B *and* C all to partition 1. Picking three obvious keys to +demonstrate partitioning would have produced a transcript in which global order was accidentally +preserved. Before you conclude that your keys spread evenly, compute +`Utils.toPositive(Utils.murmur2(key)) % partitions` for the ones you actually use. + +## RabbitMQ: FIFO per queue, and no key at all + +One consumer sees the queue in order. Two consumers do not +([`docs/output/rabbit-ordering.txt`](output/rabbit-ordering.txt)): + +``` +one queue, one consumer : [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] +one queue, two consumers : [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 1, 12] +``` + +Message 1 went to the slower consumer and finished tenth. There is no key and no partition, so +there is nothing to pin related messages to one consumer. Ordering across related messages in +RabbitMQ means one queue and one consumer — which means no horizontal scaling of that queue — +or the consistent-hash exchange, which is a plugin and effectively reintroduces partitions by +hand. + +## Pulsar: the subscription type decides, per subscriber + +Two consumers on a `Shared` subscription and two on `Key_Shared`, same topic, same messages +([`docs/output/pulsar-ordering.txt`](output/pulsar-ordering.txt)): + +``` +Shared A -> [consumer-1, consumer-2] B -> [consumer-1, consumer-2] C -> [consumer-1, consumer-2] +Key_Shared A -> [consumer-2] B -> [consumer-1] C -> [consumer-2] +``` + +Under `Shared`, every key was handled by both consumers, so two messages with the same key can be +in flight at once and no ordering claim survives. Under `Key_Shared`, each key went to exactly one +consumer — Kafka's guarantee, but chosen by the subscriber rather than fixed by a partition count +someone picked when the topic was created, and recomputed as consumers join and leave. + +That last property is the strongest single argument for Pulsar: two teams reading the same topic +can make different ordering-versus-throughput trades without negotiating. + +next: [Replay](03-replay.md) diff --git a/broker-comparison/docs/03-replay.md b/broker-comparison/docs/03-replay.md new file mode 100644 index 0000000..94db78d --- /dev/null +++ b/broker-comparison/docs/03-replay.md @@ -0,0 +1,59 @@ +prev: [Ordering](02-ordering.md) · [README](../README.md) · next: [Consumer scaling](04-consumer-scaling.md) + +# 3. Replay + +"Can I read that again?" is asked after a bad deploy, and by then it is too late to change +brokers. + +## Kafka + +``` +group replay-group-1, first read : 12 records +group replay-group-2, brand new group : 12 records +group replay-group-1, after seekToBeginning: 12 records +``` + +([`docs/output/kafka-replay.txt`](output/kafka-replay.txt)) Consuming moves an offset and removes +nothing. A new group, a reset offset and a `seekToBeginning` all read the same records, for as +long as retention keeps them. + +## RabbitMQ + +``` +first drain of the queue : 12 messages +second drain of the queue : 0 messages +queue depth afterwards : 0 +``` + +([`docs/output/rabbit-replay.txt`](output/rabbit-replay.txt)) Acknowledging deletes. There is no +offset to rewind and no second reader that can see what the first consumed. Reading a message +twice has to be arranged in advance — a second queue bound to the same exchange, or a copy written +somewhere else — and it cannot be arranged afterwards. + +This is not a defect. A router that deletes what it has delivered is much cheaper to operate than +a log, and most work queues genuinely do not need history. It only becomes a defect at the moment +you need history and do not have it. + +RabbitMQ streams (3.9+) are a separate, log-shaped feature that does support replay. They are a +different thing living in the same broker, with their own client protocol and their own semantics +— worth knowing about, and not what you get from `queueDeclare`. + +## Pulsar + +``` +subscription replay-sub, first read : 12 messages +subscription replay-sub, after seek(earliest): 12 messages +subscription replay-sub-2, brand new : 12 messages +``` + +([`docs/output/pulsar-replay.txt`](output/pulsar-replay.txt)) `seek(MessageId)` and +`seek(timestamp)` rewind a live subscription; a new subscription starting from `Earliest` reads +everything still stored. + +The difference from Kafka is the **default**, and it is the one that surprises people: Kafka keeps +a record for the retention period regardless of who read it, while Pulsar deletes a message once +every subscription has acknowledged it, unless a retention policy on the namespace says otherwise. +A Pulsar namespace with the default retention and one well-behaved subscription keeps nothing — +so the replay you are counting on requires a policy you have to set. + +next: [Consumer scaling](04-consumer-scaling.md) diff --git a/broker-comparison/docs/04-consumer-scaling.md b/broker-comparison/docs/04-consumer-scaling.md new file mode 100644 index 0000000..a6d1f02 --- /dev/null +++ b/broker-comparison/docs/04-consumer-scaling.md @@ -0,0 +1,80 @@ +prev: [Replay](03-replay.md) · [README](../README.md) · next: [Operational footprint](05-operational-footprint.md) + +# 4. Consumer scaling + +The question is what happens when you add the fifth consumer. + +## Kafka: partitions are a hard ceiling + +Five consumers in one group on a three-partition topic +([`docs/output/kafka-consumer-scaling.txt`](output/kafka-consumer-scaling.txt)): + +``` +consumer-1 -> partitions [0] +consumer-2 -> partitions [1] +consumer-3 -> partitions [2] +consumer-4 -> no partitions (idle) +consumer-5 -> no partitions (idle) +``` + +A partition belongs to at most one consumer in a group, so partition count is the ceiling. Two of +the five processes are running, connected, healthy, and doing nothing at all — and they will keep +reporting healthy forever. + +Raising the partition count later is possible and is not free: it changes which partition a key +maps to, so the per-key ordering guarantee is broken across the change for every key that moves. +In practice the partition count is a capacity decision made at design time, on incomplete +information, that you then live with. + +## RabbitMQ: no ceiling, but prefetch decides whether it is real + +Five consumers on one queue, forty messages +([`docs/output/rabbit-consumer-scaling.txt`](output/rabbit-consumer-scaling.txt)): + +``` +no basicQos at all (unlimited prefetch, the AMQP default): + consumer-1 -> 40 consumer-2..5 -> 0 + +basicQos(1): + consumer-1..5 -> 8 each +``` + +There is no structural ceiling — but with the AMQP default the broker pushes as many messages as +a consumer will accept, so the first consumer to connect can be handed the entire backlog while +four idle processes wait. `basicQos` is not a tuning knob to postpone; it is what makes the +fan-out exist. + +Spring AMQP sets it for you: `AbstractMessageListenerContainer.DEFAULT_PREFETCH_COUNT` is 250. +Better than unlimited, and still enough to concentrate any backlog smaller than 250 messages on +one consumer. + +## Pulsar: no ceiling either, and the same trap under a different name + +Five consumers, one *non-partitioned* topic, forty messages +([`docs/output/pulsar-consumer-scaling.txt`](output/pulsar-consumer-scaling.txt)): + +``` +receiverQueueSize default (1000): + one or two consumers take everything; the rest receive nothing + +receiverQueueSize(1): + consumer-1..5 -> 8 each +``` + +The topic has no partitions and five consumers still share the work, which in Kafka would require +at least five partitions decided in advance. But the default receiver queue is a thousand +messages, so the first consumers to connect pull the whole backlog into their own buffers first. +Same failure as RabbitMQ's unlimited prefetch, four times the default. + +## The shape of the answer + +- Kafka gives you ordering per key and pays for it with a fixed parallelism ceiling. +- RabbitMQ gives you unlimited competing consumers and pays for it with no ordering across them. +- Pulsar lets each subscription choose, which is genuinely the best of both — at the operational + cost measured in [the next chapter](05-operational-footprint.md). + +And all three brokers have a client-side buffer that will quietly defeat your fan-out if you leave +it at the default and your backlog is small. That is the one finding here that applies to +whichever you pick. + +next: [Operational footprint](05-operational-footprint.md) diff --git a/broker-comparison/docs/05-operational-footprint.md b/broker-comparison/docs/05-operational-footprint.md new file mode 100644 index 0000000..759ec68 --- /dev/null +++ b/broker-comparison/docs/05-operational-footprint.md @@ -0,0 +1,55 @@ +prev: [Consumer scaling](04-consumer-scaling.md) · [README](../README.md) · next: [What Spring adds](06-what-spring-adds.md) + +# 5. Operational footprint + +"Ops burden" cannot be benchmarked. The number of moving parts, the memory floor and the size of +the configuration surface can be, and they are what people are actually asking about. + +Each broker started from its shipped distribution with default configuration and a 700 MB heap +cap, on one 2-core / 3.8 GB Linux box, Temurin JDK 21 +([`docs/output/footprint.txt`](output/footprint.txt)): + +| | Kafka 4.2.1 | RabbitMQ 3.10.25 | Pulsar 4.2.4 standalone | +|---|---|---|---| +| launch to first accepted connection | 10.0 s | 15.1 s | 20.8 s | +| resident memory at idle | 333 MB | 115 MB | 610 MB | +| server processes | 1 | 1 (+ `epmd`) | 1 | +| listening ports | 9092, 9093 | 5672, 4369, 25672 | 6650, 8080, 2181 | +| unpacked distribution | 135 MB | 26 MB | 344 MB | +| settings in the shipped default config | 24 | **0** | 357 | + +Read those numbers as orders of magnitude, not as a benchmark: one sample, one small machine. + +Three things they say clearly. + +**RabbitMQ ships no configuration file at all.** The boot log records `Config file(s): (none)`; +`rabbitmq.conf.example` is entirely commented out. Everything works out of the box, and the +smallest resident footprint here is the one that is not a JVM. Against that, `epmd` on 4369 and +the inter-node port on 25672 are the Erlang distribution, which is also how clustering works and +how clustering goes wrong. + +**Pulsar standalone is three systems in one process.** Port 2181 is ZooKeeper and the on-disk +data is BookKeeper's; the standalone distribution hides that behind one command. A real +deployment does not: you operate ZooKeeper (or, from Pulsar 3.x, an alternative metadata store), +BookKeeper bookies and Pulsar brokers as three tiers with three scaling stories. The 357-setting +`standalone.conf` is the honest signal here — that is the configuration surface, and running it +seriously means learning most of it. + +**Kafka since KRaft is genuinely simpler than it was.** One process, two ports, 24 settings, no +ZooKeeper. The old "Kafka means also running ZooKeeper" objection is a version behind; anyone +comparing on that basis is comparing to 2022. + +## What the numbers do not include + +The footprint above is one node at idle. The thing that actually determines ops burden is what +happens at three in the morning: + +- **Kafka**: rebalances, consumer lag as the primary signal, partition-count decisions you cannot + cleanly reverse, and a broad ecosystem of tools that assume Kafka. +- **RabbitMQ**: the management plugin is genuinely good, queue depth is a direct and obvious + signal, and the hard problem is network partitions in a cluster — which the Erlang distribution + makes fast to detect and awkward to resolve. +- **Pulsar**: the fewest people on your team will have run it. That is not a technical property + and it is usually the deciding one. + +next: [What Spring adds](06-what-spring-adds.md) diff --git a/broker-comparison/docs/06-what-spring-adds.md b/broker-comparison/docs/06-what-spring-adds.md new file mode 100644 index 0000000..bd2dd86 --- /dev/null +++ b/broker-comparison/docs/06-what-spring-adds.md @@ -0,0 +1,52 @@ +prev: [Operational footprint](05-operational-footprint.md) · [README](../README.md) · next: [The decision table](07-the-decision-table.md) + +# 6. What Spring adds, and what Boot 4 no longer gives you for free + +The measurements in this module drive the client libraries directly, so that what is being +compared is the brokers rather than three sets of Spring defaults. In an application you would use +the Spring integrations, and they are not equivalent to one another. + +| | Kafka | RabbitMQ | Pulsar | +|---|---|---|---| +| project | Spring for Apache Kafka 4.1.1 | Spring AMQP 4.1.1 | Spring for Apache Pulsar 2.0.7 | +| starter | `spring-boot-starter-kafka` | `spring-boot-starter-amqp` | `spring-boot-starter-pulsar` | +| listener | `@KafkaListener` | `@RabbitListener` | `@PulsarListener` | +| template | `KafkaTemplate` | `RabbitTemplate` | `PulsarTemplate` | +| retry / DLQ | `DefaultErrorHandler`, `@RetryableTopic` | dead-letter exchange, container error handler | `DeadLetterPolicy` on the listener | + +All three are Boot-managed at 4.1.1, so you do not pin their versions. + +## The Boot 4 trap that applies to all three + +**In Spring Boot 4, depending on a messaging library directly rather than through its Boot starter +means you have no auto-configuration.** The auto-configuration classes moved out of +`spring-boot-autoconfigure` into per-technology modules that only the starters bring: + +| depending on | what you lose | how it presents | +|---|---|---| +| `org.springframework.kafka:spring-kafka` | `spring-boot-kafka` | `No qualifying bean of type KafkaTemplate<...>` — the context starts fine | +| `org.springframework.amqp:spring-rabbit` | `spring-boot-amqp` | no `RabbitTemplate`, no `RabbitAdmin` | +| `org.springframework.pulsar:spring-pulsar` | `spring-boot-pulsar` | no `PulsarTemplate` | + +Every Boot 3 tutorial gets this wrong now, and the symptom is a missing bean rather than anything +that names the cause. Use the starters. + +## The Jackson fork, in both Kafka and RabbitMQ + +Boot 4 moved to Jackson 3 (`tools.jackson`), and both messaging projects ship converters for +both generations. **A `2` in the class name means the previous Jackson** — which is the opposite +of the convention you would guess: + +- Spring Kafka: `JsonSerializer`/`JsonDeserializer` are Jackson 2; `JacksonJsonSerializer`, + `JacksonJsonDeserializer` and `JacksonJsonSerde` are Jackson 3. +- Spring AMQP: `Jackson2JsonMessageConverter` is Jackson 2; `JacksonJsonMessageConverter` is + Jackson 3. + +Picking the wrong one gives you `SerializationException: Can't serialize data`, whose cause is +`Java 8 date/time type java.time.Instant not supported by default` — a message about date types +for what is really a wrong-library problem. + +The [kafka-basics](../kafka-basics/README.md) and [rabbitmq](../rabbitmq/README.md) modules in +this repository cover both stacks in detail. + +next: [The decision table](07-the-decision-table.md) diff --git a/broker-comparison/docs/07-the-decision-table.md b/broker-comparison/docs/07-the-decision-table.md new file mode 100644 index 0000000..c0d38ac --- /dev/null +++ b/broker-comparison/docs/07-the-decision-table.md @@ -0,0 +1,59 @@ +prev: [What Spring adds](06-what-spring-adds.md) · [README](../README.md) + +# 7. The decision table + +Every row below is either a measurement in `docs/output/` or a statement about operating the +thing. Nothing here is a vendor claim. + +| | Kafka 4.2.1 | RabbitMQ 3.10.25 | Pulsar 4.2.4 | +|---|---|---|---| +| ordering with one consumer | FIFO per partition | FIFO per queue | FIFO | +| ordering with many consumers | per key, always | **none** | per key with `Key_Shared`, none with `Shared` | +| who chooses that | topic design | plugin, or one consumer | each subscription, independently | +| replay after the fact | yes, within retention | **no** | yes, if retention is configured | +| second independent reader | new consumer group, any time | must be arranged in advance | new subscription, any time | +| consumer parallelism ceiling | partition count | none | none | +| changing that ceiling | repartition; breaks key→partition | nothing to change | nothing to change | +| the client-side trap | none by default | unlimited prefetch (Spring: 250) | `receiverQueueSize` 1000 | +| memory at idle | 333 MB | **115 MB** | 610 MB | +| processes to operate | 1 (KRaft) | 1 + `epmd` | broker + BookKeeper + metadata store | +| shipped default settings | 24 | **0** | 357 | +| Spring project maturity | very high | very high | good, much smaller community | + +## Choosing + +**Choose RabbitMQ** when the work is a queue of tasks: each message is an instruction, order +between instructions does not matter, and once it is done it is done. It is the smallest thing +that works, it has the best out-of-the-box operability of the three, and queue depth is a metric +anyone can interpret. Choose it in the knowledge that you are giving up replay permanently — +adding it later is a broker migration, not a configuration change. + +**Choose Kafka** when the messages are events other people will want to read: when more than one +consumer will exist, when someone will need to reprocess history after a bug, or when ordering per +entity is part of the contract. Pay for it with a partition count decided too early, consumers +that scale only as far as that number, and an operational model where lag is the thing you watch. +KRaft has removed the old ZooKeeper objection. + +**Choose Pulsar** when you genuinely need what neither of the others gives you: per-subscription +choice of ordering versus fan-out, consumer counts not bounded by a number chosen at topic +creation, or multi-tenancy with real isolation. It is the most capable design here. It is also +three systems, 357 settings, five times RabbitMQ's memory floor, and the one your team has least +experience with — and that last point decides more incidents than the first three prevent. + +**If the honest answer is "we do not know yet"**, that argues for Kafka, on grounds that have +nothing to do with the technology: you can hire for it, your monitoring vendor supports it, and +the failure modes are documented by thousands of people who hit them first. That is a real +engineering argument and it is fine to make it out loud. + +## The question that dissolves the choice + +A surprising number of these decisions are made for a system that has one producer, one consumer +and fewer than a hundred messages a second. At that volume all three brokers work, none of the +measurements above will ever be reached, and the decision is entirely about what your team can +operate at three in the morning. + +The measurements matter when you can name which row you are relying on. If you cannot, you are +picking an operational burden, not a broker — and the cheapest one to operate is the one in the +`115 MB` cell. + +[README](../README.md) diff --git a/broker-comparison/docs/output/footprint.txt b/broker-comparison/docs/output/footprint.txt new file mode 100644 index 0000000..6f979aa --- /dev/null +++ b/broker-comparison/docs/output/footprint.txt @@ -0,0 +1,40 @@ +== Operational footprint, measured on one 2-core / 3.8 GB Linux box == + +Each broker started from its shipped distribution with default configuration and a +700 MB heap cap, on Temurin JDK 21. Times are one sample on a small box: treat them as +orders of magnitude, not as a benchmark. + +broker : kafka +time from launch to first + accepted connection : 10035 ms +resident memory at idle : 333 MB +server processes : 1 +listening ports : 9092 (broker), 9093 (controller) +unpacked distribution size : 135 MB +settings in shipped default + configuration file : 24 (server.properties) + +broker : rabbit +time from launch to first + accepted connection : 15053 ms +resident memory at idle : 115 MB +server processes : 1 +listening ports : 5672 (AMQP), 4369 (epmd), 25672 (inter-node) +unpacked distribution size : 26 MB +settings in shipped default + configuration file : 0 (rabbitmq.conf.example) +epmd processes : 1 +NOTE: RabbitMQ ships no configuration file at all. rabbitmq.conf.example is + entirely commented out and the boot log records "Config file(s): (none)". + Everything in that file is documentation, not a default. + +broker : pulsar +time from launch to first + accepted connection : 20801 ms +resident memory at idle : 610 MB +server processes : 1 +listening ports : 6650 (binary), 8080 (admin/REST), 2181 (ZooKeeper) +unpacked distribution size : 344 MB +settings in shipped default + configuration file : 357 (standalone.conf) + diff --git a/broker-comparison/docs/output/kafka-consumer-scaling.txt b/broker-comparison/docs/output/kafka-consumer-scaling.txt new file mode 100644 index 0000000..1944957 --- /dev/null +++ b/broker-comparison/docs/output/kafka-consumer-scaling.txt @@ -0,0 +1,16 @@ +== Kafka: partitions are the ceiling on consumer parallelism == + +topic 'orders-scaling', 3 partitions, 5 consumers in one group + + consumer-1 -> partitions [0] + consumer-2 -> partitions [1] + consumer-3 -> partitions [2] + consumer-4 -> no partitions (idle) + consumer-5 -> no partitions (idle) + +consumers with no partitions: 2 + +A partition is assigned to at most one consumer in a group, so the number +of partitions is a hard ceiling on consumer parallelism. Adding consumers +beyond it adds idle processes, not throughput. + diff --git a/broker-comparison/docs/output/kafka-ordering.txt b/broker-comparison/docs/output/kafka-ordering.txt new file mode 100644 index 0000000..b309d5c --- /dev/null +++ b/broker-comparison/docs/output/kafka-ordering.txt @@ -0,0 +1,30 @@ +== Kafka: ordering is per partition == + +produced 12 records, keys D/A/F round-robin, values 1..12 in order +(keys chosen so that murmur2 spreads them: D->0, A->1, F->2. A, B and C + all hash to partition 1 with three partitions, which is worth knowing + before you decide your keys are well distributed.) + +consumed in this order: + F=2@p2 + F=5@p2 + F=8@p2 + F=11@p2 + A=1@p1 + A=4@p1 + A=7@p1 + A=10@p1 + D=3@p0 + D=6@p0 + D=9@p0 + D=12@p0 + +per key: + F -> partition [2] values [2, 5, 8, 11] + A -> partition [1] values [1, 4, 7, 10] + D -> partition [0] values [3, 6, 9, 12] + +Order is preserved within each key because a key hashes to one partition. +Across keys it is not: the values above are not 1..12 in order, because a +consumer drains one partition's buffer before the next. + diff --git a/broker-comparison/docs/output/kafka-replay.txt b/broker-comparison/docs/output/kafka-replay.txt new file mode 100644 index 0000000..cb7d2e5 --- /dev/null +++ b/broker-comparison/docs/output/kafka-replay.txt @@ -0,0 +1,10 @@ +== Kafka: the log is the storage == + +group replay-group-1, first read : 12 records +group replay-group-2, brand new group : 12 records +group replay-group-1, after seekToBeginning: 12 records + +Consuming does not remove anything. A consumer group is a cursor over a log +that the broker keeps until retention expires, so a new group, a reset +offset or a seek all read the same records again. + diff --git a/broker-comparison/docs/output/pulsar-consumer-scaling.txt b/broker-comparison/docs/output/pulsar-consumer-scaling.txt new file mode 100644 index 0000000..f1693b5 --- /dev/null +++ b/broker-comparison/docs/output/pulsar-consumer-scaling.txt @@ -0,0 +1,30 @@ +== Pulsar: no partition ceiling, but the receiver queue decides who gets the work == + +one non-partitioned topic, 40 messages, 5 consumers, Shared subscription + +receiverQueueSize left at the default (1000): + consumer-1 -> 40 messages + consumer-2 -> 0 messages + consumer-3 -> 0 messages + consumer-4 -> 0 messages + consumer-5 -> 0 messages + consumers that received nothing : 4 + +receiverQueueSize(1): + consumer-1 -> 8 messages + consumer-2 -> 8 messages + consumer-3 -> 8 messages + consumer-4 -> 8 messages + consumer-5 -> 8 messages + consumers that received nothing : 0 + +The topic has no partitions and five consumers can still share the work -- +in Kafka the same shape needs at least five partitions, chosen when the +topic was created. But the default receiver queue is 1000 messages, so the +first consumers to connect pull the whole 40-message backlog into their own +buffers before the rest ask for anything, and the subscription looks +broken. Which consumers win is a race and varies between runs -- one +consumer taking all forty, or two taking twenty each -- but the consumers +that lose it see nothing at all. This is the same trap as RabbitMQ's +unbounded prefetch, with a different name and a much larger default. + diff --git a/broker-comparison/docs/output/pulsar-ordering.txt b/broker-comparison/docs/output/pulsar-ordering.txt new file mode 100644 index 0000000..17c7048 --- /dev/null +++ b/broker-comparison/docs/output/pulsar-ordering.txt @@ -0,0 +1,24 @@ +== Pulsar: the subscription type decides == + +12 messages, keys A/B/C, values 1..12 in order, two consumers per +subscription, receiverQueueSize 1 so the first consumer cannot take the +whole backlog. + +Shared subscription, which consumers saw each key: + A -> [consumer-1, consumer-2] + B -> [consumer-1, consumer-2] + C -> [consumer-1, consumer-2] + +Key_Shared subscription, which consumers saw each key: + A -> [consumer-1] + B -> [consumer-1] + C -> [consumer-1] + +A Shared subscription round-robins individual messages, so messages with +the same key end up on different consumers and can be processed at the +same time: there is no per-key order left to speak of. Key_Shared hashes +the key to one consumer, which is Kafka's guarantee -- except that the +assignment belongs to the subscription and is recomputed as consumers come +and go, rather than being fixed by a partition count chosen when the topic +was created. + diff --git a/broker-comparison/docs/output/pulsar-replay.txt b/broker-comparison/docs/output/pulsar-replay.txt new file mode 100644 index 0000000..e000617 --- /dev/null +++ b/broker-comparison/docs/output/pulsar-replay.txt @@ -0,0 +1,14 @@ +== Pulsar: acknowledged, but still there == + +subscription replay-sub, first read : 12 messages +subscription replay-sub, after seek(earliest): 12 messages +subscription replay-sub-2, brand new : 12 messages + +Acknowledgement moves a cursor; the message itself lives in the managed +ledger. seek(MessageId) and seek(timestamp) rewind a live subscription, +which Kafka can also do by resetting offsets. What differs is the default: +Pulsar deletes a message once every subscription has acknowledged it, +unless a retention policy on the namespace says otherwise, whereas Kafka +keeps it for the retention period regardless of who read it. A Pulsar +topic with no retention policy and no subscriptions keeps nothing. + diff --git a/broker-comparison/docs/output/rabbit-consumer-scaling.txt b/broker-comparison/docs/output/rabbit-consumer-scaling.txt new file mode 100644 index 0000000..cfa28e7 --- /dev/null +++ b/broker-comparison/docs/output/rabbit-consumer-scaling.txt @@ -0,0 +1,34 @@ +== RabbitMQ: consumers scale, and prefetch decides whether they actually do == + +one queue, 40 messages, 5 consumers, each taking 10 ms per message + +no basicQos at all (unlimited prefetch, the AMQP default): + consumer-1 -> 40 messages + consumer-2 -> 0 messages + consumer-3 -> 0 messages + consumer-4 -> 0 messages + consumer-5 -> 0 messages + consumers that received nothing : 4 + +basicQos(1): + consumer-1 -> 8 messages + consumer-2 -> 9 messages + consumer-3 -> 8 messages + consumer-4 -> 8 messages + consumer-5 -> 7 messages + consumers that received nothing : 0 + +Every consumer on a queue competes for the same messages, so adding +consumers adds throughput and there is no structural ceiling of the kind +Kafka's partition count imposes. But with the AMQP default the broker +pushes as many messages as a consumer will take, so whichever consumer +connects first can be handed the entire backlog while the others sit idle. +basicQos is not a tuning knob you get to postpone; it is what makes the +fan-out real. Spring AMQP sets it for you -- +AbstractMessageListenerContainer.DEFAULT_PREFETCH_COUNT is 250 -- which is +better than unlimited and still large enough to concentrate a small +backlog on one consumer. + +The price of all this is the ordering measurement: there is no key, so +nothing constrains related messages to one consumer. + diff --git a/broker-comparison/docs/output/rabbit-ordering.txt b/broker-comparison/docs/output/rabbit-ordering.txt new file mode 100644 index 0000000..99fa97f --- /dev/null +++ b/broker-comparison/docs/output/rabbit-ordering.txt @@ -0,0 +1,18 @@ +== RabbitMQ: FIFO per queue, per consumer == + +one queue, one consumer, 12 messages + completion order : [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] + +one queue, two consumers, prefetch 1, consumer-1 slower than consumer-2 + completion order : [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 1, 12] + out-of-order steps: 1 + +A queue is FIFO and a single consumer sees it that way. The moment a second +consumer is added the broker hands the next message to whichever consumer +is free, so the order in which work finishes is no longer the order in +which it was published. There is no key: RabbitMQ has no notion of a +partition to which related messages could be pinned. Ordering across +related messages means one queue and one consumer, and therefore no +horizontal scaling for that queue -- or a consistent-hash exchange, which +is a plugin. + diff --git a/broker-comparison/docs/output/rabbit-replay.txt b/broker-comparison/docs/output/rabbit-replay.txt new file mode 100644 index 0000000..eae3fe9 --- /dev/null +++ b/broker-comparison/docs/output/rabbit-replay.txt @@ -0,0 +1,12 @@ +== RabbitMQ: there is nothing to replay == + +first drain of the queue : 12 messages +second drain of the queue : 0 messages +queue depth afterwards : 0 + +Acknowledging a message deletes it. The broker is a router with buffers, +not a log: there is no offset to rewind and no second reader that can see +what the first one consumed. Reading the same message twice means +arranging it in advance -- a second queue bound to the same exchange, or a +copy written somewhere else -- and it cannot be arranged after the fact. + diff --git a/broker-comparison/docs/output/tests.txt b/broker-comparison/docs/output/tests.txt new file mode 100644 index 0000000..909f897 --- /dev/null +++ b/broker-comparison/docs/output/tests.txt @@ -0,0 +1,12 @@ +[INFO] Running com.ankurm.brokers.KafkaComparisonTest +[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 26.88 s -- in com.ankurm.brokers.KafkaComparisonTest +[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0 +[INFO] BUILD SUCCESS +[INFO] Running com.ankurm.brokers.RabbitComparisonTest +[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 4.732 s -- in com.ankurm.brokers.RabbitComparisonTest +[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0 +[INFO] BUILD SUCCESS +[INFO] Running com.ankurm.brokers.PulsarComparisonTest +[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 55.00 s -- in com.ankurm.brokers.PulsarComparisonTest +[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0 +[INFO] BUILD SUCCESS diff --git a/broker-comparison/pom.xml b/broker-comparison/pom.xml new file mode 100644 index 0000000..1750730 --- /dev/null +++ b/broker-comparison/pom.xml @@ -0,0 +1,66 @@ + + 4.0.0 + + + + org.springframework.boot + spring-boot-starter-parent + 4.1.1 + + + + com.ankurm + broker-comparison + 1.0 + jar + + + 25 + UTF-8 + + + + + org.springframework.boot + spring-boot-starter + + + + + org.apache.kafka + kafka-clients + + + com.rabbitmq + amqp-client + + + org.apache.pulsar + pulsar-client + + + + org.springframework.boot + spring-boot-starter-test + test + + + + org.springframework.kafka + spring-kafka-test + test + + + + + + diff --git a/broker-comparison/scripts/footprint.sh b/broker-comparison/scripts/footprint.sh new file mode 100755 index 0000000..45d9976 --- /dev/null +++ b/broker-comparison/scripts/footprint.sh @@ -0,0 +1,73 @@ +#!/usr/bin/env bash +# Measures the operational footprint of one broker: how long it takes to accept a connection, +# how much memory it holds at idle, how many processes and listening ports it opens, and how +# large its shipped default configuration is. +# +# These are proxies, not a benchmark. "Ops burden" is not directly measurable, but the number of +# moving parts, the memory floor and the size of the configuration surface are, and they are what +# people are actually asking about when they ask which broker is heavier to run. +# +# scripts/footprint.sh kafka|rabbit|pulsar +set -euo pipefail +BROKER="${1:?kafka|rabbit|pulsar}" + +extra="" +port_open() { (echo > "/dev/tcp/127.0.0.1/$1") 2>/dev/null; } +rss_mb() { ps -eo rss,cmd | grep -F "$1" | grep -v grep | awk '{s+=$1} END {printf "%.0f", s/1024}'; } +procs() { ps -eo cmd | grep -F "$1" | grep -cv grep; } +settings() { grep -cE '^[a-zA-Z]' "$1" 2>/dev/null || echo 0; } + +start=$(date +%s%3N) +case "$BROKER" in + kafka) + : "${KAFKA_HOME:?set KAFKA_HOME}" + # Wipe whatever log.dirs points at, so the format below is not refused with + # "Invalid cluster.id ... Expected X, but read Y" from a previous run. + KAFKA_LOG_DIRS=$(grep -E '^log.dirs=' "$KAFKA_HOME/config/server.properties" | cut -d= -f2) + rm -rf ${KAFKA_LOG_DIRS//,/ } + "$KAFKA_HOME/bin/kafka-storage.sh" format --standalone -t "$("$KAFKA_HOME/bin/kafka-storage.sh" random-uuid)" \ + -c "$KAFKA_HOME/config/server.properties" --ignore-formatted >/dev/null + start=$(date +%s%3N) # after formatting: kafka-storage.sh is two more JVM starts and is a + # one-off, not part of what a restart costs + setsid nohup "$KAFKA_HOME/bin/kafka-server-start.sh" "$KAFKA_HOME/config/server.properties" \ + > /tmp/kafka-start.log 2>&1 < /dev/null & + for _ in $(seq 1 90); do port_open 9092 && break; sleep 1; done + ready=$(( $(date +%s%3N) - start )) + marker="kafka.Kafka"; ports="9092 (broker), 9093 (controller)"; conf="$KAFKA_HOME/config/server.properties" + dist="$KAFKA_HOME" + ;; + rabbit) + : "${RABBITMQ_HOME:?set RABBITMQ_HOME}"; : "${ERL_ROOT:?set ERL_ROOT}" + scripts/rabbit-broker.sh >/dev/null + ready=$(( $(date +%s%3N) - start )) + # The broker is an Erlang VM, so the process to measure is beam.smp, not anything named + # "rabbitmq". epmd is a second, tiny process and is counted separately below. + marker="beam.smp"; ports="5672 (AMQP), 4369 (epmd), 25672 (inter-node)" + conf="$RABBITMQ_HOME/etc/rabbitmq/rabbitmq.conf.example"; dist="$RABBITMQ_HOME" + extra="epmd processes : $(procs epmd) +NOTE: RabbitMQ ships no configuration file at all. rabbitmq.conf.example is + entirely commented out and the boot log records \"Config file(s): (none)\". + Everything in that file is documentation, not a default." + ;; + pulsar) + : "${PULSAR_HOME:?set PULSAR_HOME}" + scripts/pulsar-broker.sh >/dev/null + ready=$(( $(date +%s%3N) - start )) + marker="PulsarStandaloneStarter"; ports="6650 (binary), 8080 (admin/REST), 2181 (ZooKeeper)" + conf="$PULSAR_HOME/conf/standalone.conf"; dist="$PULSAR_HOME" + ;; + *) echo "unknown broker $BROKER" >&2; exit 2 ;; +esac + +cat <-bin directory. Pulsar 4.2 +# supports Java 17 and 21; the transcripts here were produced on Temurin 21. +set -eu +: "${PULSAR_HOME:?set PULSAR_HOME to an unpacked apache-pulsar-*-bin directory}" +export PULSAR_MEM="${PULSAR_MEM:--Xms384m -Xmx700m -XX:MaxDirectMemorySize=384m}" + +cd "$PULSAR_HOME" +setsid nohup bin/pulsar standalone -nss -nfw > /tmp/pulsar-start.log 2>&1 < /dev/null & +for _ in $(seq 1 90); do + if [ "$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8080/admin/v2/brokers/version)" = "200" ]; then + echo "broker ready on 6650 (admin 8080), version $(curl -s http://127.0.0.1:8080/admin/v2/brokers/version)" + exit 0 + fi + sleep 2 +done +echo "broker did not start; see /tmp/pulsar-start.log" >&2 +exit 1 diff --git a/broker-comparison/scripts/rabbit-broker.sh b/broker-comparison/scripts/rabbit-broker.sh new file mode 100755 index 0000000..7b662bb --- /dev/null +++ b/broker-comparison/scripts/rabbit-broker.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Start a real RabbitMQ broker with no Docker and no root. +# +# RabbitMQ is an Erlang application, so an Erlang runtime and epmd on the path are the whole +# dependency. Point ERL_ROOT at an Erlang installation and RABBITMQ_HOME at an unpacked +# rabbitmq-server-generic-unix tarball. Ubuntu 22.04 ships Erlang 24, whose newest compatible +# broker is 3.10.25; 3.11 and later need Erlang 25. +# +# The one non-obvious step is starting epmd yourself: rabbitmq-server's own attempt to start it +# fails in a container without a resolvable hostname, and the failure surfaces as a forty-line +# Erlang crash dump ending in {'EXIT',nodistribution}. +set -eu +: "${ERL_ROOT:?set ERL_ROOT to an Erlang installation directory}" +: "${RABBITMQ_HOME:?set RABBITMQ_HOME to an unpacked rabbitmq-server-generic-unix directory}" + +export PATH="$ERL_ROOT/bin:$ERL_ROOT/erts-"*/bin":$PATH" +export RABBITMQ_MNESIA_BASE="${RABBITMQ_MNESIA_BASE:-/tmp/rmq/data}" +export RABBITMQ_LOG_BASE="${RABBITMQ_LOG_BASE:-/tmp/rmq/log}" +export RABBITMQ_NODENAME="${RABBITMQ_NODENAME:-rabbit@localhost}" +export HOME="${HOME:-/tmp/rmq}" +mkdir -p "$RABBITMQ_MNESIA_BASE" "$RABBITMQ_LOG_BASE" + +epmd -daemon 2>/dev/null || true +sleep 1 +setsid nohup "$RABBITMQ_HOME/sbin/rabbitmq-server" > /tmp/rmq/boot.log 2>&1 < /dev/null & +for _ in $(seq 1 60); do + if (echo > /dev/tcp/127.0.0.1/5672) 2>/dev/null; then echo "broker ready on 5672"; exit 0; fi + sleep 1 +done +echo "broker did not start; see /tmp/rmq/boot.log" >&2 +exit 1 diff --git a/broker-comparison/scripts/run-all.sh b/broker-comparison/scripts/run-all.sh new file mode 100755 index 0000000..0af9358 --- /dev/null +++ b/broker-comparison/scripts/run-all.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +# Regenerates every file under docs/output/. Each broker is started, measured and left running +# only for its own group of tests, because three brokers do not fit comfortably in memory on a +# small machine at the same time. +# +# Required environment: +# KAFKA_HOME unpacked kafka_2.13-4.2.1 +# ERL_ROOT an Erlang 24 installation (RabbitMQ 3.10.x) +# RABBITMQ_HOME unpacked rabbitmq-server-generic-unix-3.10.25 +# PULSAR_HOME unpacked apache-pulsar-4.2.4-bin +set -euo pipefail +cd "$(dirname "$0")/.." +mkdir -p docs/output + +# Kafka needs no external broker for the measurements: spring-kafka-test starts a real KRaft +# broker in-process. KAFKA_HOME is only used by the footprint measurement. +mvn -B -Dgroups=kafka test 2>&1 | grep -E 'Running |Tests run:|BUILD ' > docs/output/tests.txt + +scripts/rabbit-broker.sh +mvn -B -Dgroups=rabbit test 2>&1 | grep -E 'Running |Tests run:|BUILD ' >> docs/output/tests.txt + +scripts/pulsar-broker.sh +mvn -B -Dgroups=pulsar test 2>&1 | grep -E 'Running |Tests run:|BUILD ' >> docs/output/tests.txt + +{ + echo "== Operational footprint, measured on one 2-core / 3.8 GB Linux box ==" + echo + echo "Each broker started from its shipped distribution with default configuration and a" + echo "700 MB heap cap, on Temurin JDK 21. Times are one sample on a small box: treat them as" + echo "orders of magnitude, not as a benchmark." + echo + scripts/footprint.sh kafka; echo + scripts/footprint.sh rabbit; echo + scripts/footprint.sh pulsar +} > docs/output/footprint.txt + +echo "regenerated:"; ls -1 docs/output diff --git a/broker-comparison/src/main/java/com/ankurm/brokers/BrokerComparisonApplication.java b/broker-comparison/src/main/java/com/ankurm/brokers/BrokerComparisonApplication.java new file mode 100644 index 0000000..950f3b8 --- /dev/null +++ b/broker-comparison/src/main/java/com/ankurm/brokers/BrokerComparisonApplication.java @@ -0,0 +1,15 @@ +package com.ankurm.brokers; + +/** + * There is deliberately no Spring application here. The measurements in src/test drive the three + * client libraries directly so that the thing being compared is the broker's delivery model and + * not three different sets of Spring defaults. + * + *

What each Spring Boot starter adds on top — and what it silently does not add, which + * in Boot 4 is more than people expect — is in docs/06-what-spring-adds.md. + */ +public final class BrokerComparisonApplication { + + private BrokerComparisonApplication() { + } +} diff --git a/broker-comparison/src/test/java/com/ankurm/brokers/Capture.java b/broker-comparison/src/test/java/com/ankurm/brokers/Capture.java new file mode 100644 index 0000000..06954dc --- /dev/null +++ b/broker-comparison/src/test/java/com/ankurm/brokers/Capture.java @@ -0,0 +1,23 @@ +package com.ankurm.brokers; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; + +/** Writes a transcript under docs/output/. Nothing in the article is typed by hand. */ +public final class Capture { + + private Capture() { + } + + public static void write(String fileName, String heading, String body) { + Path dir = Path.of(System.getProperty("user.dir"), "docs", "output"); + try { + Files.createDirectories(dir); + Files.writeString(dir.resolve(fileName), "== " + heading + " ==\n\n" + body + "\n"); + } + catch (IOException ex) { + throw new IllegalStateException("could not write " + fileName, ex); + } + } +} diff --git a/broker-comparison/src/test/java/com/ankurm/brokers/KafkaComparisonTest.java b/broker-comparison/src/test/java/com/ankurm/brokers/KafkaComparisonTest.java new file mode 100644 index 0000000..17e5b6e --- /dev/null +++ b/broker-comparison/src/test/java/com/ankurm/brokers/KafkaComparisonTest.java @@ -0,0 +1,232 @@ +package com.ankurm.brokers; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import org.springframework.kafka.test.EmbeddedKafkaKraftBroker; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Kafka measured against a real broker: an in-process KRaft broker from spring-kafka-test, one + * node, one topic, three partitions. + * + *

The three questions are the same for all three brokers: what ordering survives, what can be + * replayed, and how far consumers scale. + */ +@Tag("kafka") +class KafkaComparisonTest { + + private static final String ORDERING = "orders-ordering"; + + private static final String REPLAY = "orders-replay"; + + private static final String SCALING = "orders-scaling"; + + private static EmbeddedKafkaKraftBroker broker; + + @BeforeAll + static void startBroker() { + broker = new EmbeddedKafkaKraftBroker(1, 3, ORDERING, REPLAY, SCALING); + broker.afterPropertiesSet(); + } + + @AfterAll + static void stopBroker() { + broker.destroy(); + } + + // ---------------------------------------------------------------- ordering + + @Test + void orderingIsPerPartitionOnly() { + produce(ORDERING, 12); + + List> received = consumeAll(ORDERING, "ordering-group", 12); + + Map> perKey = new LinkedHashMap<>(); + Map> partitionsPerKey = new LinkedHashMap<>(); + List globalOrder = new ArrayList<>(); + for (ConsumerRecord record : received) { + perKey.computeIfAbsent(record.key(), (k) -> new ArrayList<>()).add(record.value()); + partitionsPerKey.computeIfAbsent(record.key(), (k) -> new java.util.TreeSet<>()) + .add(record.partition()); + globalOrder.add(record.key() + "=" + record.value() + "@p" + record.partition()); + } + + // Per key, the sequence numbers must come back in the order they were produced. Compared + // as numbers: as strings, "10" sorts before "2" and the assertion is meaningless. + perKey.forEach((key, values) -> assertThat(values.stream().map(Integer::parseInt).toList()) + .isSorted()); + // Each key went to exactly one partition, which is why that holds... + partitionsPerKey.forEach((key, partitions) -> assertThat(partitions).hasSize(1)); + // ...and the three keys are spread over all three partitions, so the transcript below is + // actually testing something. + assertThat(partitionsPerKey.values().stream().flatMap(Set::stream).distinct().count()) + .isEqualTo(3); + List asDelivered = globalOrder.stream() + .map((entry) -> Integer.parseInt(entry.split("=")[1].split("@")[0])).toList(); + assertThat(asDelivered).isNotEqualTo( + java.util.stream.IntStream.rangeClosed(1, 12).boxed().toList()); + + StringBuilder body = new StringBuilder(); + body.append("produced 12 records, keys D/A/F round-robin, values 1..12 in order\n") + .append("(keys chosen so that murmur2 spreads them: D->0, A->1, F->2. A, B and C\n") + .append(" all hash to partition 1 with three partitions, which is worth knowing\n") + .append(" before you decide your keys are well distributed.)\n\n"); + body.append("consumed in this order:\n ").append(String.join("\n ", globalOrder)) + .append("\n\nper key:\n"); + perKey.forEach((key, values) -> body.append(" ").append(key).append(" -> partition ") + .append(partitionsPerKey.get(key)).append(" values ").append(values).append('\n')); + body.append("\nOrder is preserved within each key because a key hashes to one partition.\n") + .append("Across keys it is not: the values above are not 1..12 in order, because a\n") + .append("consumer drains one partition's buffer before the next.\n"); + Capture.write("kafka-ordering.txt", "Kafka: ordering is per partition", body.toString()); + } + + // ---------------------------------------------------------------- replay + + @Test + void anythingStillWithinRetentionCanBeReadAgain() { + produce(REPLAY, 12); + + int first = consumeAll(REPLAY, "replay-group-1", 12).size(); + int again = consumeAll(REPLAY, "replay-group-2", 12).size(); + + List> third; + try (KafkaConsumer consumer = consumer("replay-group-1")) { + consumer.subscribe(List.of(REPLAY)); + consumer.poll(Duration.ofSeconds(2)); + consumer.seekToBeginning(consumer.assignment()); + third = drain(consumer, 12); + } + + assertThat(first).isEqualTo(12); + assertThat(again).isEqualTo(12); + assertThat(third).hasSize(12); + + Capture.write("kafka-replay.txt", "Kafka: the log is the storage", + """ + group replay-group-1, first read : %d records + group replay-group-2, brand new group : %d records + group replay-group-1, after seekToBeginning: %d records + + Consuming does not remove anything. A consumer group is a cursor over a log + that the broker keeps until retention expires, so a new group, a reset + offset or a seek all read the same records again. + """.formatted(first, again, third.size())); + } + + // ---------------------------------------------------------------- consumer scaling + + @Test + void consumerParallelismIsCappedByPartitionCount() throws Exception { + produce(SCALING, 12); + + List> consumers = new ArrayList<>(); + Map> assignment = new LinkedHashMap<>(); + try { + for (int i = 1; i <= 5; i++) { + KafkaConsumer consumer = consumer("scaling-group"); + consumer.subscribe(List.of(SCALING)); + consumers.add(consumer); + } + // Poll each consumer until the group has settled and every member knows its share. + for (int round = 0; round < 8; round++) { + for (KafkaConsumer consumer : consumers) { + consumer.poll(Duration.ofMillis(500)); + } + } + for (int i = 0; i < consumers.size(); i++) { + Set partitions = new java.util.TreeSet<>(); + for (TopicPartition tp : consumers.get(i).assignment()) { + partitions.add(tp.partition()); + } + assignment.put("consumer-" + (i + 1), partitions); + } + } + finally { + consumers.forEach(KafkaConsumer::close); + } + + long idle = assignment.values().stream().filter(Set::isEmpty).count(); + assertThat(assignment).hasSize(5); + assertThat(idle).isEqualTo(2); + + StringBuilder body = new StringBuilder("topic 'orders-scaling', 3 partitions, 5 consumers in one group\n\n"); + assignment.forEach((name, partitions) -> body.append(" ").append(name).append(" -> ") + .append(partitions.isEmpty() ? "no partitions (idle)" : "partitions " + partitions) + .append('\n')); + body.append("\nconsumers with no partitions: ").append(idle).append('\n') + .append("\nA partition is assigned to at most one consumer in a group, so the number\n") + .append("of partitions is a hard ceiling on consumer parallelism. Adding consumers\n") + .append("beyond it adds idle processes, not throughput.\n"); + Capture.write("kafka-consumer-scaling.txt", + "Kafka: partitions are the ceiling on consumer parallelism", body.toString()); + } + + // ---------------------------------------------------------------- helpers + + private void produce(String topic, int count) { + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString()); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + // D, A and F, not A, B and C: with three partitions murmur2 maps A, B and C all to + // partition 1, so the obvious choice of keys would have produced a transcript in which + // global order happened to be preserved and proved nothing. D->0, A->1, F->2. + String[] keys = { "D", "A", "F" }; + try (KafkaProducer producer = new KafkaProducer<>(props)) { + for (int i = 1; i <= count; i++) { + producer.send(new ProducerRecord<>(topic, keys[i % keys.length], String.valueOf(i))); + } + producer.flush(); + } + } + + private KafkaConsumer consumer(String group) { + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString()); + props.put(ConsumerConfig.GROUP_ID_CONFIG, group); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + return new KafkaConsumer<>(props); + } + + private List> consumeAll(String topic, String group, int expected) { + try (KafkaConsumer consumer = consumer(group)) { + consumer.subscribe(List.of(topic)); + return drain(consumer, expected); + } + } + + private List> drain(KafkaConsumer consumer, int expected) { + List> received = new ArrayList<>(); + for (int i = 0; i < 20 && received.size() < expected; i++) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); + records.forEach(received::add); + } + return received; + } +} diff --git a/broker-comparison/src/test/java/com/ankurm/brokers/PulsarComparisonTest.java b/broker-comparison/src/test/java/com/ankurm/brokers/PulsarComparisonTest.java new file mode 100644 index 0000000..3a9c06c --- /dev/null +++ b/broker-comparison/src/test/java/com/ankurm/brokers/PulsarComparisonTest.java @@ -0,0 +1,298 @@ +package com.ankurm.brokers; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.SubscriptionType; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Pulsar measured against a real standalone broker on localhost:6650, started by + * scripts/pulsar-broker.sh. + * + *

Pulsar is the interesting third option because it does not force the trade the other two + * force. The subscription type is chosen per subscriber rather than being a property of the + * topic, so ordering and consumer fan-out are separate decisions. + */ +@Tag("pulsar") +class PulsarComparisonTest { + + private static PulsarClient client; + + @BeforeAll + static void connect() throws Exception { + client = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build(); + } + + @AfterAll + static void close() throws Exception { + client.close(); + } + + // ---------------------------------------------------------------- ordering + + @Test + void sharedSpreadsAKeyAndKeySharedPinsIt() throws Exception { + String topic = "persistent://public/default/ordering-" + System.nanoTime(); + produce(topic, 12); + + Map> shared = consumersPerKey(topic, "sub-shared", SubscriptionType.Shared, 12); + Map> keyShared = consumersPerKey(topic, "sub-keyshared", + SubscriptionType.Key_Shared, 12); + + // Key_Shared: every key was handled by exactly one consumer, so per-key order is safe. + keyShared.forEach((key, handlers) -> assertThat(handlers).hasSize(1)); + // Shared: at least one key was handled by both, so two messages with that key were in + // flight on two consumers and nothing orders them. + assertThat(shared.values().stream().anyMatch((handlers) -> handlers.size() > 1)).isTrue(); + + Capture.write("pulsar-ordering.txt", "Pulsar: the subscription type decides", + """ + 12 messages, keys A/B/C, values 1..12 in order, two consumers per + subscription, receiverQueueSize 1 so the first consumer cannot take the + whole backlog. + + Shared subscription, which consumers saw each key: + %s + Key_Shared subscription, which consumers saw each key: + %s + A Shared subscription round-robins individual messages, so messages with + the same key end up on different consumers and can be processed at the + same time: there is no per-key order left to speak of. Key_Shared hashes + the key to one consumer, which is Kafka's guarantee -- except that the + assignment belongs to the subscription and is recomputed as consumers come + and go, rather than being fixed by a partition count chosen when the topic + was created. + """.formatted(render3(shared), render3(keyShared))); + } + + // ---------------------------------------------------------------- replay + + @Test + void aSubscriptionCanBeRewound() throws Exception { + String topic = "persistent://public/default/replay-" + System.nanoTime(); + produce(topic, 12); + + int first; + int afterSeek; + try (Consumer consumer = consumer(topic, "replay-sub", SubscriptionType.Exclusive)) { + first = drain(consumer, 12).size(); + consumer.seek(MessageId.earliest); + afterSeek = drain(consumer, 12).size(); + } + + int newSubscription; + try (Consumer consumer = consumer(topic, "replay-sub-2", SubscriptionType.Exclusive)) { + newSubscription = drain(consumer, 12).size(); + } + + assertThat(first).isEqualTo(12); + assertThat(afterSeek).isEqualTo(12); + assertThat(newSubscription).isEqualTo(12); + + Capture.write("pulsar-replay.txt", "Pulsar: acknowledged, but still there", + """ + subscription replay-sub, first read : %d messages + subscription replay-sub, after seek(earliest): %d messages + subscription replay-sub-2, brand new : %d messages + + Acknowledgement moves a cursor; the message itself lives in the managed + ledger. seek(MessageId) and seek(timestamp) rewind a live subscription, + which Kafka can also do by resetting offsets. What differs is the default: + Pulsar deletes a message once every subscription has acknowledged it, + unless a retention policy on the namespace says otherwise, whereas Kafka + keeps it for the retention period regardless of who read it. A Pulsar + topic with no retention policy and no subscriptions keeps nothing. + """.formatted(first, afterSeek, newSubscription)); + } + + // ---------------------------------------------------------------- consumer scaling + + @Test + void sharedSubscriptionsHaveNoPartitionCeilingButHaveAReceiverQueue() throws Exception { + String topicA = "persistent://public/default/scaling-default-" + System.nanoTime(); + produce(topicA, 40); + Map withDefaultQueue = drainAcross(topicA, "scaling-sub", 5, 40, 0); + + String topicB = "persistent://public/default/scaling-q1-" + System.nanoTime(); + produce(topicB, 40); + Map withQueueOfOne = drainAcross(topicB, "scaling-sub", 5, 40, 1); + + long idleDefault = withDefaultQueue.values().stream().filter((n) -> n == 0).count(); + long idleTuned = withQueueOfOne.values().stream().filter((n) -> n == 0).count(); + + assertThat(idleTuned).isZero(); + assertThat(idleDefault).isGreaterThan(0); + + Capture.write("pulsar-consumer-scaling.txt", + "Pulsar: no partition ceiling, but the receiver queue decides who gets the work", + """ + one non-partitioned topic, 40 messages, 5 consumers, Shared subscription + + receiverQueueSize left at the default (1000): + %s consumers that received nothing : %d + + receiverQueueSize(1): + %s consumers that received nothing : %d + + The topic has no partitions and five consumers can still share the work -- + in Kafka the same shape needs at least five partitions, chosen when the + topic was created. But the default receiver queue is 1000 messages, so the + first consumers to connect pull the whole 40-message backlog into their own + buffers before the rest ask for anything, and the subscription looks + broken. Which consumers win is a race and varies between runs -- one + consumer taking all forty, or two taking twenty each -- but the consumers + that lose it see nothing at all. This is the same trap as RabbitMQ's + unbounded prefetch, with a different name and a much larger default. + """.formatted(render2(withDefaultQueue), idleDefault, + render2(withQueueOfOne), idleTuned)); + } + + // ---------------------------------------------------------------- helpers + + private void produce(String topic, int count) throws Exception { + String[] keys = { "A", "B", "C" }; + try (Producer producer = client.newProducer(Schema.STRING).topic(topic).create()) { + for (int i = 1; i <= count; i++) { + producer.newMessage().key(keys[i % keys.length]).value(String.valueOf(i)).send(); + } + } + } + + private Consumer consumer(String topic, String subscription, SubscriptionType type) + throws Exception { + return consumer(topic, subscription, type, 0); + } + + /** + * @param receiverQueueSize 0 leaves Pulsar's default of 1000 in place; anything else sets it. + * The default matters more than it looks: it is how many messages a single consumer will pull + * into its own buffer before another consumer on the same subscription gets a chance. + */ + private Consumer consumer(String topic, String subscription, SubscriptionType type, + int receiverQueueSize) throws Exception { + var builder = client.newConsumer(Schema.STRING) + .topic(topic) + .subscriptionName(subscription) + .subscriptionType(type) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + if (receiverQueueSize > 0) { + builder = builder.receiverQueueSize(receiverQueueSize); + } + return builder.subscribe(); + } + + private List drain(Consumer consumer, int expected) throws Exception { + List received = new ArrayList<>(); + while (received.size() < expected) { + Message message = consumer.receive(2, TimeUnit.SECONDS); + if (message == null) { + break; + } + received.add(message.getValue()); + consumer.acknowledge(message); + } + return received; + } + + /** Which consumers saw each key, with a receiver queue of one so nothing hoards. */ + private Map> consumersPerKey(String topic, String subscription, + SubscriptionType type, int expected) throws Exception { + Map> perKey = new LinkedHashMap<>(); + List> consumers = new ArrayList<>(); + try { + consumers.add(consumer(topic, subscription, type, 1)); + consumers.add(consumer(topic, subscription, type, 1)); + int received = 0; + while (received < expected) { + boolean any = false; + for (int i = 0; i < consumers.size(); i++) { + Message message = consumers.get(i).receive(500, TimeUnit.MILLISECONDS); + if (message != null) { + perKey.computeIfAbsent(message.getKey(), (k) -> new java.util.TreeSet<>()) + .add("consumer-" + (i + 1)); + consumers.get(i).acknowledge(message); + received++; + any = true; + } + } + if (!any) { + break; + } + } + } + finally { + for (Consumer consumer : consumers) { + consumer.close(); + } + } + return perKey; + } + + /** Drains a Shared subscription across {@code n} consumers and reports the distribution. */ + private Map drainAcross(String topic, String subscription, int n, int expected, + int receiverQueueSize) throws Exception { + Map counts = new LinkedHashMap<>(); + List> consumers = new ArrayList<>(); + try { + for (int i = 1; i <= n; i++) { + consumers.add(consumer(topic, subscription, SubscriptionType.Shared, receiverQueueSize)); + counts.put("consumer-" + i, 0); + } + int received = 0; + while (received < expected) { + boolean any = false; + for (int i = 0; i < consumers.size(); i++) { + Message message = consumers.get(i).receive(300, TimeUnit.MILLISECONDS); + if (message != null) { + consumers.get(i).acknowledge(message); + counts.merge("consumer-" + (i + 1), 1, Integer::sum); + received++; + any = true; + } + } + if (!any) { + break; + } + } + } + finally { + for (Consumer consumer : consumers) { + consumer.close(); + } + } + return counts; + } + + private static String render3(Map> perKey) { + StringBuilder out = new StringBuilder(); + perKey.entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach((entry) -> + out.append(" ").append(entry.getKey()).append(" -> ").append(entry.getValue()) + .append('\n')); + return out.toString(); + } + + private static String render2(Map counts) { + StringBuilder out = new StringBuilder(); + counts.entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach((entry) -> + out.append(" ").append(entry.getKey()).append(" -> ").append(entry.getValue()) + .append(" messages\n")); + return out.toString(); + } +} diff --git a/broker-comparison/src/test/java/com/ankurm/brokers/RabbitComparisonTest.java b/broker-comparison/src/test/java/com/ankurm/brokers/RabbitComparisonTest.java new file mode 100644 index 0000000..6c6ea8e --- /dev/null +++ b/broker-comparison/src/test/java/com/ankurm/brokers/RabbitComparisonTest.java @@ -0,0 +1,286 @@ +package com.ankurm.brokers; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; +import com.rabbitmq.client.DefaultConsumer; +import com.rabbitmq.client.Envelope; +import com.rabbitmq.client.GetResponse; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * RabbitMQ measured against a real broker on localhost:5672, started by + * scripts/rabbit-broker.sh. + * + *

The same three questions as the Kafka measurements, against a broker that owns the message + * until it is acknowledged rather than a log that keeps it. + */ +@Tag("rabbit") +class RabbitComparisonTest { + + private static Connection connection; + + @BeforeAll + static void connect() throws Exception { + ConnectionFactory factory = new ConnectionFactory(); + factory.setHost("127.0.0.1"); + factory.setPort(5672); + connection = factory.newConnection(); + } + + @AfterAll + static void close() throws Exception { + connection.close(); + } + + // ---------------------------------------------------------------- ordering + + @Test + void oneConsumerIsFifoAndTwoConsumersAreNot() throws Exception { + String single = declare("order-single"); + publish(single, 12); + List fifo = getAll(single, 12); + + String shared = declare("order-shared"); + publish(shared, 12); + List completionOrder = consumeWithTwoConsumers(shared, 12); + + assertThat(fifo).containsExactly("1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12"); + + int inversions = 0; + for (int i = 1; i < completionOrder.size(); i++) { + if (Integer.parseInt(completionOrder.get(i)) < Integer.parseInt(completionOrder.get(i - 1))) { + inversions++; + } + } + + Capture.write("rabbit-ordering.txt", "RabbitMQ: FIFO per queue, per consumer", + """ + one queue, one consumer, 12 messages + completion order : %s + + one queue, two consumers, prefetch 1, consumer-1 slower than consumer-2 + completion order : %s + out-of-order steps: %d + + A queue is FIFO and a single consumer sees it that way. The moment a second + consumer is added the broker hands the next message to whichever consumer + is free, so the order in which work finishes is no longer the order in + which it was published. There is no key: RabbitMQ has no notion of a + partition to which related messages could be pinned. Ordering across + related messages means one queue and one consumer, and therefore no + horizontal scaling for that queue -- or a consistent-hash exchange, which + is a plugin. + """.formatted(fifo, completionOrder, inversions)); + } + + // ---------------------------------------------------------------- replay + + @Test + void anAcknowledgedMessageIsGone() throws Exception { + String queue = declare("replay-probe"); + publish(queue, 12); + + List first = getAll(queue, 12); + List second = getAll(queue, 12); + + int depth = depth(queue); + + assertThat(first).hasSize(12); + assertThat(second).isEmpty(); + assertThat(depth).isZero(); + + Capture.write("rabbit-replay.txt", "RabbitMQ: there is nothing to replay", + """ + first drain of the queue : %d messages + second drain of the queue : %d messages + queue depth afterwards : %d + + Acknowledging a message deletes it. The broker is a router with buffers, + not a log: there is no offset to rewind and no second reader that can see + what the first one consumed. Reading the same message twice means + arranging it in advance -- a second queue bound to the same exchange, or a + copy written somewhere else -- and it cannot be arranged after the fact. + """.formatted(first.size(), second.size(), depth)); + } + + // ---------------------------------------------------------------- consumer scaling + + @Test + void consumersScaleWithoutAPartitionCeilingButPrefetchDecidesWho() throws Exception { + String unlimited = declare("scaling-unlimited"); + publish(unlimited, 40); + Map withDefaultPrefetch = drainAcross(unlimited, 5, 40, 0); + + String throttled = declare("scaling-prefetch-1"); + publish(throttled, 40); + Map withPrefetchOne = drainAcross(throttled, 5, 40, 1); + + long idleDefault = withDefaultPrefetch.values().stream().filter((n) -> n == 0).count(); + long idleTuned = withPrefetchOne.values().stream().filter((n) -> n == 0).count(); + + assertThat(idleTuned).isZero(); + + Capture.write("rabbit-consumer-scaling.txt", + "RabbitMQ: consumers scale, and prefetch decides whether they actually do", + """ + one queue, 40 messages, 5 consumers, each taking 10 ms per message + + no basicQos at all (unlimited prefetch, the AMQP default): + %s consumers that received nothing : %d + + basicQos(1): + %s consumers that received nothing : %d + + Every consumer on a queue competes for the same messages, so adding + consumers adds throughput and there is no structural ceiling of the kind + Kafka's partition count imposes. But with the AMQP default the broker + pushes as many messages as a consumer will take, so whichever consumer + connects first can be handed the entire backlog while the others sit idle. + basicQos is not a tuning knob you get to postpone; it is what makes the + fan-out real. Spring AMQP sets it for you -- + AbstractMessageListenerContainer.DEFAULT_PREFETCH_COUNT is 250 -- which is + better than unlimited and still large enough to concentrate a small + backlog on one consumer. + + The price of all this is the ordering measurement: there is no key, so + nothing constrains related messages to one consumer. + """.formatted(render(withDefaultPrefetch), idleDefault, + render(withPrefetchOne), idleTuned)); + } + + // ---------------------------------------------------------------- helpers + + private String declare(String name) throws Exception { + try (Channel channel = connection.createChannel()) { + channel.queueDelete(name); + channel.queueDeclare(name, false, false, true, null); + } + return name; + } + + private void publish(String queue, int count) throws Exception { + try (Channel channel = connection.createChannel()) { + for (int i = 1; i <= count; i++) { + channel.basicPublish("", queue, null, String.valueOf(i).getBytes(StandardCharsets.UTF_8)); + } + } + } + + private List getAll(String queue, int max) throws Exception { + List received = new ArrayList<>(); + try (Channel channel = connection.createChannel()) { + for (int i = 0; i < max; i++) { + GetResponse response = channel.basicGet(queue, true); + if (response == null) { + break; + } + received.add(new String(response.getBody(), StandardCharsets.UTF_8)); + } + } + return received; + } + + private int depth(String queue) throws Exception { + try (Channel channel = connection.createChannel()) { + return channel.queueDeclarePassive(queue).getMessageCount(); + } + } + + private List consumeWithTwoConsumers(String queue, int count) throws Exception { + List completion = new CopyOnWriteArrayList<>(); + CountDownLatch done = new CountDownLatch(count); + List channels = new ArrayList<>(); + for (int i = 1; i <= 2; i++) { + long delay = (i == 1) ? 60 : 5; + Channel channel = connection.createChannel(); + channel.basicQos(1); + channels.add(channel); + channel.basicConsume(queue, false, new DefaultConsumer(channel) { + @Override + public void handleDelivery(String tag, Envelope envelope, AMQP.BasicProperties props, + byte[] body) throws IOException { + try { + Thread.sleep(delay); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + completion.add(new String(body, StandardCharsets.UTF_8)); + getChannel().basicAck(envelope.getDeliveryTag(), false); + done.countDown(); + } + }); + } + done.await(30, TimeUnit.SECONDS); + for (Channel channel : channels) { + channel.close(); + } + return List.copyOf(completion); + } + + /** Drains a queue across {@code n} competing consumers. prefetch 0 means "do not set it". */ + private Map drainAcross(String queue, int n, int expected, int prefetch) + throws Exception { + Map perConsumer = new ConcurrentHashMap<>(); + CountDownLatch done = new CountDownLatch(expected); + List channels = new ArrayList<>(); + for (int i = 1; i <= n; i++) { + String name = "consumer-" + i; + Channel channel = connection.createChannel(); + if (prefetch > 0) { + channel.basicQos(prefetch); + } + channels.add(channel); + perConsumer.put(name, new AtomicInteger()); + channel.basicConsume(queue, false, new DefaultConsumer(channel) { + @Override + public void handleDelivery(String tag, Envelope envelope, AMQP.BasicProperties props, + byte[] body) throws IOException { + perConsumer.get(name).incrementAndGet(); + try { + Thread.sleep(10); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + getChannel().basicAck(envelope.getDeliveryTag(), false); + done.countDown(); + } + }); + } + done.await(30, TimeUnit.SECONDS); + for (Channel channel : channels) { + channel.close(); + } + Map counts = new LinkedHashMap<>(); + perConsumer.forEach((name, count) -> counts.put(name, count.get())); + return counts; + } + + private static String render(Map counts) { + StringBuilder out = new StringBuilder(); + counts.entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach((entry) -> + out.append(" ").append(entry.getKey()).append(" -> ").append(entry.getValue()) + .append(" messages\n")); + return out.toString(); + } +}