kafka-streams: capture the ambiguous null tombstone compile error

Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01RjfyTgmERTWLqzafu34Cth
Add kafka-streams module: fraud detection topology tested with TopologyTestDriver
2026-10-14 07:12:07 +00:00 · 2026-10-14 06:44:56 +00:00 · 2026-10-13 02:43:36 +00:00 · 2026-10-03 19:56:40 +00:00 · 2026-09-28 09:19:55 +00:00 · 2026-09-04 01:03:48 +05:30
142 changed files with 4751 additions and 3 deletions
@@ -1,9 +1,10 @@
 # spring-messaging-demo

 Companion code for the messaging series on [ankurm.com](https://ankurm.com). Each directory is a
-self-contained Maven project for one article, with its own `pom.xml`, its own numbered
-documentation chapters, and its own captured output under `docs/output/` — regenerated by that
-module's `scripts/run-all.sh`, never typed by hand.
+self-contained Maven project for one article, with its own `pom.xml` and its own captured
+output, regenerated from the test suite rather than typed by hand. Most modules keep that
+output under `docs/output/` alongside numbered documentation chapters; `saga/` and `share-groups/` keep it
+directly under `output/` instead, with the equivalent depth in the article's own accordion sections.

 | Module | Article | What it demonstrates |
 |---|---|---|
@@ -13,6 +14,9 @@ module's `scripts/run-all.sh`, never typed by hand.
 | [`sse-websocket/`](sse-websocket/README.md) | [Server-Sent Events and WebSocket on Spring Boot 4: SseEmitter, STOMP, and Which to Pick](https://ankurm.com/spring-boot-4-sse-websocket-stomp/) | The two browser-facing options side by side: the SSE lifecycle and the three ways a stream ends, STOMP's routing defaults, and the size limit that is enforced by Tomcat rather than by Spring |
 | [`protocol-comparison/`](protocol-comparison/README.md) | [RSocket vs gRPC vs WebSocket on Spring Boot 4.1: When Each One Wins](https://ankurm.com/rsocket-vs-grpc-vs-websocket-spring-boot-4-1/) | One application serving the same two operations over all three, benchmarked in one JVM, ending in the only measurement that transfers off the box: what each server produces for a consumer that asked for a hundred |
 | [`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 |
+| [`saga/`](saga/README.md) | Saga Pattern in Spring Boot: Orchestration vs Choreography with Kafka | The same order/payment/inventory saga built twice against a real broker &mdash; choreography and orchestration &mdash; driven through the same inventory failure so the compensating transaction can be compared side by side, plus a real duplicate-delivery test for the idempotent-consumer problem |
+| [`share-groups/`](share-groups/README.md) | [Queues for Kafka: Share Groups with Spring for Apache Kafka 4](https://ankurm.com/kafka-share-groups-spring-kafka-4-queues-for-kafka/) | Share groups against a real Kafka 4.3.1 broker: six consumers on two partitions, the acquire mode that decides whether they really share, the three acknowledgement types, a group that starts at the end of the topic, and the ordering you give up |
+| [`kafka-streams/`](kafka-streams/README.md) | Kafka Streams with Spring Boot 4: Real-Time Fraud Detection with Vector Similarity | A fraud topology tested entirely under `TopologyTestDriver`: cosine-similarity scoring, a 5-minute windowed count that alerts exactly once, and a KTable blocklist join with a tombstone, plus the Spring wiring that needs no broker to load |

 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
@@ -0,0 +1,44 @@
+# kafka-streams
+
+Companion code for *Kafka Streams with Spring Boot 4: Real-Time Fraud Detection with Vector
+Similarity* on [ankurm.com](https://ankurm.com). The whole topology runs under
+`TopologyTestDriver`, so **no Kafka broker is needed** to run anything here.
+
+| Stack | Version |
+|---|---|
+| JDK | 25 (Temurin 25.0.4.1+1) |
+| Spring Boot | 4.1.1 |
+| Spring for Apache Kafka | 4.1.1 |
+| kafka-streams / kafka-streams-test-utils | 4.2.1 (what Boot 4.1.1 manages) |
+
+## Run it
+
+```bash
+./scripts/run-all.sh        # runs the tests, rewrites output/*.txt
+```
+
+## What is where
+
+| File | Role |
+|---|---|
+| `FraudTopology` | The topology as plain Kafka Streams code: similarity scoring, 5-minute velocity count, blocklist KTable join |
+| `Embedder` | A hand-built 6-number feature vector and cosine similarity (not a learned model) |
+| `FraudStreamsConfig` | The only Spring wiring: `@EnableKafkaStreams` plus handing the `StreamsBuilder` to the topology |
+| `FraudTopologyTest` | Five tests; each one writes the matching `output/NN-*.txt` |
+| `ContextLoadsTest` | Loads the Spring context with `auto-startup=false` to show the wiring without a broker |
+
+## Captured output
+
+Every file in `output/` is written by a test that also asserts it.
+
+| File | Shows |
+|---|---|
+| `01-similarity-scores.txt` | Scores for five sample transactions against the 0.98 threshold |
+| `02-similarity-alerts.txt` | Only the fraud-shaped transaction produces an alert |
+| `03-velocity-window.txt` | Six transactions in two minutes give one alert; the next window starts from one |
+| `04-blocklist-table-join.txt` | Join before blocking, after blocking, and after a tombstone |
+| `05-topology-description.txt` | `Topology.describe()` of the whole graph |
+| `06-spring-context.txt` | The Spring factory bean is wired and not running |
+| `07-ambiguous-null-tombstone.txt` | The real `javac` error for `pipeInput("c5", null)`; source in `src/broken/`, regenerated by `scripts/compile-broken.sh` |
+
+The article carries the explanations; there is deliberately no `docs/` folder.
@@ -0,0 +1,6 @@
+c1,4700.0,3,crypto                       score=1.0000
+c1,3000.0,2,electronics                  score=1.0000
+c1,45.0,18,grocery                       score=-0.1185
+c1,120.0,13,electronics                  score=0.1387
+c1,2500.0,14,crypto                      score=0.2136
+threshold: 0.98
@@ -0,0 +1,2 @@
+alerts <- c1 SIMILAR:1.0000:c1,4700.0,3,crypto
+alerts emitted: 1
@@ -0,0 +1,5 @@
+alerts <- c9 VELOCITY:4 in 5m
+alerts emitted for 6 transactions in 2 minutes: 1
+alerts after one more transaction 10 minutes later: 0
+state store window 2026-10-14T10:00:00Z count=6
+state store window 2026-10-14T10:10:00Z count=1
@@ -0,0 +1,3 @@
+before blocking, alerts: 0
+alerts <- c5 BLOCKED:reported stolen
+after the tombstone, alerts: 0
@@ -0,0 +1,39 @@
+Topologies:
+   Sub-topology: 0
+    Source: KSTREAM-SOURCE-0000000000 (topics: [transactions])
+      --> KSTREAM-AGGREGATE-0000000005, KSTREAM-JOIN-0000000013, KSTREAM-MAPVALUES-0000000001
+    Processor: KSTREAM-AGGREGATE-0000000005 (stores: [txn-count-5m])
+      --> KTABLE-TOSTREAM-0000000006
+      <-- KSTREAM-SOURCE-0000000000
+    Processor: KSTREAM-MAPVALUES-0000000001 (stores: [])
+      --> score
+      <-- KSTREAM-SOURCE-0000000000
+    Processor: KTABLE-TOSTREAM-0000000006 (stores: [])
+      --> KSTREAM-FILTER-0000000007
+      <-- KSTREAM-AGGREGATE-0000000005
+    Processor: KSTREAM-FILTER-0000000007 (stores: [])
+      --> KSTREAM-MAP-0000000008
+      <-- KTABLE-TOSTREAM-0000000006
+    Processor: score (stores: [])
+      --> KSTREAM-FILTER-0000000003
+      <-- KSTREAM-MAPVALUES-0000000001
+    Processor: KSTREAM-FILTER-0000000003 (stores: [])
+      --> KSTREAM-SINK-0000000004
+      <-- score
+    Processor: KSTREAM-JOIN-0000000013 (stores: [blocked-cards-STATE-STORE-0000000010])
+      --> KSTREAM-SINK-0000000014
+      <-- KSTREAM-SOURCE-0000000000
+    Processor: KSTREAM-MAP-0000000008 (stores: [])
+      --> KSTREAM-SINK-0000000009
+      <-- KSTREAM-FILTER-0000000007
+    Source: KSTREAM-SOURCE-0000000011 (topics: [blocked-cards])
+      --> KTABLE-SOURCE-0000000012
+    Sink: KSTREAM-SINK-0000000004 (topic: alerts)
+      <-- KSTREAM-FILTER-0000000003
+    Sink: KSTREAM-SINK-0000000009 (topic: alerts)
+      <-- KSTREAM-MAP-0000000008
+    Sink: KSTREAM-SINK-0000000014 (topic: alerts)
+      <-- KSTREAM-JOIN-0000000013
+    Processor: KTABLE-SOURCE-0000000012 (stores: [blocked-cards-STATE-STORE-0000000010])
+      --> none
+      <-- KSTREAM-SOURCE-0000000011
@@ -0,0 +1,3 @@
+StreamsBuilderFactoryBean present: true
+auto-startup: false
+running: false
@@ -0,0 +1,8 @@
+src/broken/AmbiguousNullTombstone.java:9: error: reference to pipeInput is ambiguous
+        blocked.pipeInput("c5", null);
+               ^
+  both method pipeInput(K,V) in TestInputTopic and method pipeInput(V,Instant) in TestInputTopic match
+  where K,V are type-variables:
+    K extends Object declared in class TestInputTopic
+    V extends Object declared in class TestInputTopic
+1 error
@@ -0,0 +1,56 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0">
+  <modelVersion>4.0.0</modelVersion>
+
+  <!-- Same parent as the other modules, so spring-kafka and kafka-streams versions are Boot-managed. -->
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>4.1.1</version>
+    <relativePath/>
+  </parent>
+
+  <groupId>com.ankurm</groupId>
+  <artifactId>kafka-streams-fraud</artifactId>
+  <version>1.0</version>
+  <packaging>jar</packaging>
+
+  <properties>
+    <java.version>25</java.version>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-kafka</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-streams</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-streams-test-utils</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>
@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+# Captures the real javac error for the null-tombstone mistake into output/07-ambiguous-null-tombstone.txt
+set -uo pipefail
+cd "$(dirname "$0")/.."
+mvn -B -q dependency:build-classpath -Dmdep.outputFile=target/cp.txt >/dev/null
+mkdir -p target/broken
+javac -d target/broken -cp "$(cat target/cp.txt)" src/broken/AmbiguousNullTombstone.java 2>&1 \
+  | grep -v '^Picked up' > output/07-ambiguous-null-tombstone.txt
+cat output/07-ambiguous-null-tombstone.txt
@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+# Regenerates every file under output/ by running the test suite. No broker needed.
+set -euo pipefail
+cd "$(dirname "$0")/.."
+rm -f output/*.txt
+mvn -B -q test
+scripts/compile-broken.sh >/dev/null || true
+ls output
@@ -0,0 +1,11 @@
+import org.apache.kafka.streams.TestInputTopic;
+
+/**
+ * Deliberately does not compile; it lives outside src/main and src/test so the build ignores it.
+ * scripts/compile-broken.sh runs javac on it and captures the compiler's own message.
+ */
+public class AmbiguousNullTombstone {
+    static void tombstone(TestInputTopic<String, String> blocked) {
+        blocked.pipeInput("c5", null);
+    }
+}
@@ -0,0 +1,51 @@
+package com.ankurm.streams;
+
+import java.util.List;
+
+/**
+ * Turns a transaction into a vector and scores it by cosine similarity to known-fraud vectors.
+ *
+ * <p>This is a hand-built feature vector, not a learned embedding: [log amount, sin(hour),
+ * cos(hour), is-crypto, is-electronics, is-grocery]. The topology around it would be identical if
+ * {@link #embed} called an embedding model; only this one method would change.
+ */
+public final class Embedder {
+
+    private Embedder() {}
+
+    /** Prototypes of confirmed fraud: large, late at night, in risky categories. */
+    static final List<double[]> FRAUD_PROTOTYPES = List.of(
+            embed(new Txn("proto", 4800, 3, "crypto")),
+            embed(new Txn("proto", 2900, 2, "electronics")));
+
+    public static double[] embed(Txn t) {
+        double angle = 2 * Math.PI * t.hour() / 24.0;
+        return new double[] {
+                Math.log10(Math.max(1, t.amount())) / 5.0,
+                Math.sin(angle),
+                Math.cos(angle),
+                "crypto".equals(t.category()) ? 1 : 0,
+                "electronics".equals(t.category()) ? 1 : 0,
+                "grocery".equals(t.category()) ? 1 : 0 };
+    }
+
+    public static double cosine(double[] a, double[] b) {
+        double dot = 0, na = 0, nb = 0;
+        for (int i = 0; i < a.length; i++) {
+            dot += a[i] * b[i];
+            na += a[i] * a[i];
+            nb += b[i] * b[i];
+        }
+        return (na == 0 || nb == 0) ? 0 : dot / (Math.sqrt(na) * Math.sqrt(nb));
+    }
+
+    /** Highest similarity to any fraud prototype, in [-1, 1]. */
+    public static double fraudScore(Txn t) {
+        double[] v = embed(t);
+        double best = -1;
+        for (double[] p : FRAUD_PROTOTYPES) {
+            best = Math.max(best, cosine(v, p));
+        }
+        return best;
+    }
+}
@@ -0,0 +1,20 @@
+package com.ankurm.streams;
+
+import org.apache.kafka.streams.StreamsBuilder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafkaStreams;
+
+/**
+ * The Spring side is only this: {@code @EnableKafkaStreams} supplies a {@link StreamsBuilder}
+ * bean, and this method hands it to the plain-Java topology before the streams instance starts.
+ */
+@Configuration
+@EnableKafkaStreams
+public class FraudStreamsConfig {
+
+    @Autowired
+    public void wire(StreamsBuilder builder) {
+        FraudTopology.wire(builder);
+    }
+}
@@ -0,0 +1,64 @@
+package com.ankurm.streams;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.Grouped;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KTable;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.streams.kstream.TimeWindows;
+
+import java.time.Duration;
+
+/**
+ * The topology, as plain Kafka Streams code with no Spring in it. That is deliberate: a topology
+ * that only needs a {@link StreamsBuilder} can be tested with TopologyTestDriver in milliseconds,
+ * with no broker.
+ *
+ * <pre>
+ * transactions --+--> mapValues(score) --filter(score >= 0.98)--------> alerts   "SIMILAR:..."
+ *                |
+ *                +--> groupByKey --> 5-minute tumbling count (store txn-count-5m)
+ *                |        --filter(count == 4)------------------------> alerts   "VELOCITY:..."
+ *                |
+ *                +--> join(blocked-cards KTable)-----------------------> alerts   "BLOCKED:..."
+ * </pre>
+ */
+public final class FraudTopology {
+
+    public static final String IN = "transactions";
+    public static final String BLOCKED = "blocked-cards";
+    public static final String OUT = "alerts";
+    public static final String COUNT_STORE = "txn-count-5m";
+    public static final double SIMILARITY_THRESHOLD = 0.98;
+    public static final long VELOCITY_LIMIT = 4;
+
+    private FraudTopology() {}
+
+    public static void wire(StreamsBuilder b) {
+        KStream<String, String> raw = b.stream(IN, Consumed.with(Serdes.String(), Serdes.String()));
+
+        // 1. Similarity: score every transaction, keep the ones close to a known-fraud vector.
+        raw.mapValues(v -> Txn.decode(v))
+                .mapValues(t -> "SIMILAR:" + String.format("%.4f", Embedder.fraudScore(t)) + ":" + t.encode(),
+                        org.apache.kafka.streams.kstream.Named.as("score"))
+                .filter((k, v) -> Double.parseDouble(v.split(":")[1]) >= SIMILARITY_THRESHOLD)
+                .to(OUT, Produced.with(Serdes.String(), Serdes.String()));
+
+        // 2. Velocity: count per card in 5-minute tumbling windows, alert on exactly the 4th.
+        raw.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
+                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
+                .count(Materialized.as(COUNT_STORE))
+                .toStream()
+                .filter((w, c) -> c == VELOCITY_LIMIT)
+                .map((w, c) -> org.apache.kafka.streams.KeyValue.pair(w.key(), "VELOCITY:" + c + " in 5m"))
+                .to(OUT, Produced.with(Serdes.String(), Serdes.String()));
+
+        // 3. Blocklist: a KTable of blocked cards, joined to each transaction as it arrives.
+        KTable<String, String> blocked = b.table(BLOCKED, Consumed.with(Serdes.String(), Serdes.String()));
+        raw.join(blocked, (txn, reason) -> "BLOCKED:" + reason)
+                .to(OUT, Produced.with(Serdes.String(), Serdes.String()));
+    }
+}
@@ -0,0 +1,11 @@
+package com.ankurm.streams;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class StreamsApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(StreamsApplication.class, args);
+    }
+}
@@ -0,0 +1,15 @@
+package com.ankurm.streams;
+
+/** A card transaction. */
+public record Txn(String card, double amount, int hour, String category) {
+
+    /** Wire format: {@code card,amount,hour,category}. Plain text keeps the serde trivial to read. */
+    public String encode() {
+        return card + "," + amount + "," + hour + "," + category;
+    }
+
+    public static Txn decode(String s) {
+        String[] p = s.split(",", 4);
+        return new Txn(p[0], Double.parseDouble(p[1]), Integer.parseInt(p[2]), p[3]);
+    }
+}
@@ -0,0 +1,4 @@
+spring.kafka.bootstrap-servers=localhost:9092
+spring.kafka.streams.application-id=fraud-streams
+# Off by default here so the context can load without a broker; set true when you have one.
+spring.kafka.streams.auto-startup=false
@@ -0,0 +1,27 @@
+package com.ankurm.streams;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.ApplicationContext;
+import org.springframework.kafka.config.StreamsBuilderFactoryBean;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/** Proves the Spring wiring: the factory bean exists and holds a topology, without a broker. */
+@SpringBootTest
+class ContextLoadsTest {
+
+    @Autowired ApplicationContext ctx;
+
+    @Test
+    void streamsFactoryBeanIsWiredAndNotStarted() throws Exception {
+        var t = new Transcript("06-spring-context.txt");
+        var fb = ctx.getBean(StreamsBuilderFactoryBean.class);
+        t.line("StreamsBuilderFactoryBean present: " + (fb != null));
+        t.line("auto-startup: " + fb.isAutoStartup());
+        t.line("running: " + fb.isRunning());
+        t.write();
+        assertFalse(fb.isRunning());
+    }
+}
@@ -0,0 +1,140 @@
+package com.ankurm.streams;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.TestOutputTopic;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.state.WindowStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class FraudTopologyTest {
+
+    private TopologyTestDriver driver;
+    private TestInputTopic<String, String> txns;
+    private TestInputTopic<String, String> blocked;
+    private TestOutputTopic<String, String> alerts;
+    private Topology topology;
+
+    @BeforeEach
+    void setUp() {
+        StreamsBuilder b = new StreamsBuilder();
+        FraudTopology.wire(b);
+        topology = b.build();
+        Properties p = new Properties();
+        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "fraud-test");
+        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1");
+        driver = new TopologyTestDriver(topology, p, Instant.parse("2026-10-14T10:00:00Z"));
+        txns = driver.createInputTopic(FraudTopology.IN, new StringSerializer(), new StringSerializer());
+        blocked = driver.createInputTopic(FraudTopology.BLOCKED, new StringSerializer(), new StringSerializer());
+        alerts = driver.createOutputTopic(FraudTopology.OUT, new StringDeserializer(), new StringDeserializer());
+    }
+
+    @AfterEach
+    void tearDown() { driver.close(); }
+
+    @Test
+    void similarityScoresSeparateFraudShapedFromOrdinary() throws Exception {
+        var t = new Transcript("01-similarity-scores.txt");
+        var samples = List.of(
+                new Txn("c1", 4700, 3, "crypto"),
+                new Txn("c1", 3000, 2, "electronics"),
+                new Txn("c1", 45, 18, "grocery"),
+                new Txn("c1", 120, 13, "electronics"),
+                new Txn("c1", 2500, 14, "crypto"));
+        for (Txn x : samples) {
+            t.line(String.format("%-40s score=%.4f", x.encode(), Embedder.fraudScore(x)));
+        }
+        t.line("threshold: " + FraudTopology.SIMILARITY_THRESHOLD);
+        t.write();
+
+        assertTrue(Embedder.fraudScore(samples.get(0)) >= FraudTopology.SIMILARITY_THRESHOLD);
+        assertTrue(Embedder.fraudScore(samples.get(1)) >= FraudTopology.SIMILARITY_THRESHOLD);
+        assertTrue(Embedder.fraudScore(samples.get(2)) < FraudTopology.SIMILARITY_THRESHOLD);
+    }
+
+    @Test
+    void similarTransactionsRaiseAlertsOrdinaryOnesDoNot() throws Exception {
+        var t = new Transcript("02-similarity-alerts.txt");
+        txns.pipeInput("c1", new Txn("c1", 4700, 3, "crypto").encode());
+        txns.pipeInput("c2", new Txn("c2", 45, 18, "grocery").encode());
+        var out = alerts.readKeyValuesToList();
+        out.forEach(kv -> t.line("alerts <- " + kv.key + " " + kv.value));
+        t.line("alerts emitted: " + out.size());
+        t.write();
+
+        assertEquals(1, out.size());
+        assertEquals("c1", out.get(0).key);
+        assertTrue(out.get(0).value.startsWith("SIMILAR:"));
+    }
+
+    @Test
+    void velocityAlertFiresOnTheFourthTransactionInAFiveMinuteWindowOnly() throws Exception {
+        var t = new Transcript("03-velocity-window.txt");
+        Instant start = Instant.parse("2026-10-14T10:00:30Z");
+        for (int i = 0; i < 6; i++) {
+            txns.pipeInput("c9", new Txn("c9", 20, 12, "grocery").encode(), start.plusSeconds(i * 20L));
+        }
+        var out = alerts.readKeyValuesToList();
+        out.forEach(kv -> t.line("alerts <- " + kv.key + " " + kv.value));
+        t.line("alerts emitted for 6 transactions in 2 minutes: " + out.size());
+
+        // The same count in the next window starts again from one.
+        txns.pipeInput("c9", new Txn("c9", 20, 12, "grocery").encode(), start.plusSeconds(600));
+        t.line("alerts after one more transaction 10 minutes later: " + alerts.readKeyValuesToList().size());
+
+        WindowStore<String, Long> store = driver.getWindowStore(FraudTopology.COUNT_STORE);
+        try (KeyValueIterator<org.apache.kafka.streams.kstream.Windowed<String>, Long> it =
+                     store.fetchAll(Instant.parse("2026-10-14T09:00:00Z"), Instant.parse("2026-10-14T11:00:00Z"))) {
+            while (it.hasNext()) {
+                var kv = it.next();
+                t.line("state store window " + kv.key.window().startTime() + " count=" + kv.value);
+            }
+        }
+        t.write();
+
+        assertEquals(1, out.size());
+        assertEquals("VELOCITY:4 in 5m", out.get(0).value);
+    }
+
+    @Test
+    void blocklistTableJoinsEachTransactionAsItArrives() throws Exception {
+        var t = new Transcript("04-blocklist-table-join.txt");
+        txns.pipeInput("c5", new Txn("c5", 15, 12, "grocery").encode());
+        t.line("before blocking, alerts: " + alerts.readKeyValuesToList().size());
+
+        blocked.pipeInput("c5", "reported stolen");
+        txns.pipeInput("c5", new Txn("c5", 15, 12, "grocery").encode());
+        var out = alerts.readKeyValuesToList();
+        out.forEach(kv -> t.line("alerts <- " + kv.key + " " + kv.value));
+
+        blocked.pipeInput("c5", (String) null);   // tombstone removes the card from the table
+        txns.pipeInput("c5", new Txn("c5", 15, 12, "grocery").encode());
+        t.line("after the tombstone, alerts: " + alerts.readKeyValuesToList().size());
+        t.write();
+
+        assertEquals(1, out.size());
+        assertEquals("BLOCKED:reported stolen", out.get(0).value);
+    }
+
+    @Test
+    void describeTheTopology() throws Exception {
+        var t = new Transcript("05-topology-description.txt");
+        t.line(topology.describe().toString().trim());
+        t.write();
+        assertTrue(topology.describe().toString().contains(FraudTopology.COUNT_STORE));
+    }
+}
@@ -0,0 +1,23 @@
+package com.ankurm.streams;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Collects lines and writes them to output/NN-name.txt, so the test IS the transcript. */
+final class Transcript {
+    private final List<String> lines = new ArrayList<>();
+    private final String file;
+
+    Transcript(String file) { this.file = file; }
+
+    void line(String s) { lines.add(s); }
+
+    void write() throws IOException {
+        Path dir = Path.of("output");
+        Files.createDirectories(dir);
+        Files.write(dir.resolve(file), lines);
+    }
+}
@@ -0,0 +1 @@
+target/
@@ -0,0 +1,13 @@
+# kafka4-migration: Spring Kafka 3 to 4 on Spring Boot 3.5 to 4.1
+
+Companion code for the ankurm.com post "Spring Kafka 3 to 4 Migration Guide".
+
+| Folder | What it is |
+|---|---|
+| `legacy/` | Producer, `@KafkaListener` and an `@EmbeddedKafka` test on Boot 3.5.16 / Spring Kafka 3.3.16 / kafka-clients 3.9.2 (JDK 21), using the Jackson 2 `JsonSerializer` |
+| `migrated/` | The same app on Boot 4.1.1 / Spring Kafka 4.1.1 / kafka-clients 4.2.1 (JDK 25), using `spring-boot-starter-kafka` and `JacksonJsonSerializer` |
+| `scripts/run-all.sh` | Builds the intermediate stages from `legacy/` and regenerates every file in `output/` |
+| `output/` | Transcripts for each stage |
+
+Spring Kafka 4.1.1 is the latest GA; 4.2.0-M1 and 4.2.0-M2 are milestones and are not tested here.
+No external broker is needed: the tests use `@EmbeddedKafka`. Tested on 28 September 2026.
@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>3.5.16</version>
+    <relativePath/>
+  </parent>
+  <groupId>com.ankurm</groupId>
+  <artifactId>kafka-legacy</artifactId>
+  <version>1.0.0</version>
+  <properties><java.version>21</java.version></properties>
+  <dependencies>
+    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency>
+    <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>
+    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-json</artifactId></dependency>
+    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency>
+    <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka-test</artifactId><scope>test</scope></dependency>
+  </dependencies>
+</project>
@@ -0,0 +1,9 @@
+package com.ankurm.kafka;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class App {
+    public static void main(String[] args) { SpringApplication.run(App.class, args); }
+}
@@ -0,0 +1,3 @@
+package com.ankurm.kafka;
+
+public record OrderEvent(String id, int qty) { }
@@ -0,0 +1,18 @@
+package com.ankurm.kafka;
+
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.List;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+@Component
+public class OrderListener {
+    public final List<String> received = new CopyOnWriteArrayList<>();
+
+    @KafkaListener(topics = "orders")
+    void on(OrderEvent event, @Header(KafkaHeaders.RECEIVED_KEY) String key) {
+        received.add("key=" + key + " id=" + event.id() + " qty=" + event.qty());
+    }
+}
@@ -0,0 +1,14 @@
+package com.ankurm.kafka;
+
+import java.nio.charset.StandardCharsets;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+
+/** Serialises one event with the JSON serializer, using only the runtime classpath (no test jars). */
+public class SerdeProbe {
+    public static void main(String[] args) {
+        try (JsonSerializer<OrderEvent> s = new JsonSerializer<>()) {
+            byte[] bytes = s.serialize("orders", new OrderEvent("o-1", 3));
+            System.out.println("SERIALIZED " + new String(bytes, StandardCharsets.UTF_8));
+        }
+    }
+}
@@ -0,0 +1,8 @@
+spring.kafka.consumer.group-id=orders-group
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
+spring.kafka.consumer.properties.spring.json.trusted.packages=com.ankurm.kafka
+spring.kafka.consumer.properties.spring.json.value.default.type=com.ankurm.kafka.OrderEvent
@@ -0,0 +1,27 @@
+package com.ankurm.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+
+@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
+@EmbeddedKafka(topics = "orders", partitions = 1)
+class OrderFlowTest {
+    @Autowired KafkaTemplate<String, OrderEvent> template;
+    @Autowired OrderListener listener;
+    @Autowired EmbeddedKafkaBroker broker;
+
+    @Test
+    void sendAndReceive() {
+        template.send("orders", "k1", new OrderEvent("o-1", 3));
+        await().untilAsserted(() -> assertThat(listener.received).hasSize(1));
+        System.out.println("RESULT " + listener.received.get(0));
+        System.out.println("BROKERS " + broker.getBrokersAsString().replaceAll(":\\d+", ":PORT"));
+    }
+}
@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>4.1.1</version>
+    <relativePath/>
+  </parent>
+  <groupId>com.ankurm</groupId>
+  <artifactId>kafka-migrated</artifactId>
+  <version>1.0.0</version>
+  <properties><java.version>25</java.version></properties>
+  <dependencies>
+    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency>
+    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-kafka</artifactId></dependency>
+    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jackson</artifactId></dependency>
+    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency>
+    <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka-test</artifactId><scope>test</scope></dependency>
+  </dependencies>
+</project>
@@ -0,0 +1,9 @@
+package com.ankurm.kafka;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class App {
+    public static void main(String[] args) { SpringApplication.run(App.class, args); }
+}
@@ -0,0 +1,3 @@
+package com.ankurm.kafka;
+
+public record OrderEvent(String id, int qty) { }
@@ -0,0 +1,18 @@
+package com.ankurm.kafka;
+
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.List;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+@Component
+public class OrderListener {
+    public final List<String> received = new CopyOnWriteArrayList<>();
+
+    @KafkaListener(topics = "orders")
+    void on(OrderEvent event, @Header(KafkaHeaders.RECEIVED_KEY) String key) {
+        received.add("key=" + key + " id=" + event.id() + " qty=" + event.qty());
+    }
+}
@@ -0,0 +1,14 @@
+package com.ankurm.kafka;
+
+import java.nio.charset.StandardCharsets;
+import org.springframework.kafka.support.serializer.JacksonJsonSerializer;
+
+/** Serialises one event with the JSON serializer, using only the runtime classpath (no test jars). */
+public class SerdeProbe {
+    public static void main(String[] args) {
+        try (JacksonJsonSerializer<OrderEvent> s = new JacksonJsonSerializer<>()) {
+            byte[] bytes = s.serialize("orders", new OrderEvent("o-1", 3));
+            System.out.println("SERIALIZED " + new String(bytes, StandardCharsets.UTF_8));
+        }
+    }
+}
@@ -0,0 +1,8 @@
+spring.kafka.consumer.group-id=orders-group
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JacksonJsonSerializer
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JacksonJsonDeserializer
+spring.kafka.consumer.properties.spring.json.trusted.packages=com.ankurm.kafka
+spring.kafka.consumer.properties.spring.json.value.default.type=com.ankurm.kafka.OrderEvent
@@ -0,0 +1,27 @@
+package com.ankurm.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+
+@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
+@EmbeddedKafka(topics = "orders", partitions = 1)
+class OrderFlowTest {
+    @Autowired KafkaTemplate<String, OrderEvent> template;
+    @Autowired OrderListener listener;
+    @Autowired EmbeddedKafkaBroker broker;
+
+    @Test
+    void sendAndReceive() {
+        template.send("orders", "k1", new OrderEvent("o-1", 3));
+        await().untilAsserted(() -> assertThat(listener.received).hasSize(1));
+        System.out.println("RESULT " + listener.received.get(0));
+        System.out.println("BROKERS " + broker.getBrokersAsString().replaceAll(":\\d+", ":PORT"));
+    }
+}
@@ -0,0 +1,12 @@
+$ mvn test   # Boot 3.5.16, JDK 21
+RESULT key=k1 id=o-1 qty=3
+BROKERS 127.0.0.1:PORT
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 7.853 s -- in com.ankurm.kafka.OrderFlowTest
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
+[INFO] BUILD SUCCESS
+
+org.apache.kafka:kafka-clients:jar:3.9.2:compile
+org.springframework.kafka:spring-kafka:jar:3.3.16:compile
+
+$ java -cp target/classes:<runtime classpath> SerdeProbe
+SERIALIZED {"id":"o-1","qty":3}
@@ -0,0 +1,5 @@
+$ mvn test   # only the Boot version changed to 4.1.1
+[ERROR] Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 5.728 s <<< FAILURE! -- in com.ankurm.kafka.OrderFlowTest
+[ERROR]   OrderFlowTest.sendAndReceive ? UnsatisfiedDependency Error creating bean with name 'com.ankurm.kafka.OrderFlowTest': Unsatisfied dependency expressed through field 'template': No qualifying bean of type 'org.springframework.kafka.core.KafkaTemplate<java.lang.String, com.ankurm.kafka.OrderEvent>' available: expected at 
+[ERROR] Tests run: 1, Failures: 0, Errors: 1, Skipped: 0
+[INFO] BUILD FAILURE
@@ -0,0 +1,15 @@
+$ mvn test   # spring-kafka replaced by spring-boot-starter-kafka; nothing else changed
+RESULT key=k1 id=o-1 qty=3
+BROKERS localhost:PORT
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 9.853 s -- in com.ankurm.kafka.OrderFlowTest
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
+[INFO] BUILD SUCCESS
+
+org.apache.kafka:kafka-clients:jar:4.2.1:compile
+org.springframework.boot:spring-boot-kafka:jar:4.1.1:compile
+org.springframework.kafka:spring-kafka:jar:4.1.1:compile
+
+$ mvn dependency:list | grep jackson   # runtime scope, then test scope
+runtime: tools.jackson.core:jackson-core:jar:3.1.5:compile
+runtime: tools.jackson.core:jackson-databind:jar:3.1.5:compile
+test-only: com.fasterxml.jackson.core:jackson-databind:jar:2.21.5:test
@@ -0,0 +1,3 @@
+$ compile and run SerdeProbe on the runtime classpath only (JsonSerializer from Spring Kafka 4.1.1)
+[ERROR]   class file for com.fasterxml.jackson.core.type.TypeReference not found
+[ERROR] s2/src/main/java/com/ankurm/kafka/SerdeProbe.java:[9,45] cannot access com.fasterxml.jackson.core.type.TypeReference
@@ -0,0 +1,13 @@
+$ mvn test   # final migrated project (JacksonJsonSerializer, spring-boot-starter-jackson)
+RESULT key=k1 id=o-1 qty=3
+BROKERS localhost:PORT
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 8.478 s -- in com.ankurm.kafka.OrderFlowTest
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
+[INFO] BUILD SUCCESS
+
+org.apache.kafka:kafka-clients:jar:4.2.1:compile
+org.springframework.boot:spring-boot-kafka:jar:4.1.1:compile
+org.springframework.kafka:spring-kafka:jar:4.1.1:compile
+
+$ java -cp target/classes:<runtime classpath> SerdeProbe
+SERIALIZED {"id":"o-1","qty":3}
@@ -0,0 +1,2 @@
+$ jar tf spring-kafka-4.1.1.jar | grep support/serializer/(Json|Jackson)
+JacksonJsonDeserializer.class JacksonJsonSerde.class JacksonJsonSerializer.class JacksonJsonTypeResolver.class JsonDeserializer.class JsonSerde.class JsonSerializer.class JsonTypeResolver.class 
@@ -0,0 +1,44 @@
+#!/usr/bin/env bash
+# Regenerates output/*.txt. Needs Maven, JDK 21 (J21), JDK 25 (J25) and Maven Central access.
+set -u
+HERE=$(cd "$(dirname "$0")/.." && pwd); OUT=$HERE/output; W=${WORK:-/tmp/k4-work}
+J21=${J21:-/usr/lib/jvm/java-21-openjdk-amd64}; J25=${J25:-/tmp/tools/jdk-25.0.4.1+1}; MVN=${MVN:-mvn}
+mkdir -p "$OUT"; rm -rf "$W"; mkdir -p "$W"
+mv_() { local jh=$1 d=$2; shift 2; (cd "$d" && JAVA_HOME=$jh $MVN -B "$@" 2>&1 | grep -v JAVA_TOOL_OPTIONS | sed "s#$W/##g"); }
+keep='^(RESULT|BROKERS)|Tests run:.*Fail|BUILD|ERROR.*(cannot access|No qualifying)'
+stage() { rm -rf "$W/$1"; cp -r "$HERE/$2" "$W/$1"; rm -rf "$W/$1/target"; }
+probe() { # probe <javahome> <dir>: compile and run SerdeProbe on the RUNTIME classpath only
+  local jh=$1 d=$2
+  (cd "$d" && JAVA_HOME=$jh $MVN -B -q compile dependency:build-classpath -Dmdep.includeScope=runtime -Dmdep.outputFile="$W/cp.txt" 2>&1 | grep -v JAVA_TOOL_OPTIONS | grep -E 'ERROR.*(cannot access|class file)' | sed "s#$W/##g" | sort -u
+   [ -f "$W/cp.txt" ] && JAVA_HOME=$jh $jh/bin/java -cp target/classes:$(cat "$W/cp.txt") com.ankurm.kafka.SerdeProbe 2>&1 | grep -v JAVA_TOOL_OPTIONS | head -3); rm -f "$W/cp.txt"; }
+versions() { (cd "$2" && JAVA_HOME=$1 $MVN -B dependency:list 2>&1 | grep -E 'org.springframework.kafka:spring-kafka:|org.springframework.boot:spring-boot-kafka:|org.apache.kafka:kafka-clients:jar:[0-9]' | sed 's/^\[INFO\] *//; s/ -- module.*//' | sort -u); }
+
+stage s0 legacy
+{ echo '$ mvn test   # Boot 3.5.16, JDK 21'; mv_ "$J21" "$W/s0" test | grep -E "$keep"
+  echo; versions "$J21" "$W/s0"
+  echo; echo '$ java -cp target/classes:<runtime classpath> SerdeProbe'; probe "$J21" "$W/s0"; } > "$OUT/01-legacy-kafka3.txt"
+
+stage s1 legacy; rm "$W/s1/src/main/java/com/ankurm/kafka/SerdeProbe.java"; sed -i 's/3.5.16/4.1.1/; s/<java.version>21/<java.version>25/' "$W/s1/pom.xml"
+{ echo '$ mvn test   # only the Boot version changed to 4.1.1'; mv_ "$J25" "$W/s1" test | grep -E "$keep" | cut -c1-330; } > "$OUT/02-boot4-no-starter.txt"
+
+cp -r "$W/s1" "$W/s2"; rm -rf "$W/s2/target"
+sed -i 's#<groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>#<groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-kafka</artifactId>#' "$W/s2/pom.xml"
+{ echo '$ mvn test   # spring-kafka replaced by spring-boot-starter-kafka; nothing else changed'; mv_ "$J25" "$W/s2" test | grep -E "$keep"
+  echo; versions "$J25" "$W/s2"
+  echo; echo '$ mvn dependency:list | grep jackson   # runtime scope, then test scope'
+  (cd "$W/s2" && JAVA_HOME=$J25 $MVN -B dependency:list -DincludeScope=runtime 2>&1 | grep -E 'jackson-(core|databind)' | sed 's/^\[INFO\] *//; s/ -- module.*//' | sort -u | sed 's/^/runtime: /'
+   JAVA_HOME=$J25 $MVN -B dependency:list -DexcludeScope=runtime 2>&1 | grep -E 'com.fasterxml.jackson.core:jackson-databind' | sed 's/^\[INFO\] *//; s/ -- module.*//' | sort -u | sed 's/^/test-only: /'); } > "$OUT/03-starter-added-tests-pass.txt"
+
+cp "$HERE/legacy/src/main/java/com/ankurm/kafka/SerdeProbe.java" "$W/s2/src/main/java/com/ankurm/kafka/"
+{ echo '$ compile and run SerdeProbe on the runtime classpath only (JsonSerializer from Spring Kafka 4.1.1)'; probe "$J25" "$W/s2"; } > "$OUT/04-runtime-classpath-has-no-jackson2.txt"
+
+stage m migrated
+{ echo '$ mvn test   # final migrated project (JacksonJsonSerializer, spring-boot-starter-jackson)'; mv_ "$J25" "$W/m" test | grep -E "$keep"
+  echo; versions "$J25" "$W/m"
+  echo; echo '$ java -cp target/classes:<runtime classpath> SerdeProbe'; probe "$J25" "$W/m"; } > "$OUT/05-migrated.txt"
+
+# 06 API facts from the Spring Kafka 4.1.1 jar
+J=$HOME/.m2/repository/org/springframework/kafka/spring-kafka/4.1.1/spring-kafka-4.1.1.jar
+{ echo '$ jar tf spring-kafka-4.1.1.jar | grep support/serializer/(Json|Jackson)'
+  env -u JAVA_TOOL_OPTIONS $J25/bin/jar tf "$J" | grep -E 'support/serializer/(Json|Jackson)' | sed 's#.*/##' | sort | tr '\n' ' '; echo; } > "$OUT/06-serializer-classes.txt"
+echo done
@@ -0,0 +1,91 @@
+# `protocol-comparison` — RSocket vs gRPC vs WebSocket, same use case, measured
+
+Companion project for
+[**RSocket vs gRPC vs WebSocket on Spring Boot 4.1: When Each One Wins**](https://ankurm.com/rsocket-vs-grpc-vs-websocket-spring-boot-4-1/)
+on ankurm.com.
+
+**One application serves the same two operations over all three protocols.** Same JVM, same heap,
+same data generator, same five fields — so a difference in the numbers is a difference in the
+transports rather than in three separately-written demos.
+[`./scripts/run-all.sh`](scripts/run-all.sh) re-measures everything under
+[`docs/output/`](docs/output/).
+
+## Versions
+
+| | Version | Notes |
+|---|---|---|
+| JDK | 25 (Temurin 25.0.4.1+1) | current LTS |
+| Spring Boot | 4.1.1 | latest GA of the 4.1 line |
+| Spring Framework | 7.0.9 | Boot-managed |
+| grpc-java | **1.83.1** | Boot-managed. Boot 4.1.0 pinned 1.80.0 — it moved |
+| protobuf-java | **4.35.1** | Boot-managed. Boot 4.1.0 pinned 4.34.2 |
+| rsocket-java | 1.1.5 | Boot-managed |
+| Spring gRPC | 1.1.1 | the programming model; separate version from Boot's |
+| Jackson | 3.1.5 (`tools.jackson`) | plus `jackson-dataformat-cbor`, RSocket's default |
+
+Read from `repo1.maven.org/.../maven-metadata.xml` and Boot's own `spring-boot-dependencies` POM.
+
+## Quickstart
+
+```bash
+./scripts/run-all.sh     # six tests, then re-measure docs/output/
+mvn test                 # the same without the capture
+mvn spring-boot:run      # gRPC 9090, RSocket 7000, WebSocket ws://localhost:8080/quotes
+```
+
+Nothing to install: `protoc` and the gRPC codegen plugin resolve as Maven artifacts for your OS
+and architecture, and all three servers start inside the test JVM.
+
+## The operations
+
+| | gRPC | RSocket route | WebSocket command |
+|---|---|---|---|
+| one quote | `QuoteService/GetQuote` | `quote` | `QUOTE <symbol>` |
+| N quotes | `QuoteService/StreamQuotes` | `quotes` | `STREAM <symbol> <n> <delayMs>` |
+| unbounded | `QuoteService/StreamUnbounded` | `quotes.unbounded` | `UNBOUNDED <symbol>` |
+| same, `isReady()`-gated | `QuoteService/StreamUnboundedReady` | — | — |
+| server-side produced count | `QuoteService/Produced` | `produced` | `PRODUCED` |
+
+## Documentation
+
+1. [The same use case, three ways](docs/01-three-protocols.md)
+2. [How they measure — and why a loopback benchmark flatters the wrong one](docs/02-benchmarks.md)
+3. [Payload: the one comparison that needs no benchmark](docs/03-payload.md)
+4. [Back-pressure: the measurement that decides it](docs/04-backpressure.md)
+5. [Choosing](docs/05-choosing.md)
+
+## Captured output
+
+| File | What it shows |
+|---|---|
+| [`payload-sizes.txt`](docs/output/payload-sizes.txt) | one message as protobuf, JSON and CBOR, in hex |
+| [`request-response.txt`](docs/output/request-response.txt) | 5 000 sequential calls, p50/p99/mean per protocol |
+| [`stream-throughput.txt`](docs/output/stream-throughput.txt) | 50 000 messages on one connection |
+| [`backpressure.txt`](docs/output/backpressure.txt) | how many the server produced for a client that wanted 100 |
+| [`grpc-isready.txt`](docs/output/grpc-isready.txt) | the documented gRPC fix, tested — and it did not work here |
+| [`rsocket-codecs.txt`](docs/output/rsocket-codecs.txt) | what Boot configures for RSocket, and why CBOR wins |
+| [`tests.txt`](docs/output/tests.txt) | 6 tests |
+
+## Read the benchmark honestly
+
+Client and server are the same process on one loopback interface, two cores. **There is no
+network.** That deletes protobuf's size advantage, and it flatters whichever protocol does the
+least work per message — which is why raw WebSocket wins both speed benchmarks and why that is
+not the recommendation. Treat the ratios as indicative and the absolutes as meaningless off this
+box.
+
+The one result that *does* transfer is the back-pressure table, because it is a property of the
+protocols rather than of the link.
+
+## Four things this module exists to prove
+
+1. **RSocket's `request(n)` is real and the others have no equivalent.** For a client that took
+   100 messages then went quiet for two seconds, the servers produced **100**, **1 346 233** and
+   **311 617** respectively.
+2. **`ServerCallStreamObserver.isReady()` did not bound a gRPC producer here.** The documented fix
+   was tested; the transcript is committed as measured, including its instability.
+3. **Raw WebSocket is the fastest and has the worst tail.** p50 of 124 µs against a p99 of
+   1 448 µs — a 12x spread, where gRPC's is 3.5x.
+4. **CBOR keeps the field names, and it wins by list order.** It is binary JSON, not a schema
+   format: 67 bytes against protobuf's 35 and JSON's 83. Boot puts `JacksonCborEncoder` ahead of
+   `JacksonJsonEncoder`, and a bare `RSocketStrategies.create()` has neither.
@@ -0,0 +1,83 @@
+# 1. The same use case, three ways
+
+Next: [2. How they measure](02-benchmarks.md)
+
+---
+
+One application in this module serves the same two operations over gRPC, RSocket and a raw
+WebSocket: fetch one quote, and stream N quotes. Same JVM, same heap, same
+[`QuoteSource`](../src/main/java/com/ankurm/protocols/QuoteSource.java), same five fields.
+
+That constraint is the point. Cross-protocol benchmarks published on the internet almost always
+compare three different applications, and end up measuring three different serialisation
+libraries, three JIT states and three thread pools.
+
+## What each one is, in one sentence
+
+| | gRPC | RSocket | Raw WebSocket |
+|---|---|---|---|
+| Transport | HTTP/2 | TCP, WebSocket, or others | HTTP/1.1 Upgrade |
+| Schema | `.proto`, mandatory, code-generated | none required | none |
+| Encoding here | protobuf | CBOR (Spring's default) | JSON |
+| Interaction models | 4, declared in the `.proto` | 4, chosen by return type | whatever you invent |
+| Demand signalling | none in the API | `request(n)` on the wire | none at all |
+| Deadlines | per call, absolute, propagating | none built in | none |
+| Browser client | needs grpc-web + a proxy | yes, over the WebSocket transport | yes, natively |
+| Boot 4 starters | `spring-boot-starter-grpc-server` / `-client` | `spring-boot-starter-rsocket` | `spring-boot-starter-websocket` |
+
+## The three server sides, side by side
+
+**gRPC** ([`GrpcQuoteService`](../src/main/java/com/ankurm/protocols/GrpcQuoteService.java)) is
+generated-class inheritance. There is no registration code: the generated `ImplBase` is a
+`BindableService` and Boot registers every such bean.
+
+```java
+@Override
+public void getQuote(QuoteRequest request, StreamObserver<Quote> observer) {
+    observer.onNext(toProto(source.at(request.getSymbol(), 0)));
+    observer.onCompleted();
+}
+```
+
+**RSocket** ([`RSocketQuoteController`](../src/main/java/com/ankurm/protocols/RSocketQuoteController.java))
+is annotation routing, and the interaction model *is the return type*:
+
+```java
+@MessageMapping("quote")
+public Mono<Quote> quote(String symbol) { ... }        // request/response
+
+@MessageMapping("quotes")
+public Flux<Quote> quotes(StreamSpec spec) { ... }     // request/stream
+```
+
+**Raw WebSocket** ([`WebSocketQuoteHandler`](../src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java))
+is a `switch` on a string, because there is nothing else:
+
+```java
+switch (parts[0]) {
+    case "QUOTE"  -> send(session, source.at(parts[1], 0));
+    case "STREAM" -> { for (int i = 0; i < count; i++) { send(...); } }
+    ...
+}
+```
+
+That third block is the honest summary of what a raw WebSocket gives you: a bidirectional pipe
+for text or bytes. Everything above it &mdash; routing, correlation, errors, versioning, demand
+&mdash; is yours to invent, and inventing it is how teams end up with a private, undocumented
+protocol that only its authors can debug. STOMP exists to stop that, and is covered in
+[`../sse-websocket`](../../sse-websocket/README.md).
+
+## Boot 4 version notes
+
+Boot 4.1.1's BOM pins **grpc-java 1.83.1** and **protobuf-java 4.35.1**, both deliberately older
+than the newest releases on Maven Central. Overriding them independently is how you get a
+`NoSuchMethodError` between grpc-java and its shaded Netty. Note that these have moved since Boot
+4.1.0, which pinned 1.80.0 and 4.34.2 &mdash; so a `spring.grpc.*` guide written against 4.1.0 is
+already describing different jars.
+
+RSocket is **rsocket-java 1.1.5**, and `spring-boot-starter-rsocket` brings
+`jackson-dataformat-cbor` on purpose: CBOR, not JSON, is Spring's default RSocket data mime type.
+
+---
+
+Next: [2. How they measure](02-benchmarks.md)
@@ -0,0 +1,74 @@
+# 2. How they measure
+
+Previous: [1. Three protocols](01-three-protocols.md) &middot; Next: [3. Payload](03-payload.md)
+
+---
+
+## Read this before the numbers
+
+Client and server are **the same process on one machine**, talking over loopback: two cores,
+3.9&nbsp;GB, JDK 25, Boot 4.1.1. That has three consequences, and ignoring them makes the numbers
+say the opposite of the truth.
+
+1. **There is no network.** On a real link, a 35-byte message and an 83-byte message differ by
+   more than the framing cost, and protobuf's size advantage starts paying. Here it does not.
+2. **A protocol that does less looks faster.** WebSocket wins both benchmarks below partly because
+   it has no flow control, no deadlines, no schema and no per-call metadata. That is a real cost
+   difference, and it is also exactly what you give up.
+3. **Ratios, not absolutes.** Absolute microsecond figures from a two-core sandbox mean nothing
+   for your hardware. The ordering and the rough factors are what transfer.
+
+Re-measured on every `./scripts/run-all.sh`, and re-measured **in their own JVM invocation**.
+That is not tidiness. The back-pressure tests in [chapter 4](04-backpressure.md) leave unbounded
+producers spinning, and running the throughput benchmark after them on a two-core box halved
+every number — gRPC fell from 44 801 msgs/s to 18 477. If you take one operational lesson from
+this module rather than a protocol one, let it be that: a benchmark that shares a machine with
+anything is measuring the machine.
+
+## Request/response
+
+5&nbsp;000 sequential calls after 2&nbsp;000 warm-up
+([`request-response.txt`](output/request-response.txt)):
+
+```
+gRPC         n=5000   p50=   291.3 us  p99=   1015.3 us  mean=   324.4 us  ~3,083 calls/s
+RSocket      n=5000   p50=   300.9 us  p99=   1337.4 us  mean=   350.9 us  ~2,850 calls/s
+WebSocket    n=5000   p50=   124.3 us  p99=   1447.8 us  mean=   172.8 us  ~5,786 calls/s
+```
+
+**gRPC and RSocket are the same speed.** Within noise across runs the two swap places; treat them
+as indistinguishable for request/response on a warm connection.
+
+**Raw WebSocket is about twice as fast at the median &mdash; and has the worst tail.** p50 of
+124&nbsp;&micro;s against a p99 of 1&nbsp;448&nbsp;&micro;s is a 12x spread. gRPC's is 3.5x. The
+median is cheap because a text frame over an already-open socket is nearly free; the tail is bad
+because nothing is scheduling anything, so a GC pause or a scheduling hiccup lands on the request
+undiluted.
+
+If your SLO is a percentile rather than an average &mdash; and it should be &mdash; that table
+does not say what a first glance suggests.
+
+## Server streaming
+
+50&nbsp;000 messages on one connection
+([`stream-throughput.txt`](output/stream-throughput.txt)):
+
+```
+gRPC          50,000 msgs in  1.116 s  =      44,801 msgs/s  ( 22.32 us each)
+RSocket       50,000 msgs in  0.702 s  =      71,175 msgs/s  ( 14.05 us each)
+WebSocket     50,000 msgs in  0.459 s  =     109,022 msgs/s  (  9.17 us each)
+```
+
+Same ordering, wider gaps: WebSocket about 2.4x gRPC, RSocket about 1.6x.
+
+Now read the WebSocket row again with its harness in view. The client's inbox is an
+`ArrayBlockingQueue` sized to hold **all fifty thousand messages**, because there was nothing else
+to do with them &mdash; a raw WebSocket consumer that falls behind can buffer or drop, and those
+are the only two options. WebSocket "won" this benchmark by buffering the entire stream in the
+client's heap.
+
+That is not a rhetorical point. It is the next chapter.
+
+---
+
+Previous: [1. Three protocols](01-three-protocols.md) &middot; Next: [3. Payload](03-payload.md)
@@ -0,0 +1,56 @@
+# 3. Payload: the one comparison that needs no benchmark
+
+Previous: [2. Benchmarks](02-benchmarks.md) &middot; Next: [4. Back-pressure](04-backpressure.md)
+
+---
+
+Five fields &mdash; a symbol, a sequence number, two doubles and a timestamp &mdash; encoded three
+ways ([`payload-sizes.txt`](output/payload-sizes.txt)):
+
+```
+protobuf :  35 bytes  0a044141504c102a197b14ae47e11a5940215c8fc2f5281c5940288080abb4fef39203
+JSON     :  83 bytes  {"symbol":"AAPL","seq":42,"bid":100.42,"ask":100.44,"epochMicros":1772000000000000}
+CBOR     :  67 bytes  bf6673796d626f6c644141504c63736571182a63626964fb40591ae147ae147b6361736bfb40591c28f5c28f5c6b65706f63684d6963726f731b00064b9fe68ac000ff
+JSON is 2.37x protobuf; CBOR is 1.91x protobuf
+```
+
+This is a property of the formats, not of a machine, so it is the one row of the comparison that
+transfers to your hardware unchanged.
+
+**Protobuf is 35 bytes because field names are integers.** `0a04 41 41 50 4c` is field 1, length
+4, `AAPL`. There is no `"symbol"` on the wire at all &mdash; the name lives in the `.proto` file
+that both sides compiled against. That is where the size comes from, and it is also the whole
+argument about schemas: the saving and the coupling are the same fact.
+
+**CBOR is 67 bytes and still writes the field names.** `6673796d626f6c` is a 6-character text
+string, `symbol`. CBOR is binary JSON, not a schema format: it saves the punctuation and encodes
+numbers compactly, and it keeps every key. So its advantage over JSON is real but modest, and it
+does not require the coupling.
+
+Concretely, on a stream of a million quotes: 35&nbsp;MB, 67&nbsp;MB, 83&nbsp;MB. On a loopback
+that difference disappears into memory bandwidth &mdash; which is exactly why the benchmarks in
+[chapter 2](02-benchmarks.md) do not reward it and a real link would.
+
+Two practical notes:
+
+- **Spring's RSocket default is CBOR**, not JSON, which is why
+  `spring-boot-starter-rsocket` brings `jackson-dataformat-cbor`. If you are debugging with
+  `tcpdump` and expecting to read your payloads, that is why you cannot. The mechanism is codec
+  ordering, printed by
+  [`RSocketDefaultsTest`](../src/test/java/com/ankurm/protocols/RSocketDefaultsTest.java):
+
+  ```
+  encoders : [CharSequenceEncoder, ByteBufferEncoder, ByteArrayEncoder, DataBufferEncoder, JacksonCborEncoder, JacksonJsonEncoder]
+  bare RSocketStrategies.create() : [CharSequenceEncoder, ByteBufferEncoder, ByteArrayEncoder, DataBufferEncoder]
+  ```
+
+  CBOR sits *ahead of* JSON in the Boot-configured list, which is the whole of why it wins. Note
+  also that a bare `RSocketStrategies.create()` carries neither — an `RSocketRequester` built
+  without injecting Boot's strategies cannot encode your objects at all. And the class names
+  follow the Boot 4 rule: `JacksonCborEncoder`, not `Jackson2…`.
+- **You can put protobuf on RSocket.** The encoding and the protocol are independent choices;
+  the row above is what each stack does *by default*, not what it is capable of.
+
+---
+
+Previous: [2. Benchmarks](02-benchmarks.md) &middot; Next: [4. Back-pressure](04-backpressure.md)
@@ -0,0 +1,112 @@
+# 4. Back-pressure: the measurement that decides it
+
+Previous: [3. Payload](03-payload.md) &middot; Next: [5. Choosing](05-choosing.md)
+
+---
+
+Every protocol comparison eventually reduces to one question, and it is not throughput. It is:
+
+> **When the consumer stops keeping up, what happens?**
+
+[`BackPressureTest`](../src/test/java/com/ankurm/protocols/BackPressureTest.java) asks it the
+simplest way there is. Each server offers an unbounded stream. Each client takes a hundred
+messages and then does nothing. How many did the server produce?
+
+The sweep over quiet periods is what makes the answer conclusive. A number that stays flat as the
+wait grows means something bounded the producer. A number that keeps climbing means nothing did.
+
+[`backpressure.txt`](output/backpressure.txt):
+
+```
+protocol       quiet 0.5s   quiet 1s   quiet 2s
+RSocket               100        100        100
+gRPC              103,425    449,318  1,208,219
+WebSocket          19,664    143,870    320,255
+```
+
+RSocket produced **exactly one hundred**, at every quiet period. gRPC produced over a million in
+two seconds. Raw WebSocket, three hundred thousand. These are re-measured on every run and the
+absolute counts move by tens of percent; the three orders of magnitude between the rows do not.
+
+(The growth is faster than linear because the loop is still being JIT-compiled during the shortest
+run. The shape is the finding, not the exponent.)
+
+## Why RSocket stops
+
+`request(n)` is a **frame on the wire**. When the client's subscriber calls `subscription.request(100)`,
+a REQUEST_N frame carries that number to the server, and Reactor's `Flux.generate` is called
+exactly a hundred times and then not again. Nothing blocks; there is no buffer to size and no
+thread parked. The producer simply is not invoked.
+
+That is what "Reactive Streams over the network" means, and it is the entire reason RSocket
+exists. The same property is why RSocket's streaming throughput in
+[chapter 2](02-benchmarks.md) is respectable rather than the highest: it is doing real work per
+batch that the others are not doing at all.
+
+## Why the other two do not
+
+**Raw WebSocket has no mechanism.** The frame protocol has no notion of demand. The server writes
+until something underneath refuses &mdash; a socket buffer, Spring's `sendBufferSizeLimit`, the
+heap. Every one of those 320&nbsp;255 messages was serialised and written for a client that had
+asked for a hundred.
+
+**gRPC has HTTP/2 flow control, and it did not help.** This is the result I did not expect.
+`StreamObserver.onNext` on a gRPC server **does not block**, so a plain server-streaming loop keeps
+producing regardless of what the client is doing.
+
+The documented answer is `ServerCallStreamObserver.isReady()`, and this module tests it rather than
+citing it ([`grpc-isready.txt`](output/grpc-isready.txt)):
+
+```
+handler                      quiet 0.5s   quiet 2s
+plain onNext loop               258,145  1,421,778
+isReady() checked before write    667,259  1,044,018
+```
+
+**It did not bound the producer.** The `isReady()` row is not lower than the plain one in any
+useful sense &mdash; it produced 667&nbsp;thousand messages in half a second for a client that
+wanted a hundred &mdash; and across runs it has come out both far above and far below the plain
+loop, which is its own kind of answer. I
+have not established whether the surplus accumulates in the server's outbound queue, in the client
+transport, or is discarded after the deframer, and I am not going to assert a mechanism I have not
+read the source for.
+
+What the numbers do establish is narrower and still worth knowing: **`isReady()` is not a demand
+signal**, and on a fast link with a stalled consumer it does not behave like one. If you are
+streaming from gRPC to a consumer that can fall behind, the bound has to come from your own
+design &mdash; a paged API, an explicit "send me more" message on a bidirectional stream, or a
+`count` in the request as `StreamQuotes` uses.
+
+## The honest caveat
+
+A loopback connection is the worst case for this test in the fairest possible way: the wire drains
+instantly, so nothing downstream ever pushes back. On a slow real link, TCP and the HTTP/2 window
+*would* eventually stop the gRPC and WebSocket producers &mdash; by blocking a thread or by
+filling a buffer, but they would stop.
+
+That does not rescue them, because the failure mode you actually meet is not "the network is
+slow", it is "one consumer of forty is slow", and a shared server discovers that as latency for
+everyone else or as a heap that will not shrink. The difference RSocket makes is that the slow
+consumer's slowness is *expressed*, in a number, that the producer can act on before anything
+fills up.
+
+## The other thing gRPC does not do for you
+
+Abandoning a gRPC stream produced this, repeatedly, in the server log:
+
+```
+java.lang.IllegalStateException: Failed to close the call
+Caused by: java.lang.IllegalStateException: call already closed
+```
+
+That is the cancellation story from
+[the Spring gRPC article](https://ankurm.com/spring-grpc-spring-boot-4/) showing up in practice:
+gRPC does not interrupt your thread when a client leaves, it sets a flag on the `Context`, and a
+handler that touches an already-closed stream throws. The loop in
+[`GrpcQuoteService`](../src/main/java/com/ankurm/protocols/GrpcQuoteService.java) checks
+`Context.current().isCancelled()` and returns without calling `onCompleted` for exactly this
+reason.
+
+---
+
+Previous: [3. Payload](03-payload.md) &middot; Next: [5. Choosing](05-choosing.md)
@@ -0,0 +1,63 @@
+# 5. Choosing
+
+Previous: [4. Back-pressure](04-backpressure.md)
+
+---
+
+## The decision table
+
+| If | Pick | Because |
+|---|---|---|
+| Service-to-service RPC, polyglot team, you want a schema | **gRPC** | The `.proto` is a contract other languages generate from. Deadlines are absolute and propagate. Boot 4 makes it a first-class starter. |
+| A stream whose consumer can fall behind | **RSocket** | `request(n)` is the only demand signal that crosses the wire. 100 vs 1.3 million is not a tuning difference. |
+| A browser is one of the peers | **WebSocket**, with STOMP on it | gRPC needs grpc-web and a proxy. RSocket over WebSocket works but the client ecosystem is thin. |
+| Server pushes, client never speaks | **none of these** — use SSE | See [`../sse-websocket`](../../sse-websocket/README.md). |
+| You want the lowest possible median latency and control both ends | **raw WebSocket** | It is fastest here because it does the least. Read the p99 before deciding that is what you want. |
+| Long-lived bidirectional exchange with independent streams | **RSocket channel** or **gRPC bidi** | Both give you two independent streams; only RSocket gives each of them demand. |
+
+## What each one really costs
+
+**gRPC** costs a build step, a schema you must version, and an operational story for HTTP/2 through
+your proxies. In exchange you get the best-supported cross-language RPC there is, and deadlines
+&mdash; which is a bigger deal than it sounds, because an absolute deadline that propagates across
+hops is the single most effective defence against one slow dependency taking down a system.
+
+**RSocket** costs ecosystem. It is the least deployed of the three, the client libraries outside
+Java and JavaScript are thin, and if you are not already reactive, `Mono` and `Flux` in your
+service signatures is a real commitment. In exchange you get the only demand signalling in the
+list, plus resumption and leasing.
+
+**Raw WebSocket** costs everything you then invent: routing, correlation, error shape, versioning,
+and the demand mechanism you will eventually need. That is the cost STOMP exists to prevent, and
+if you find yourself designing a text protocol with a command word at the front &mdash; as
+[`WebSocketQuoteHandler`](../src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java)
+deliberately does &mdash; that is the signal to stop and use one.
+
+## Should you use any of them?
+
+Three cases where the answer is no:
+
+- **Request/response between two services that both speak HTTP already.** A REST endpoint with a
+  `RestClient` and a timeout is fine, and everything in your organisation already knows how to
+  observe, cache, proxy and debug it. Reach for gRPC when the schema or the deadline propagation
+  is the thing you need, not because it benchmarks faster.
+- **Streaming that is really batching.** If the consumer processes a page at a time anyway,
+  paginated HTTP has back-pressure for free: the consumer asks for the next page when it is ready.
+  That is `request(n)` with a URL.
+- **Events that outlive the connection.** None of these three persist anything. A dropped
+  connection loses the messages in flight. If that matters, the answer is a broker &mdash;
+  [`../kafka-basics`](../../kafka-basics/README.md),
+  [`../rabbitmq`](../../rabbitmq/README.md), and the comparison in
+  [`../broker-comparison`](../../broker-comparison/README.md).
+
+## And the benchmark caveat, one more time
+
+Every number in this module was measured with client and server in one JVM on one machine. That
+deletes the network, which is where protobuf's 35 bytes would pay and where the tail latencies
+would look completely different. The measurement that does transfer is
+[chapter 4](04-backpressure.md), because "how many did the server produce for a consumer that
+wanted a hundred" is a property of the protocol and not of the link.
+
+---
+
+Previous: [4. Back-pressure](04-backpressure.md)
@@ -0,0 +1,10 @@
+=== unbounded stream; client takes 100, then goes quiet ===
+protocol       quiet 0.5s   quiet 1s   quiet 2s
+RSocket               100        100        100
+gRPC              103,425    449,318  1,208,219
+WebSocket          19,664    143,870    320,255
+RSocket is flat: the server produced exactly what request(n) asked for.
+gRPC grows 11.7x between 0.5 s and 2 s; WebSocket grows 16.3x.
+Growth that tracks the wait, rather than settling at a buffer size,
+means nothing bounded the producer. (Growth is faster than linear
+because the loop is still being JIT-compiled during the first run.)
@@ -0,0 +1,4 @@
+=== gRPC server streaming, client takes 100 then stops ===
+handler                      quiet 0.5s   quiet 2s
+plain onNext loop               258,145  1,421,778
+isReady() checked before write    667,259  1,044,018
@@ -0,0 +1,5 @@
+=== one Quote, five fields, three encodings ===
+protobuf :  35 bytes  0a044141504c102a197b14ae47e11a5940215c8fc2f5281c5940288080abb4fef39203
+JSON     :  83 bytes  {"symbol":"AAPL","seq":42,"bid":100.42,"ask":100.44,"epochMicros":1772000000000000}
+CBOR     :  67 bytes  bf6673796d626f6c644141504c63736571182a63626964fb40591ae147ae147b6361736bfb40591c28f5c28f5c6b65706f63684d6963726f731b00064b9fe68ac000ff
+JSON is 2.37x protobuf; CBOR is 1.91x protobuf
@@ -0,0 +1,4 @@
+=== request/response, 5000 sequential calls after 2000 warm-up, loopback, JDK 25.0.4.1 ===
+gRPC         n=5000   p50=   302.0 us  p99=    994.7 us  mean=   335.1 us  ~2,984 calls/s
+RSocket      n=5000   p50=   265.9 us  p99=   1166.2 us  mean=   311.9 us  ~3,206 calls/s
+WebSocket    n=5000   p50=   122.8 us  p99=   1893.9 us  mean=   184.8 us  ~5,410 calls/s
@@ -0,0 +1,4 @@
+=== Boot-configured RSocketStrategies ===
+encoders : [CharSequenceEncoder, ByteBufferEncoder, ByteArrayEncoder, DataBufferEncoder, JacksonCborEncoder, JacksonJsonEncoder]
+decoders : [StringDecoder, ByteBufferDecoder, ByteArrayDecoder, DataBufferDecoder, JacksonCborDecoder, JacksonJsonDecoder]
+bare RSocketStrategies.create() encoders : [CharSequenceEncoder, ByteBufferEncoder, ByteArrayEncoder, DataBufferEncoder]
@@ -0,0 +1,4 @@
+=== server streaming, 50000 messages on one connection, loopback ===
+gRPC          50,000 msgs in  1.341 s  =      37,285 msgs/s  ( 26.82 us each)
+RSocket       50,000 msgs in  0.900 s  =      55,580 msgs/s  ( 17.99 us each)
+WebSocket     50,000 msgs in  0.839 s  =      59,630 msgs/s  ( 16.77 us each)
@@ -0,0 +1,6 @@
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.321 s -- in com.ankurm.protocols.PayloadSizeTest
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 11.76 s -- in com.ankurm.protocols.RequestResponseBenchmarkTest
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 11.29 s -- in com.ankurm.protocols.BackPressureTest
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 7.409 s -- in com.ankurm.protocols.GrpcIsReadyTest
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 6.474 s -- in com.ankurm.protocols.RSocketDefaultsTest
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 7.244 s -- in com.ankurm.protocols.StreamThroughputBenchmarkTest
@@ -0,0 +1,95 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>4.1.1</version>
+    <relativePath/>
+  </parent>
+
+  <groupId>com.ankurm</groupId>
+  <artifactId>protocol-comparison</artifactId>
+  <version>1.0</version>
+  <packaging>jar</packaging>
+
+  <properties>
+    <java.version>25</java.version>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <dependencies>
+    <!-- One application speaks all three protocols, so the comparison is not confounded by
+         different JVMs, different heaps or different warm-up states. -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-webmvc</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-websocket</artifactId>
+    </dependency>
+    <!-- RSocket. The starter drags in reactor-netty and jackson-dataformat-cbor; CBOR is not
+         an accident, it is RSocket's default data mime type in Spring. -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-rsocket</artifactId>
+    </dependency>
+    <!-- gRPC. Two starters, both Boot-managed. See ankurm.com/spring-grpc-spring-boot-4/ -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-grpc-server</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-grpc-client</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-jackson</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor</groupId>
+      <artifactId>reactor-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <!-- protoc and the gRPC codegen plugin resolve as Maven artifacts for this OS and
+           architecture. Nothing to install locally. Both versions come from the Boot BOM;
+           overriding them independently is how you get a NoSuchMethodError between grpc-java
+           and its shaded Netty. -->
+      <plugin>
+        <groupId>io.github.ascopes</groupId>
+        <artifactId>protobuf-maven-plugin</artifactId>
+        <version>3.9.0</version>
+        <configuration>
+          <protocVersion>${protobuf-java.version}</protocVersion>
+          <binaryMavenPlugins>
+            <binaryMavenPlugin>
+              <groupId>io.grpc</groupId>
+              <artifactId>protoc-gen-grpc-java</artifactId>
+              <version>${grpc-java.version}</version>
+            </binaryMavenPlugin>
+          </binaryMavenPlugins>
+        </configuration>
+        <executions>
+          <execution><goals><goal>generate</goal></goals></execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>
@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+# Regenerate every file under docs/output/. Needs a JDK and nothing else: protoc and the gRPC
+# codegen plugin resolve as Maven artifacts, and all three servers start inside the test JVM.
+#
+# The benchmark numbers are NOT deterministic. They are re-measured on every run and the ratios
+# between the three protocols are what the documentation quotes.
+set -eu
+cd "$(dirname "$0")/.."
+LOG=/tmp/protocol-comparison-test.log
+
+# Everything, for the test count and the non-timing transcripts.
+MAVEN_OPTS=${MAVEN_OPTS:--Xmx900m} mvn -B test > "$LOG" 2>&1
+
+# The two timing benchmarks are then re-run ALONE. This is not tidiness: the back-pressure and
+# isReady tests leave unbounded producers spinning, and on a two-core box that halves every
+# throughput number measured after them. Benchmarks share a JVM with nothing.
+BENCH=/tmp/protocol-comparison-bench.log
+MAVEN_OPTS=${MAVEN_OPTS:--Xmx900m} mvn -B test \
+  -Dtest=RequestResponseBenchmarkTest,StreamThroughputBenchmarkTest > "$BENCH" 2>&1
+
+# Interleaved log lines from other contexts have to come out, or the transcripts are unreadable.
+clean() { grep -vE '^2026-|^\s*$|^WARNING|^[[:space:]]+at |^Caused by|^java\.|^org\.|Initializing|Completed initialization'; }
+
+sed -n '/=== one Quote, five fields/,/^JSON is /p'        "$LOG" | clean > docs/output/payload-sizes.txt
+sed -n '/=== request\/response/,/^WebSocket  /p'       "$BENCH" | clean > docs/output/request-response.txt
+sed -n '/=== server streaming, 50000/,/^WebSocket  /p' "$BENCH" | clean > docs/output/stream-throughput.txt
+{ grep -m1 '=== unbounded stream' "$LOG"; grep -m1 '^protocol ' "$LOG"
+  grep -m1 '^RSocket  *[0-9]' "$LOG"; grep -m1 '^gRPC  *[0-9]' "$LOG"
+  grep -m1 '^WebSocket  *[0-9]' "$LOG"
+  sed -n '/^RSocket is flat/,/JIT-compiled/p' "$LOG"; } > docs/output/backpressure.txt
+{ grep -m1 '=== gRPC server streaming' "$LOG"; grep -m1 '^handler ' "$LOG"
+  grep -m1 '^plain onNext loop' "$LOG"; grep -m1 '^isReady() checked' "$LOG"; } \
+  > docs/output/grpc-isready.txt
+
+sed -n '/=== Boot-configured RSocketStrategies/,/^bare RSocketStrategies/p' "$LOG" | clean \
+  > docs/output/rsocket-codecs.txt
+
+grep -E 'Tests run:.*in com\.ankurm' "$LOG" | sed 's/^\[INFO\] //' > docs/output/tests.txt
+
+echo "regenerated:"; ls -1 docs/output/
@@ -0,0 +1,121 @@
+package com.ankurm.protocols;
+
+import com.ankurm.protocols.grpc.Quote;
+import com.ankurm.protocols.grpc.QuoteRequest;
+import com.ankurm.protocols.grpc.QuoteServiceGrpc;
+import com.ankurm.protocols.grpc.StreamRequest;
+import io.grpc.Context;
+import io.grpc.stub.ServerCallStreamObserver;
+import io.grpc.stub.StreamObserver;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.springframework.stereotype.Service;
+
+/**
+ * The gRPC side. No registration code: the generated {@code ImplBase} is a
+ * {@code BindableService} and Boot registers every such bean with the server.
+ *
+ * <p>The cancellation check in {@link #streamQuotes} is not decoration. gRPC does not interrupt
+ * your thread when a client goes away &mdash; it sets a flag on the {@link Context} &mdash; so a
+ * loop that never looks keeps producing into a stream nobody is reading. That is covered at
+ * length in <a href="https://ankurm.com/spring-grpc-spring-boot-4/">the Spring gRPC article</a>;
+ * here it matters because the back-pressure benchmark deliberately abandons a stream.
+ *
+ * @see <a href="../../../../../docs/04-backpressure.md">docs/04-backpressure.md</a>
+ */
+@Service
+public class GrpcQuoteService extends QuoteServiceGrpc.QuoteServiceImplBase {
+
+    /** Counts what the SERVER produced, comparable with the RSocket and WebSocket counters. */
+    static final AtomicLong PRODUCED = new AtomicLong();
+
+    private final QuoteSource source;
+
+    GrpcQuoteService(QuoteSource source) {
+        this.source = source;
+    }
+
+    @Override
+    public void getQuote(QuoteRequest request, StreamObserver<Quote> observer) {
+        observer.onNext(toProto(source.at(request.getSymbol(), 0)));
+        observer.onCompleted();
+    }
+
+    /**
+     * An effectively unbounded stream, for the back-pressure comparison. A consumer that stops
+     * calling {@code next()} closes the HTTP/2 flow-control window, and this loop blocks inside
+     * {@code onNext} &mdash; so the server stops producing, but it stops by blocking a thread
+     * rather than by being asked to stop.
+     */
+    @Override
+    public void streamUnbounded(QuoteRequest request, StreamObserver<Quote> observer) {
+        PRODUCED.set(0);
+        while (!Context.current().isCancelled()) {
+            observer.onNext(toProto(source.at(request.getSymbol(), PRODUCED.get())));
+            PRODUCED.incrementAndGet();
+        }
+    }
+
+    /**
+     * The fix, and the reason the unfixed version is worth measuring.
+     *
+     * <p>{@code StreamObserver.onNext} on a gRPC server <strong>never blocks</strong>. If the
+     * client is not reading, the message is queued in the server's outbound buffer and the loop
+     * keeps going &mdash; HTTP/2 flow control governs the wire, not your code. The only thing
+     * that connects the two is {@link ServerCallStreamObserver#isReady()}, and using it means
+     * restructuring the handler around {@code setOnReadyHandler} rather than writing a loop.
+     *
+     * <p>The spin here is deliberately the simplest possible demonstration rather than the
+     * shape you would ship; a real implementation registers an on-ready handler and returns.
+     */
+    @Override
+    public void streamUnboundedReady(QuoteRequest request, StreamObserver<Quote> observer) {
+        PRODUCED.set(0);
+        ServerCallStreamObserver<Quote> ready = (ServerCallStreamObserver<Quote>) observer;
+        while (!ready.isCancelled()) {
+            if (!ready.isReady()) {
+                Thread.onSpinWait();
+                continue;
+            }
+            ready.onNext(toProto(source.at(request.getSymbol(), PRODUCED.get())));
+            PRODUCED.incrementAndGet();
+        }
+    }
+
+    @Override
+    public void produced(QuoteRequest request, StreamObserver<com.ankurm.protocols.grpc.ProducedCount> observer) {
+        observer.onNext(com.ankurm.protocols.grpc.ProducedCount.newBuilder()
+                .setCount(PRODUCED.get()).build());
+        observer.onCompleted();
+    }
+
+    @Override
+    public void streamQuotes(StreamRequest request, StreamObserver<Quote> observer) {
+        for (int i = 0; i < request.getCount(); i++) {
+            if (Context.current().isCancelled()) {
+                // Do NOT call onCompleted/onError here: the stream is already closed and
+                // touching it throws IllegalStateException. Just return.
+                return;
+            }
+            observer.onNext(toProto(source.at(request.getSymbol(), i)));
+            if (request.getDelayMs() > 0) {
+                try {
+                    Thread.sleep(request.getDelayMs());
+                }
+                catch (InterruptedException ex) {
+                    Thread.currentThread().interrupt();
+                    return;
+                }
+            }
+        }
+        observer.onCompleted();
+    }
+
+    static Quote toProto(com.ankurm.protocols.Quote q) {
+        return Quote.newBuilder()
+                .setSymbol(q.symbol()).setSeq(q.seq())
+                .setBid(q.bid()).setAsk(q.ask())
+                .setEpochMicros(q.epochMicros())
+                .build();
+    }
+}
@@ -0,0 +1,22 @@
+package com.ankurm.protocols;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+ * One application, three protocols, one set of data.
+ *
+ * <p>gRPC on 9090, RSocket on 7000, WebSocket on the servlet port. Serving the same two
+ * operations from the same JVM is the only way to make a benchmark mean anything: the heap, the
+ * JIT state, the CPU and the data generator are shared, so a difference in the numbers is a
+ * difference in the transports.
+ *
+ * @see <a href="../../../../../docs/01-three-protocols.md">docs/01-three-protocols.md</a>
+ */
+@SpringBootApplication
+public class ProtocolComparisonApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(ProtocolComparisonApplication.class, args);
+    }
+}
@@ -0,0 +1,8 @@
+package com.ankurm.protocols;
+
+/**
+ * The JSON/CBOR shape, field-for-field identical to the protobuf {@code Quote} message, so the
+ * encoded sizes in {@code docs/output/payload-sizes.txt} compare like with like.
+ */
+public record Quote(String symbol, long seq, double bid, double ask, long epochMicros) {
+}
@@ -0,0 +1,22 @@
+package com.ankurm.protocols;
+
+import java.time.Instant;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * Deterministic quote generation, shared by all three protocol adapters.
+ *
+ * <p>Deliberately cheap and deliberately allocation-free apart from the record: a benchmark of
+ * three transports must not spend its time in the thing behind them.
+ */
+@Component
+public class QuoteSource {
+
+    public Quote at(String symbol, long seq) {
+        double base = 100.0 + (seq % 97) * 0.01;
+        Instant now = Instant.now();
+        return new Quote(symbol, seq, base, base + 0.02,
+                now.getEpochSecond() * 1_000_000L + now.getNano() / 1_000L);
+    }
+}
@@ -0,0 +1,72 @@
+package com.ankurm.protocols;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.stereotype.Controller;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * The RSocket side. The same two operations, and one extra that exists only to make
+ * back-pressure visible.
+ *
+ * <p>Note what the signatures say that the other two protocols' cannot: {@code Mono} means
+ * request/response, {@code Flux} means request/stream, and the choice of interaction model is
+ * the return type rather than a separate declaration. RSocket has four (fire-and-forget,
+ * request/response, request/stream, channel) and Spring picks from the method shape.
+ *
+ * @see <a href="../../../../../docs/04-backpressure.md">docs/04-backpressure.md</a>
+ */
+@Controller
+public class RSocketQuoteController {
+
+    private static final Logger log = LoggerFactory.getLogger(RSocketQuoteController.class);
+
+    /** Counts what the SERVER produced, which is the number the back-pressure test cares about. */
+    static final AtomicLong PRODUCED = new AtomicLong();
+
+    private final QuoteSource source;
+
+    RSocketQuoteController(QuoteSource source) {
+        this.source = source;
+    }
+
+    @MessageMapping("quote")
+    public Mono<Quote> quote(String symbol) {
+        return Mono.just(source.at(symbol, 0));
+    }
+
+    @MessageMapping("quotes")
+    public Flux<Quote> quotes(StreamSpec spec) {
+        Flux<Quote> flux = Flux.range(0, spec.count()).map(i -> source.at(spec.symbol(), i));
+        return spec.delayMs() > 0 ? flux.delayElements(Duration.ofMillis(spec.delayMs())) : flux;
+    }
+
+    /**
+     * An unbounded generator. It emits only when the transport asks, so the count it reaches is
+     * a direct read-out of how much the consumer requested &mdash; which is the whole argument
+     * for RSocket.
+     */
+    @MessageMapping("quotes.unbounded")
+    public Flux<Quote> unbounded(String symbol) {
+        PRODUCED.set(0);
+        return Flux.generate(() -> 0L, (seq, sink) -> {
+            sink.next(source.at(symbol, seq));
+            PRODUCED.incrementAndGet();
+            return seq + 1;
+        });
+    }
+
+    @MessageMapping("produced")
+    public Mono<Long> produced() {
+        return Mono.just(PRODUCED.get());
+    }
+
+    /** Request payload for {@code quotes}. */
+    public record StreamSpec(String symbol, int count, int delayMs) {
+    }
+}
@@ -0,0 +1,27 @@
+package com.ankurm.protocols;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.config.annotation.EnableWebSocket;
+import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
+import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
+
+/**
+ * {@code @EnableWebSocket}, not {@code @EnableWebSocketMessageBroker}: this module wants the raw
+ * transport, with no broker in the path, so the benchmark measures a socket rather than a routing
+ * layer. The {@code sse-websocket} module in this repository is the STOMP version.
+ */
+@Configuration
+@EnableWebSocket
+public class WebSocketConfig implements WebSocketConfigurer {
+
+    private final WebSocketQuoteHandler handler;
+
+    WebSocketConfig(WebSocketQuoteHandler handler) {
+        this.handler = handler;
+    }
+
+    @Override
+    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
+        registry.addHandler(handler, "/quotes").setAllowedOriginPatterns("*");
+    }
+}
@@ -0,0 +1,94 @@
+package com.ankurm.protocols;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+import tools.jackson.databind.ObjectMapper;
+
+/**
+ * The WebSocket side, with no STOMP and no framework on top: a two-word text protocol, so the
+ * benchmark measures the transport and not a broker.
+ *
+ * <pre>
+ *   QUOTE &lt;symbol&gt;                       -&gt; one JSON quote
+ *   STREAM &lt;symbol&gt; &lt;count&gt; &lt;delayMs&gt;    -&gt; count JSON quotes, then "END"
+ *   UNBOUNDED &lt;symbol&gt;                   -&gt; quotes until the session dies
+ * </pre>
+ *
+ * <p>The absence worth noticing is in {@code UNBOUNDED}: there is nowhere for a consumer to say
+ * how many it wants. {@code sendMessage} either succeeds, blocks, or eventually throws when
+ * Spring's send buffer limit is exceeded. That is not a gap in this handler &mdash; it is what a
+ * raw WebSocket offers.
+ *
+ * @see <a href="../../../../../docs/04-backpressure.md">docs/04-backpressure.md</a>
+ */
+@Component
+public class WebSocketQuoteHandler extends TextWebSocketHandler {
+
+    /** Counts what the SERVER wrote, comparable with RSocket's PRODUCED. */
+    static final AtomicLong PRODUCED = new AtomicLong();
+
+    private final QuoteSource source;
+    private final ObjectMapper mapper;
+
+    WebSocketQuoteHandler(QuoteSource source, ObjectMapper mapper) {
+        this.source = source;
+        this.mapper = mapper;
+    }
+
+    @Override
+    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
+        String[] parts = message.getPayload().trim().split("\\s+");
+        switch (parts[0]) {
+            case "QUOTE" -> send(session, source.at(parts[1], 0));
+            case "STREAM" -> {
+                int count = Integer.parseInt(parts[2]);
+                int delay = parts.length > 3 ? Integer.parseInt(parts[3]) : 0;
+                for (int i = 0; i < count; i++) {
+                    send(session, source.at(parts[1], i));
+                    if (delay > 0) {
+                        Thread.sleep(delay);
+                    }
+                }
+                session.sendMessage(new TextMessage("END"));
+            }
+            case "UNBOUNDED" -> {
+                PRODUCED.set(0);
+                // No requestN. Nothing here can be told to slow down: the loop writes until the
+                // socket, the send buffer or the send-time limit stops it. Catching the write
+                // failure keeps an abandoned stream from filling the log with one stack trace
+                // per run -- and note that catching it is the ONLY notification that arrives.
+                try {
+                    while (session.isOpen()) {
+                        send(session, source.at(parts[1], PRODUCED.get()));
+                        PRODUCED.incrementAndGet();
+                    }
+                }
+                catch (Exception ex) {
+                    // Broken pipe. The client left; there was no other way to find out.
+                }
+            }
+            case "PRODUCED" -> session.sendMessage(new TextMessage(Long.toString(PRODUCED.get())));
+            default -> session.sendMessage(new TextMessage("ERR unknown command"));
+        }
+    }
+
+    @Override
+    public void handleTransportError(WebSocketSession session, Throwable exception) {
+        // Where an over-buffered slow consumer surfaces: as a transport error on the writer,
+        // long after the consumer stopped reading.
+        PRODUCED.addAndGet(0);
+    }
+
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
+    }
+
+    private void send(WebSocketSession session, Quote quote) throws Exception {
+        session.sendMessage(new TextMessage(mapper.writeValueAsString(quote)));
+    }
+}
@@ -0,0 +1,45 @@
+syntax = "proto3";
+
+package quotes;
+
+option java_multiple_files = true;
+option java_package = "com.ankurm.protocols.grpc";
+
+// The same two operations the RSocket and WebSocket sides expose, so the comparison is between
+// transports rather than between designs.
+service QuoteService {
+  // Request/response.
+  rpc GetQuote (QuoteRequest) returns (Quote);
+  // Server streaming: count quotes, as fast as the transport allows.
+  rpc StreamQuotes (StreamRequest) returns (stream Quote);
+  // Never completes on its own. Used to show what stops a gRPC server producing.
+  rpc StreamUnbounded (QuoteRequest) returns (stream Quote);
+  // The same unbounded stream, but the server checks ServerCallStreamObserver.isReady()
+  // before every write. Same API, different memory profile.
+  rpc StreamUnboundedReady (QuoteRequest) returns (stream Quote);
+  // How many messages the server has produced on the current unbounded stream.
+  rpc Produced (QuoteRequest) returns (ProducedCount);
+}
+
+message ProducedCount {
+  int64 count = 1;
+}
+
+message QuoteRequest {
+  string symbol = 1;
+}
+
+message StreamRequest {
+  string symbol = 1;
+  int32 count = 2;
+  // Milliseconds to sleep between emissions; 0 means "as fast as possible".
+  int32 delay_ms = 3;
+}
+
+message Quote {
+  string symbol = 1;
+  int64 seq = 2;
+  double bid = 3;
+  double ask = 4;
+  int64 epoch_micros = 5;
+}
@@ -0,0 +1,22 @@
+spring:
+  application:
+    name: protocol-comparison
+  rsocket:
+    server:
+      # RSocket gets its own port and its own Reactor Netty server. TCP here; "websocket" is the
+      # other transport and is what you use when the client is a browser.
+      port: 7000
+      transport: tcp
+  grpc:
+    server:
+      # spring.grpc.server.port -- note the spring. prefix and that "port: -1" means in-process
+      # only. See the property table in ankurm.com/spring-grpc-spring-boot-4/
+      port: 9090
+
+server:
+  port: 8080
+
+logging:
+  level:
+    io.grpc: WARN
+    io.rsocket: WARN
@@ -0,0 +1,156 @@
+package com.ankurm.protocols;
+
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.ankurm.protocols.grpc.QuoteRequest;
+import com.ankurm.protocols.grpc.QuoteServiceGrpc;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscription;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+import reactor.core.publisher.BaseSubscriber;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The measurement that decides the article.
+ *
+ * <p>Each protocol serves an unbounded stream. Each client takes a hundred messages and then does
+ * nothing for a while. The question is: <strong>how many did the server produce?</strong>
+ *
+ * <p>The sweep over quiet periods is what makes the answer conclusive. A number that stays flat
+ * as the quiet period grows means something bounded the producer. A number that grows in
+ * proportion to the wait means <em>nothing did</em> &mdash; the producer is running flat out and
+ * the only limit is the clock.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"spring.rsocket.server.port=7003", "spring.grpc.server.port=9093"})
+class BackPressureTest {
+
+    private static final int WANTED = 100;
+    private static final long[] QUIET_MS = {500, 1_000, 2_000};
+
+    @LocalServerPort
+    int httpPort;
+
+    @Autowired
+    RSocketStrategies strategies;
+
+    @Test
+    void howManyDidTheServerProduce() throws Exception {
+        System.out.println("=== unbounded stream; client takes " + WANTED + ", then goes quiet ===");
+        System.out.printf("%-14s %10s %10s %10s%n", "protocol", "quiet 0.5s", "quiet 1s", "quiet 2s");
+
+        long[] rs = new long[QUIET_MS.length];
+        long[] gr = new long[QUIET_MS.length];
+        long[] ws = new long[QUIET_MS.length];
+        for (int i = 0; i < QUIET_MS.length; i++) {
+            rs[i] = rsocket(QUIET_MS[i]);
+            gr[i] = grpc(QUIET_MS[i]);
+            ws[i] = websocket(QUIET_MS[i]);
+        }
+        row("RSocket", rs);
+        row("gRPC", gr);
+        row("WebSocket", ws);
+
+        System.out.println();
+        System.out.println("RSocket is flat: the server produced exactly what request(n) asked for.");
+        System.out.printf("gRPC grows %.1fx between 0.5 s and 2 s; WebSocket grows %.1fx.%n",
+                gr[2] / (double) gr[0], ws[2] / (double) ws[0]);
+        System.out.println("Growth that tracks the wait, rather than settling at a buffer size,");
+        System.out.println("means nothing bounded the producer. (Growth is faster than linear");
+        System.out.println("because the loop is still being JIT-compiled during the first run.)");
+
+        // RSocket delivered exactly the demand, at every quiet period.
+        assertThat(rs).containsOnly(WANTED);
+        // The other two grew with the wait rather than settling at a buffer size.
+        assertThat(gr[2]).isGreaterThan(gr[0] * 2);
+        assertThat(ws[2]).isGreaterThan(ws[0] * 2);
+    }
+
+    private void row(String label, long[] v) {
+        System.out.printf("%-14s %,10d %,10d %,10d%n", label, v[0], v[1], v[2]);
+    }
+
+    /**
+     * RSocket: {@code request(n)} is a frame on the wire. The server is told the number, produces
+     * that many, and stops. Nothing blocks and nothing buffers &mdash; the producer is simply not
+     * called again.
+     */
+    private long rsocket(long quietMs) throws Exception {
+        RSocketRequester requester = RSocketRequester.builder()
+                .rsocketStrategies(strategies).tcp("localhost", 7003);
+        try {
+            AtomicLong received = new AtomicLong();
+            BaseSubscriber<Quote> subscriber = new BaseSubscriber<>() {
+                @Override
+                protected void hookOnSubscribe(Subscription subscription) {
+                    subscription.request(WANTED);   // and never again
+                }
+
+                @Override
+                protected void hookOnNext(Quote value) {
+                    received.incrementAndGet();
+                }
+            };
+            requester.route("quotes.unbounded").data("AAPL")
+                    .retrieveFlux(Quote.class).subscribe(subscriber);
+            Thread.sleep(quietMs);
+            long produced = requester.route("produced").retrieveMono(Long.class).block();
+            subscriber.dispose();
+            assertThat(received.get()).isEqualTo(WANTED);
+            return produced;
+        }
+        finally {
+            requester.dispose();
+        }
+    }
+
+    /**
+     * gRPC: there is no application-level demand signal. The blocking stub's {@code Iterator}
+     * requests one message per {@code next()}, and the handler writes with
+     * {@code StreamObserver.onNext}, which never blocks.
+     */
+    private long grpc(long quietMs) throws Exception {
+        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9093)
+                .usePlaintext().build();
+        try {
+            var stub = QuoteServiceGrpc.newBlockingStub(channel);
+            QuoteRequest req = QuoteRequest.newBuilder().setSymbol("AAPL").build();
+            Iterator<com.ankurm.protocols.grpc.Quote> it = stub.streamUnbounded(req);
+            long received = 0;
+            while (received < WANTED && it.hasNext()) {
+                it.next();
+                received++;
+            }
+            Thread.sleep(quietMs);
+            return stub.produced(req).getCount();
+        }
+        finally {
+            channel.shutdownNow();
+        }
+    }
+
+    /** Raw WebSocket: the frame protocol has no notion of demand at all. */
+    private long websocket(long quietMs) throws Exception {
+        try (WsClient client = new WsClient(httpPort, WANTED)) {
+            client.send("UNBOUNDED AAPL");
+            long received = 0;
+            while (received < WANTED && client.take(5_000) != null) {
+                received++;
+            }
+            Thread.sleep(quietMs);
+            // Ask on a second connection: the first one's server-side thread is busy writing.
+            try (WsClient asker = new WsClient(httpPort, 8)) {
+                asker.send("PRODUCED");
+                return Long.parseLong(asker.take(5_000));
+            }
+        }
+    }
+}
@@ -0,0 +1,29 @@
+package com.ankurm.protocols;
+
+import java.util.Arrays;
+
+/** Latency bookkeeping, kept in one place so the three protocols are measured identically. */
+final class Bench {
+
+    private final long[] nanos;
+    private int i;
+
+    Bench(int n) {
+        this.nanos = new long[n];
+    }
+
+    void record(long ns) {
+        nanos[i++] = ns;
+    }
+
+    /** p50, p99 and mean in microseconds, plus calls per second derived from the mean. */
+    String summary(String label) {
+        long[] sorted = Arrays.copyOf(nanos, i);
+        Arrays.sort(sorted);
+        double p50 = sorted[(int) (sorted.length * 0.50)] / 1000.0;
+        double p99 = sorted[(int) (sorted.length * 0.99)] / 1000.0;
+        double mean = Arrays.stream(sorted).average().orElse(0) / 1000.0;
+        return String.format("%-12s n=%-6d p50=%8.1f us  p99=%9.1f us  mean=%8.1f us  ~%,.0f calls/s",
+                label, sorted.length, p50, p99, mean, 1_000_000.0 / mean);
+    }
+}
@@ -0,0 +1,66 @@
+package com.ankurm.protocols;
+
+import java.util.Iterator;
+
+import com.ankurm.protocols.grpc.QuoteRequest;
+import com.ankurm.protocols.grpc.QuoteServiceGrpc;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+/**
+ * The documented gRPC answer to over-production, tested rather than assumed.
+ *
+ * <p>Every guide to gRPC flow control says the same thing: {@code StreamObserver.onNext} does not
+ * block, so a server-streaming handler must consult
+ * {@code ServerCallStreamObserver.isReady()} before writing. This test runs the identical
+ * scenario against a handler that does exactly that.
+ *
+ * <p>It did not help. The transcript is committed as measured &mdash; on a loopback connection
+ * with a client that has stopped reading, the ready flag stays set and the loop keeps producing
+ * at the same rate. I have not established whether the messages accumulate in the server's
+ * outbound queue, in the client transport, or are discarded after the deframer, and I am not
+ * going to assert a mechanism I have not read the source for. What the numbers do establish is
+ * narrower and still useful: <strong>{@code isReady()} is not a substitute for a demand
+ * signal.</strong>
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"spring.rsocket.server.port=7004", "spring.grpc.server.port=9094"})
+class GrpcIsReadyTest {
+
+    private static final int WANTED = 100;
+
+    @Test
+    void isReadyDoesNotBoundTheProducerHere() throws Exception {
+        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9094)
+                .usePlaintext().build();
+        try {
+            var stub = QuoteServiceGrpc.newBlockingStub(channel);
+            QuoteRequest req = QuoteRequest.newBuilder().setSymbol("AAPL").build();
+
+            System.out.println("=== gRPC server streaming, client takes 100 then stops ===");
+            System.out.printf("%-28s %10s %10s%n", "handler", "quiet 0.5s", "quiet 2s");
+            System.out.printf("%-28s %,10d %,10d%n", "plain onNext loop",
+                    run(stub, req, false, 500), run(stub, req, false, 2_000));
+            System.out.printf("%-28s %,10d %,10d%n", "isReady() checked before write",
+                    run(stub, req, true, 500), run(stub, req, true, 2_000));
+        }
+        finally {
+            channel.shutdownNow();
+        }
+    }
+
+    private long run(QuoteServiceGrpc.QuoteServiceBlockingStub stub, QuoteRequest req,
+                     boolean ready, long quietMs) throws Exception {
+        Iterator<com.ankurm.protocols.grpc.Quote> it =
+                ready ? stub.streamUnboundedReady(req) : stub.streamUnbounded(req);
+        long received = 0;
+        while (received < WANTED && it.hasNext()) {
+            it.next();
+            received++;
+        }
+        Thread.sleep(quietMs);
+        return stub.produced(req).getCount();
+    }
+}
@@ -0,0 +1,46 @@
+package com.ankurm.protocols;
+
+import java.util.HexFormat;
+
+import com.ankurm.protocols.grpc.Quote;
+import org.junit.jupiter.api.Test;
+import tools.jackson.databind.ObjectMapper;
+import tools.jackson.dataformat.cbor.CBORMapper;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The same five fields, encoded three ways.
+ *
+ * <p>This is the one comparison that needs no server, no warm-up and no statistics: it is a
+ * property of the wire formats. It is also the honest place to start, because on a loopback
+ * benchmark the encoding is most of what separates the three protocols.
+ */
+class PayloadSizeTest {
+
+    private static final com.ankurm.protocols.Quote POJO =
+            new com.ankurm.protocols.Quote("AAPL", 42, 100.42, 100.44, 1_772_000_000_000_000L);
+
+    @Test
+    void protobufVersusJsonVersusCbor() {
+        byte[] proto = Quote.newBuilder()
+                .setSymbol("AAPL").setSeq(42)
+                .setBid(100.42).setAsk(100.44)
+                .setEpochMicros(1_772_000_000_000_000L)
+                .build().toByteArray();
+
+        byte[] json = new ObjectMapper().writeValueAsBytes(POJO);
+        byte[] cbor = new CBORMapper().writeValueAsBytes(POJO);
+
+        System.out.println("=== one Quote, five fields, three encodings ===");
+        System.out.printf("protobuf : %3d bytes  %s%n", proto.length, HexFormat.of().formatHex(proto));
+        System.out.printf("JSON     : %3d bytes  %s%n", json.length, new String(json));
+        System.out.printf("CBOR     : %3d bytes  %s%n", cbor.length, HexFormat.of().formatHex(cbor));
+        System.out.printf("JSON is %.2fx protobuf; CBOR is %.2fx protobuf%n",
+                json.length / (double) proto.length, cbor.length / (double) proto.length);
+
+        // Protobuf is smallest because field names are numbers and there is no framing text.
+        assertThat(proto.length).isLessThan(cbor.length);
+        assertThat(cbor.length).isLessThan(json.length);
+    }
+}
@@ -0,0 +1,38 @@
+package com.ankurm.protocols;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * What Boot actually configures for RSocket, printed rather than quoted.
+ *
+ * <p>A bare {@code RSocketStrategies.create()} carries no JSON and no CBOR at all &mdash; only
+ * the string and buffer codecs. Everything interesting comes from Boot's
+ * {@code RSocketStrategiesAutoConfiguration}, which is also where the default data mime type is
+ * decided, and it is not the one most people assume.
+ */
+@SpringBootTest(properties = {"spring.rsocket.server.port=7005", "spring.grpc.server.port=9095"})
+class RSocketDefaultsTest {
+
+    @Autowired
+    RSocketStrategies strategies;
+
+    @Test
+    void bootConfiguredCodecs() {
+        var encoders = strategies.encoders().stream().map(e -> e.getClass().getSimpleName()).toList();
+        var decoders = strategies.decoders().stream().map(d -> d.getClass().getSimpleName()).toList();
+
+        System.out.println("=== Boot-configured RSocketStrategies ===");
+        System.out.println("encoders : " + encoders);
+        System.out.println("decoders : " + decoders);
+        System.out.println("bare RSocketStrategies.create() encoders : "
+                + RSocketStrategies.create().encoders().stream()
+                        .map(e -> e.getClass().getSimpleName()).toList());
+
+        assertThat(encoders).isNotEmpty();
+    }
+}
@@ -0,0 +1,111 @@
+package com.ankurm.protocols;
+
+import com.ankurm.protocols.grpc.QuoteRequest;
+import com.ankurm.protocols.grpc.QuoteServiceGrpc;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Sequential request/response round trips, all three protocols, same JVM, same data source.
+ *
+ * <p>Read the numbers as a comparison of the three, not as absolutes. Client and server are the
+ * same process on one loopback interface with two cores, so the network &mdash; the thing that
+ * dominates every real deployment &mdash; is absent. That deletes the advantage a smaller
+ * encoding would have on a real link and leaves framing and dispatch cost, which is exactly the
+ * part a loopback measures well.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"spring.rsocket.server.port=7001", "spring.grpc.server.port=9091"})
+class RequestResponseBenchmarkTest {
+
+    private static final int WARMUP = 2_000;
+    private static final int MEASURED = 5_000;
+
+    @LocalServerPort
+    int httpPort;
+
+    @Autowired
+    RSocketStrategies strategies;
+
+    @Test
+    void threeProtocolsOneOperation() throws Exception {
+        System.out.println("=== request/response, " + MEASURED + " sequential calls after "
+                + WARMUP + " warm-up, loopback, JDK " + System.getProperty("java.version") + " ===");
+        System.out.println(grpc());
+        System.out.println(rsocket());
+        System.out.println(websocket());
+    }
+
+    private String grpc() {
+        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9091)
+                .usePlaintext().build();
+        try {
+            var stub = QuoteServiceGrpc.newBlockingStub(channel);
+            QuoteRequest req = QuoteRequest.newBuilder().setSymbol("AAPL").build();
+            for (int i = 0; i < WARMUP; i++) {
+                stub.getQuote(req);
+            }
+            Bench bench = new Bench(MEASURED);
+            for (int i = 0; i < MEASURED; i++) {
+                long t0 = System.nanoTime();
+                var quote = stub.getQuote(req);
+                bench.record(System.nanoTime() - t0);
+                assertThat(quote.getSymbol()).isEqualTo("AAPL");
+            }
+            return bench.summary("gRPC");
+        }
+        finally {
+            channel.shutdownNow();
+        }
+    }
+
+    private String rsocket() {
+        RSocketRequester requester = RSocketRequester.builder()
+                .rsocketStrategies(strategies).tcp("localhost", 7001);
+        try {
+            for (int i = 0; i < WARMUP; i++) {
+                requester.route("quote").data("AAPL").retrieveMono(Quote.class).block();
+            }
+            Bench bench = new Bench(MEASURED);
+            for (int i = 0; i < MEASURED; i++) {
+                long t0 = System.nanoTime();
+                Quote quote = requester.route("quote").data("AAPL")
+                        .retrieveMono(Quote.class).block();
+                bench.record(System.nanoTime() - t0);
+                assertThat(quote).isNotNull();
+            }
+            return bench.summary("RSocket");
+        }
+        finally {
+            requester.dispose();
+        }
+    }
+
+    private String websocket() throws Exception {
+        try (WsClient client = new WsClient(httpPort, 64)) {
+            for (int i = 0; i < WARMUP; i++) {
+                client.send("QUOTE AAPL");
+                assertThat(client.take(5_000)).isNotNull();
+            }
+            Bench bench = new Bench(MEASURED);
+            for (int i = 0; i < MEASURED; i++) {
+                long t0 = System.nanoTime();
+                client.send("QUOTE AAPL");
+                String reply = client.take(5_000);
+                bench.record(System.nanoTime() - t0);
+                assertThat(reply).contains("AAPL");
+            }
+            return bench.summary("WebSocket");
+        }
+    }
+}
@@ -0,0 +1,120 @@
+package com.ankurm.protocols;
+
+import java.util.Iterator;
+
+import com.ankurm.protocols.grpc.StreamRequest;
+import com.ankurm.protocols.grpc.QuoteServiceGrpc;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Server streaming: how fast can one connection carry N messages one way?
+ *
+ * <p>This is the shape most of these protocols are actually chosen for &mdash; a price feed, a
+ * log tail, a progress stream &mdash; and it separates them differently from request/response,
+ * because the per-message cost stops being dominated by a round trip.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"spring.rsocket.server.port=7002", "spring.grpc.server.port=9092"})
+class StreamThroughputBenchmarkTest {
+
+    private static final int WARMUP = 5_000;
+    private static final int N = 50_000;
+
+    @LocalServerPort
+    int httpPort;
+
+    @Autowired
+    RSocketStrategies strategies;
+
+    @Test
+    void fiftyThousandMessagesEachWay() throws Exception {
+        System.out.println("=== server streaming, " + N + " messages on one connection, loopback ===");
+        System.out.println(grpc());
+        System.out.println(rsocket());
+        System.out.println(websocket());
+    }
+
+    private String grpc() {
+        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9092)
+                .usePlaintext().build();
+        try {
+            var stub = QuoteServiceGrpc.newBlockingStub(channel);
+            drain(stub, WARMUP);
+            long t0 = System.nanoTime();
+            int seen = drain(stub, N);
+            return report("gRPC", seen, System.nanoTime() - t0);
+        }
+        finally {
+            channel.shutdownNow();
+        }
+    }
+
+    private int drain(QuoteServiceGrpc.QuoteServiceBlockingStub stub, int count) {
+        // The blocking stub returns an Iterator. Errors surface mid-iteration rather than at the
+        // call, which is the first thing that surprises people about gRPC server streaming.
+        Iterator<com.ankurm.protocols.grpc.Quote> it = stub.streamQuotes(
+                StreamRequest.newBuilder().setSymbol("AAPL").setCount(count).build());
+        int seen = 0;
+        while (it.hasNext()) {
+            it.next();
+            seen++;
+        }
+        return seen;
+    }
+
+    private String rsocket() {
+        RSocketRequester requester = RSocketRequester.builder()
+                .rsocketStrategies(strategies).tcp("localhost", 7002);
+        try {
+            var warm = new RSocketQuoteController.StreamSpec("AAPL", WARMUP, 0);
+            requester.route("quotes").data(warm).retrieveFlux(Quote.class).blockLast();
+
+            var spec = new RSocketQuoteController.StreamSpec("AAPL", N, 0);
+            long t0 = System.nanoTime();
+            long seen = requester.route("quotes").data(spec)
+                    .retrieveFlux(Quote.class).count().block();
+            return report("RSocket", (int) seen, System.nanoTime() - t0);
+        }
+        finally {
+            requester.dispose();
+        }
+    }
+
+    private String websocket() throws Exception {
+        // A generous inbox, because there is nothing else to do with messages that arrive faster
+        // than they are read. This IS the WebSocket back-pressure story, in one constant.
+        try (WsClient client = new WsClient(httpPort, N + WARMUP + 16)) {
+            client.send("STREAM AAPL " + WARMUP + " 0");
+            int warm = 0;
+            String m;
+            while ((m = client.take(20_000)) != null && !"END".equals(m)) {
+                warm++;
+            }
+            assertThat(warm).isEqualTo(WARMUP);
+
+            long t0 = System.nanoTime();
+            client.send("STREAM AAPL " + N + " 0");
+            int seen = 0;
+            while ((m = client.take(20_000)) != null && !"END".equals(m)) {
+                seen++;
+            }
+            return report("WebSocket", seen, System.nanoTime() - t0);
+        }
+    }
+
+    private String report(String label, int seen, long elapsedNanos) {
+        assertThat(seen).isEqualTo(N);
+        double seconds = elapsedNanos / 1e9;
+        return String.format("%-12s %,7d msgs in %6.3f s  =  %,10.0f msgs/s  (%6.2f us each)",
+                label, seen, seconds, seen / seconds, elapsedNanos / 1000.0 / seen);
+    }
+}
@@ -0,0 +1,60 @@
+package com.ankurm.protocols;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import jakarta.websocket.ContainerProvider;
+import jakarta.websocket.WebSocketContainer;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.client.standard.StandardWebSocketClient;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+
+/**
+ * A minimal raw-WebSocket client for the benchmark.
+ *
+ * <p>The queue is the point of interest. A WebSocket client has no way to tell the server how
+ * many messages it is ready for, so the only thing it can do with an over-eager producer is
+ * buffer them &mdash; here, in a bounded queue that starts dropping. That is the shape of every
+ * real WebSocket consumer, and it is what {@code BackPressureTest} contrasts with RSocket.
+ */
+final class WsClient implements AutoCloseable {
+
+    private final WebSocketSession session;
+    private final BlockingQueue<String> inbox;
+
+    WsClient(int port, int inboxCapacity) throws Exception {
+        this.inbox = new ArrayBlockingQueue<>(inboxCapacity);
+        WebSocketContainer container = ContainerProvider.getWebSocketContainer();
+        container.setDefaultMaxTextMessageBufferSize(1024 * 1024);
+        this.session = new StandardWebSocketClient(container)
+                .execute(new TextWebSocketHandler() {
+                    @Override
+                    protected void handleTextMessage(WebSocketSession s, TextMessage message) {
+                        inbox.offer(message.getPayload());   // drops when full; nothing else to do
+                    }
+                }, "ws://localhost:" + port + "/quotes").get();
+    }
+
+    void send(String command) throws Exception {
+        session.sendMessage(new TextMessage(command));
+    }
+
+    String take(long timeoutMs) throws Exception {
+        return inbox.poll(timeoutMs, TimeUnit.MILLISECONDS);
+    }
+
+    int drained() {
+        return inbox.size();
+    }
+
+    boolean isOpen() {
+        return session.isOpen();
+    }
+
+    @Override
+    public void close() throws Exception {
+        session.close();
+    }
+}
@@ -0,0 +1,61 @@
+# saga
+
+Companion code for *Saga Pattern in Spring Boot: Orchestration vs Choreography with Kafka* on
+[ankurm.com](https://ankurm.com).
+
+The same order/payment/inventory saga, implemented twice against a real (embedded) Kafka
+broker: once as **choreography** (every participant reacts to the previous step's event and
+decides for itself what happens next) and once as **orchestration** (one class issues commands
+and makes every decision; participants only ever answer a command). Both versions are driven
+through the same failure -- insufficient inventory -- so the compensating transaction that
+undoes the reserved payment can be compared side by side. A fourth test demonstrates the
+idempotent-consumer problem concretely: the same command, redelivered, reserving stock only
+once.
+
+## Versions
+
+| Artifact | Version |
+|---|---|
+| Spring Boot | 4.1.1 (GA; verified against Maven Central's `maven-metadata.xml` -- `4.2.0-M2` is the newest entry there but is a milestone, not GA) |
+| Spring Kafka | 4.1.1, managed by the Spring Boot BOM |
+| JDK | 25 (LTS) |
+| H2 | runtime, in-memory, for the demo only |
+
+## Quickstart
+
+```
+mvn -pl saga test -Dtest=ChoreographyHappyPathTest
+mvn -pl saga test -Dtest=ChoreographyCompensationTest
+mvn -pl saga test -Dtest=OrchestrationSagaTest
+mvn -pl saga test -Dtest=IdempotentConsumerTest
+```
+
+Each test class starts its own embedded Kafka broker and its own isolated in-memory H2
+database (a random schema name per test, via `@DynamicPropertySource`), so running them
+individually or together (`mvn -pl saga test`) gives the same results.
+
+## What each class is
+
+| Class | Role |
+|---|---|
+| `choreography.ChoreographySagaStarter` | Creates the order, publishes `OrderCreated`, and does nothing else -- the only choreography class that "starts" anything |
+| `choreography.PaymentChoreographyListener` | Reserves payment on `OrderCreated`; refunds it on `InventoryRejected` (the compensating transaction) |
+| `choreography.InventoryChoreographyListener` | The saga's one failure trigger: reacts to `PaymentReserved`, decides stock is or isn't available, publishes accordingly |
+| `choreography.OrderChoreographyListener` | Confirms or cancels the order; never decides anything, only reacts |
+| `orchestration.OrderSagaOrchestrator` | Every saga decision lives here: what to do after each reply, including issuing the compensating `RefundPaymentCommand` |
+| `orchestration.PaymentOrchestrationHandler` / `InventoryOrchestrationHandler` | Carry out exactly one command each and reply; never decide what happens next |
+| `idempotency.IdempotencyGuard` | `claim(commandId)` -- true the first time a command id is seen, false on every redelivery, backed by a real UNIQUE constraint |
+| `inventory.InventoryService` | In-memory stock only -- this module is about saga coordination, not inventory persistence |
+
+## Captured output (`output/`)
+
+| File | What it captures |
+|---|---|
+| `00-choreography-happy-path.txt` | Enough stock: payment reserved, inventory reserved, order confirmed, with no single class aware of the whole sequence |
+| `01-choreography-compensation.txt` | Not enough stock: inventory rejects, payment refunds itself in response, order ends cancelled |
+| `02-orchestration-saga.txt` | Both scenarios again, driven by `OrderSagaOrchestrator` instead -- same outcomes, one class making every decision |
+| `03-idempotent-consumer-duplicate.txt` | The same `ReserveInventoryCommand`, same commandId, delivered twice: one processed-command row, one stock reservation, one reply |
+
+No `docs/` chapter directory in this repository -- the intermediate and reference-depth
+material that would normally live there is in accordion sections inside the WordPress post
+itself.
@@ -0,0 +1,17 @@
+$ mvn -pl saga test -Dtest=ChoreographyHappyPathTest
+(real embedded Kafka broker via @EmbeddedKafka; ChoreographySagaStarter.placeOrder() publishes
+ OrderCreated, and every step after that is a listener reacting to the previous one's event --
+ no single class decides the whole saga)
+
+Order 1 status: CONFIRMED
+Stock remaining for GADGET-1: 8
+Payment record status: RESERVED
+
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
+
+Stock started at 10, the order asked for 2, and 8 is what's left -- the reservation really ran.
+Five separate listeners fired in sequence to get here: PaymentChoreographyListener reacted to
+OrderCreated and reserved payment; InventoryChoreographyListener reacted to PaymentReserved,
+found enough stock, and published InventoryReserved; OrderChoreographyListener reacted to that
+and confirmed the order. None of those three classes knows about the other two's existence --
+each one only knows "when I see event X, I do Y, then publish Z".
@@ -0,0 +1,15 @@
+$ mvn -pl saga test -Dtest=ChoreographyCompensationTest
+(same saga, stock deliberately too low: 1 unit on hand, order asks for 5)
+
+Order 1 status: CANCELLED
+Stock remaining for GADGET-1 (untouched by the rejected reservation): 1
+Payment record status: REFUNDED
+
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
+
+InventoryChoreographyListener's tryReserve() refused before touching the stock map at all --
+the 1 unit is exactly where it started. That rejection published InventoryRejected, which
+PaymentChoreographyListener is also subscribed to; it refunded the payment it had reserved
+earlier and published PaymentRefunded, which OrderChoreographyListener used to cancel the
+order. The compensating transaction (the refund) is ordinary code in a class that already
+existed for an unrelated reason -- nothing new coordinates it.
@@ -0,0 +1,20 @@
+$ mvn -pl saga test -Dtest=OrchestrationSagaTest
+(the same two scenarios -- enough stock, and not enough stock -- run through
+ OrderSagaOrchestrator instead of through independent listeners)
+
+[happy path] Order 1 status: CONFIRMED
+[happy path] Stock remaining for GADGET-ORCH-OK: 8
+[happy path] Payment record status: RESERVED
+[compensation] Order 2 status: CANCELLED
+[compensation] Stock remaining for GADGET-ORCH-FAIL (untouched): 1
+[compensation] Payment record status: REFUNDED
+
+[INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0
+
+Same outcomes as the choreography transcripts, reached a structurally different way. Here,
+PaymentOrchestrationHandler and InventoryOrchestrationHandler never decide what happens next --
+they carry out exactly one command each and reply. OrderSagaOrchestrator is the only class that
+reads a reply and decides what command to send next, including the decision, in the second
+test, to send RefundPaymentCommand once InventoryReply says no. Compare this transcript's two
+tests with the two choreography transcripts above: identical business outcomes, from listener
+classes that know nothing about the saga as a whole versus one class that knows all of it.
@@ -0,0 +1,16 @@
+$ mvn -pl saga test -Dtest=IdempotentConsumerTest
+(the exact same ReserveInventoryCommand -- same commandId -- published twice by hand, to
+ simulate a redelivery; stock starts at 10, the command asks for 3)
+
+ProcessedCommand rows for commandId f07ec35c-853c-4405-a024-b62d304dfabb: 1
+Stock remaining for GADGET-DUP after two deliveries of the same command: 7
+InventoryReply messages actually published: 1
+  reply: InventoryReply[commandId=f07ec35c-853c-4405-a024-b62d304dfabb, orderId=999, success=true, reason=null]
+
+[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
+
+Two deliveries, one row in processed_commands, one actual reservation: 10 - 3 = 7, not 4. The
+second delivery hit IdempotencyGuard.claim(), got false back because the UNIQUE constraint on
+commandId rejected the second insert, and returned from the listener before calling
+InventoryService.tryReserve() at all -- it never touched stock, and it never published a second
+InventoryReply. Exactly one reply reached the topic for two deliveries of the same command.
@@ -0,0 +1,74 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0">
+  <modelVersion>4.0.0</modelVersion>
+
+  <!-- Same convention as the rest of this repo: inherit spring-boot-starter-parent so Kafka,
+       JPA and the test stack are all Boot-managed. Boot 4.1.1 manages Spring Kafka 4.1.1. -->
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>4.1.1</version>
+    <relativePath/>
+  </parent>
+
+  <groupId>com.ankurm</groupId>
+  <artifactId>saga</artifactId>
+  <version>1.0</version>
+  <packaging>jar</packaging>
+
+  <properties>
+    <java.version>25</java.version>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter</artifactId>
+    </dependency>
+    <!-- spring-boot-starter-kafka, not a bare spring-kafka dependency: see
+         kafka-error-handling/pom.xml in this same repo. Boot 4's auto-configuration lives in
+         spring-boot-kafka, which the starter brings and spring-kafka alone does not. -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-kafka</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-jackson</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-data-jpa</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.h2database</groupId>
+      <artifactId>h2</artifactId>
+      <scope>runtime</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>
@@ -0,0 +1,12 @@
+package com.ankurm.sagademo;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class SagaDemoApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(SagaDemoApplication.class, args);
+    }
+}
@@ -0,0 +1,34 @@
+package com.ankurm.sagademo;
+
+/**
+ * Every topic this module uses, in one place, so the two coordination styles in the post --
+ * choreography and orchestration -- never accidentally share a topic. The "choreo." group
+ * carries plain domain events; the "orch." group splits into commands (the orchestrator telling
+ * a participant what to do) and replies (the participant reporting back), which is the concrete
+ * difference between the two styles: choreography has only events, orchestration has commands
+ * and replies to them.
+ */
+public final class Topics {
+
+    private Topics() {
+    }
+
+    // Choreography: each participant listens for the previous step's event and decides for
+    // itself what to publish next. No single place knows the whole saga.
+    public static final String CHOREO_ORDER_CREATED = "choreo.order.created";
+    public static final String CHOREO_PAYMENT_RESERVED = "choreo.payment.reserved";
+    public static final String CHOREO_INVENTORY_RESERVED = "choreo.inventory.reserved";
+    public static final String CHOREO_INVENTORY_REJECTED = "choreo.inventory.rejected";
+    public static final String CHOREO_PAYMENT_REFUNDED = "choreo.payment.refunded";
+    public static final String CHOREO_ORDER_CONFIRMED = "choreo.order.confirmed";
+    public static final String CHOREO_ORDER_CANCELLED = "choreo.order.cancelled";
+
+    // Orchestration: the orchestrator issues commands and participants reply; only the
+    // orchestrator decides what happens next.
+    public static final String ORCH_CMD_RESERVE_PAYMENT = "orch.cmd.reserve-payment";
+    public static final String ORCH_REPLY_PAYMENT = "orch.reply.payment";
+    public static final String ORCH_CMD_RESERVE_INVENTORY = "orch.cmd.reserve-inventory";
+    public static final String ORCH_REPLY_INVENTORY = "orch.reply.inventory";
+    public static final String ORCH_CMD_REFUND_PAYMENT = "orch.cmd.refund-payment";
+    public static final String ORCH_REPLY_REFUND = "orch.reply.refund";
+}
@@ -0,0 +1,30 @@
+package com.ankurm.sagademo.choreography;
+
+import com.ankurm.sagademo.Topics;
+import com.ankurm.sagademo.events.OrderCreated;
+import com.ankurm.sagademo.orders.Order;
+import com.ankurm.sagademo.orders.OrderService;
+
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/** The only thing that starts a choreography saga: create the order, publish one event, and
+ *  walk away. Everything after this call is other services reacting to events, not this class
+ *  coordinating anything. */
+@Component
+public class ChoreographySagaStarter {
+
+    private final OrderService orders;
+    private final KafkaTemplate<String, Object> kafka;
+
+    public ChoreographySagaStarter(OrderService orders, KafkaTemplate<String, Object> kafka) {
+        this.orders = orders;
+        this.kafka = kafka;
+    }
+
+    public Order placeOrder(String sku, int qty, int amountCents) {
+        Order order = orders.createPendingOrder(sku, qty, amountCents, "CHOREOGRAPHY");
+        kafka.send(Topics.CHOREO_ORDER_CREATED, new OrderCreated(order.getId(), sku, qty, amountCents));
+        return order;
+    }
+}
@@ -0,0 +1,40 @@
+package com.ankurm.sagademo.choreography;
+
+import com.ankurm.sagademo.Topics;
+import com.ankurm.sagademo.events.InventoryRejected;
+import com.ankurm.sagademo.events.InventoryReserved;
+import com.ankurm.sagademo.events.PaymentReserved;
+import com.ankurm.sagademo.inventory.InventoryService;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * This is the only place in the choreography saga that knows what "not enough stock" means --
+ * it reacts to a payment event, makes a decision with no other service's input, and publishes
+ * one of two outcomes. Whichever one it publishes, it has no idea what will happen next; it
+ * just knows it told the truth about its own stock.
+ */
+@Component
+public class InventoryChoreographyListener {
+
+    private final InventoryService inventory;
+    private final KafkaTemplate<String, Object> kafka;
+
+    public InventoryChoreographyListener(InventoryService inventory, KafkaTemplate<String, Object> kafka) {
+        this.inventory = inventory;
+        this.kafka = kafka;
+    }
+
+    @KafkaListener(topics = Topics.CHOREO_PAYMENT_RESERVED, groupId = "choreo-inventory")
+    public void onPaymentReserved(PaymentReserved event) {
+        boolean reserved = inventory.tryReserve(event.sku(), event.qty());
+        if (reserved) {
+            kafka.send(Topics.CHOREO_INVENTORY_RESERVED, new InventoryReserved(event.orderId()));
+        } else {
+            kafka.send(Topics.CHOREO_INVENTORY_REJECTED,
+                    new InventoryRejected(event.orderId(), "insufficient stock for " + event.sku()));
+        }
+    }
+}
@@ -0,0 +1,34 @@
+package com.ankurm.sagademo.choreography;
+
+import com.ankurm.sagademo.Topics;
+import com.ankurm.sagademo.events.InventoryReserved;
+import com.ankurm.sagademo.events.PaymentRefunded;
+import com.ankurm.sagademo.orders.OrderService;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * The order itself has no say in any of this -- it is confirmed or cancelled by whichever event
+ * happens to arrive. That passivity is the point: in choreography, the order record is just
+ * data other services' events update, not a thing that drives the saga.
+ */
+@Component
+public class OrderChoreographyListener {
+
+    private final OrderService orders;
+
+    public OrderChoreographyListener(OrderService orders) {
+        this.orders = orders;
+    }
+
+    @KafkaListener(topics = Topics.CHOREO_INVENTORY_RESERVED, groupId = "choreo-order-confirm")
+    public void onInventoryReserved(InventoryReserved event) {
+        orders.confirm(event.orderId());
+    }
+
+    @KafkaListener(topics = Topics.CHOREO_PAYMENT_REFUNDED, groupId = "choreo-order-cancel")
+    public void onPaymentRefunded(PaymentRefunded event) {
+        orders.cancel(event.orderId());
+    }
+}
@@ -0,0 +1,47 @@
+package com.ankurm.sagademo.choreography;
+
+import com.ankurm.sagademo.Topics;
+import com.ankurm.sagademo.events.InventoryRejected;
+import com.ankurm.sagademo.events.OrderCreated;
+import com.ankurm.sagademo.events.PaymentRefunded;
+import com.ankurm.sagademo.events.PaymentReserved;
+import com.ankurm.sagademo.payments.PaymentService;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * Payment's two jobs in the choreography saga, and nothing else. This class never hears about
+ * the order being confirmed or cancelled -- it reacts to its own two events and moves on. Who
+ * decides what happens to the order as a whole is not answered anywhere in this class, which is
+ * exactly the property choreography trades for having no single orchestrator to maintain.
+ */
+@Component
+public class PaymentChoreographyListener {
+
+    private final PaymentService payments;
+    private final KafkaTemplate<String, Object> kafka;
+
+    public PaymentChoreographyListener(PaymentService payments, KafkaTemplate<String, Object> kafka) {
+        this.payments = payments;
+        this.kafka = kafka;
+    }
+
+    @KafkaListener(topics = Topics.CHOREO_ORDER_CREATED, groupId = "choreo-payment")
+    public void onOrderCreated(OrderCreated event) {
+        payments.reserve(event.orderId(), event.amountCents());
+        kafka.send(Topics.CHOREO_PAYMENT_RESERVED,
+                new PaymentReserved(event.orderId(), event.sku(), event.qty()));
+    }
+
+    /** The compensating transaction. Nothing calls this directly -- it runs because this
+     *  listener happens to also be subscribed to the rejection event, which is choreography's
+     *  defining trait: the participant that caused the original effect is the one that notices
+     *  the failure and undoes it, not a coordinator telling it to. */
+    @KafkaListener(topics = Topics.CHOREO_INVENTORY_REJECTED, groupId = "choreo-payment-compensation")
+    public void onInventoryRejected(InventoryRejected event) {
+        payments.refund(event.orderId());
+        kafka.send(Topics.CHOREO_PAYMENT_REFUNDED, new PaymentRefunded(event.orderId()));
+    }
+}
@@ -0,0 +1,6 @@
+package com.ankurm.sagademo.events;
+
+/** The saga's one failure trigger in this demo, deliberately placed in inventory so both
+ *  coordination styles can be compared against exactly the same failure. */
+public record InventoryRejected(Long orderId, String reason) {
+}
@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record InventoryReply(String commandId, Long orderId, boolean success, String reason) {
+}
@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record InventoryReserved(Long orderId) {
+}
@@ -0,0 +1,7 @@
+package com.ankurm.sagademo.events;
+
+/** Choreography's starting event. Nothing downstream of this was told what to do -- each
+ *  listener decides for itself, from its own domain knowledge, what reacting to this event
+ *  means. */
+public record OrderCreated(Long orderId, String sku, int qty, int amountCents) {
+}
@@ -0,0 +1,6 @@
+package com.ankurm.sagademo.events;
+
+/** The compensating transaction's own completion event -- published after undoing the
+ *  payment reservation that an earlier, now-abandoned step in the saga made. */
+public record PaymentRefunded(Long orderId) {
+}
@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record PaymentReply(String commandId, Long orderId, boolean success) {
+}
@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record PaymentReserved(Long orderId, String sku, int qty) {
+}
@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record RefundPaymentCommand(String commandId, Long orderId) {
+}
@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record RefundReply(String commandId, Long orderId) {
+}
@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record ReserveInventoryCommand(String commandId, Long orderId, String sku, int qty) {
+}
@@ -0,0 +1,8 @@
+package com.ankurm.sagademo.events;
+
+/** commandId is the idempotency key -- see IdempotentCommand in the idempotency package. The
+ *  orchestrator sets it once per logical step and reuses the exact same value if it ever
+ *  resends the command, which is what lets a participant tell "the same command, redelivered"
+ *  apart from "a brand new command that happens to target the same order". */
+public record ReservePaymentCommand(String commandId, Long orderId, int amountCents) {
+}
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Claude	78fc41e586	kafka-streams: capture the ambiguous null tombstone compile error Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01RjfyTgmERTWLqzafu34Cth	2026-10-14 07:12:07 +00:00
asmhatreandClaude Sonnet 5.5	59fcdbe079	Add kafka-streams module: fraud detection topology tested with TopologyTestDriver Similarity scoring, a 5-minute windowed count that alerts once, and a KTable blocklist join, with the Spring wiring on Spring Boot 4.1.1 and Kafka Streams 4.2.1. Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01RjfyTgmERTWLqzafu34Cth	2026-10-14 06:44:56 +00:00
Claude	5db56e70a0	Add share-groups module: Kafka 4.3.1 share groups with Spring for Apache Kafka 4.1.1 Six consumers on two partitions under a consumer group, a share group with the default share.acquire.mode and a share group with record_limit; the three acknowledgement types and a record that is released until Kafka stops redelivering it; a new share group starting at latest; the ordering a share group gives up; and Spring Kafka's share containers in implicit and manual acknowledgement modes. Tests run against a real broker (scripts/broker-up.sh) and write output/. Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Lm655hdgDo5xsSLNomgdsy	2026-10-13 02:43:36 +00:00
Claude	4a5d3318bf	Add the saga module: Saga Pattern in Spring Boot, orchestration vs choreography with Kafka	2026-10-03 19:56:40 +00:00
asmhatreandClaude Sonnet 5	c1b40b0db9	Add kafka4-migration module: Spring Kafka 3.3 to 4.1 migration on Boot 3.5 to 4.1 Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Gth99spB1CzQmBFofXmU4Z	2026-09-28 09:19:55 +00:00
asmhatre	d56c60824e	Add the protocol-comparison module	2026-09-04 01:03:48 +05:30