-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();
+ }
+}