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

README.md

@@ -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 — choreography and orchestration — 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

kafka-streams/README.md

@@ -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.

kafka-streams/output/01-similarity-scores.txt

@@ -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

kafka-streams/output/02-similarity-alerts.txt

@@ -0,0 +1,2 @@
+alerts <- c1 SIMILAR:1.0000:c1,4700.0,3,crypto
+alerts emitted: 1

kafka-streams/output/03-velocity-window.txt

@@ -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

kafka-streams/output/04-blocklist-table-join.txt

@@ -0,0 +1,3 @@
+before blocking, alerts: 0
+alerts <- c5 BLOCKED:reported stolen
+after the tombstone, alerts: 0

kafka-streams/output/05-topology-description.txt

@@ -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

kafka-streams/output/06-spring-context.txt

@@ -0,0 +1,3 @@
+StreamsBuilderFactoryBean present: true
+auto-startup: false
+running: false

kafka-streams/output/07-ambiguous-null-tombstone.txt

@@ -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

kafka-streams/pom.xml

@@ -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>

kafka-streams/scripts/compile-broken.sh

@@ -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

kafka-streams/scripts/run-all.sh

@@ -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

kafka-streams/src/broken/AmbiguousNullTombstone.java

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

kafka-streams/src/main/java/com/ankurm/streams/Embedder.java

@@ -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;
+    }
+}

kafka-streams/src/main/java/com/ankurm/streams/FraudStreamsConfig.java

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

kafka-streams/src/main/java/com/ankurm/streams/FraudTopology.java

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

kafka-streams/src/main/java/com/ankurm/streams/StreamsApplication.java

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

kafka-streams/src/main/java/com/ankurm/streams/Txn.java

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

kafka-streams/src/main/resources/application.properties

@@ -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

kafka-streams/src/test/java/com/ankurm/streams/ContextLoadsTest.java

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

kafka-streams/src/test/java/com/ankurm/streams/FraudTopologyTest.java

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

kafka-streams/src/test/java/com/ankurm/streams/Transcript.java

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

kafka4-migration/.gitignore

@@ -0,0 +1 @@
+target/

kafka4-migration/README.md

@@ -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.

kafka4-migration/legacy/pom.xml

@@ -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>

kafka4-migration/legacy/src/main/java/com/ankurm/kafka/App.java

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

kafka4-migration/legacy/src/main/java/com/ankurm/kafka/OrderEvent.java

@@ -0,0 +1,3 @@
+package com.ankurm.kafka;
+
+public record OrderEvent(String id, int qty) { }

kafka4-migration/legacy/src/main/java/com/ankurm/kafka/OrderListener.java

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

kafka4-migration/legacy/src/main/java/com/ankurm/kafka/SerdeProbe.java

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

kafka4-migration/legacy/src/main/resources/application.properties

@@ -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

kafka4-migration/legacy/src/test/java/com/ankurm/kafka/OrderFlowTest.java

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

kafka4-migration/migrated/pom.xml

@@ -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>

kafka4-migration/migrated/src/main/java/com/ankurm/kafka/App.java

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

kafka4-migration/migrated/src/main/java/com/ankurm/kafka/OrderEvent.java

@@ -0,0 +1,3 @@
+package com.ankurm.kafka;
+
+public record OrderEvent(String id, int qty) { }

kafka4-migration/migrated/src/main/java/com/ankurm/kafka/OrderListener.java

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

kafka4-migration/migrated/src/main/java/com/ankurm/kafka/SerdeProbe.java

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

kafka4-migration/migrated/src/main/resources/application.properties

@@ -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

kafka4-migration/migrated/src/test/java/com/ankurm/kafka/OrderFlowTest.java

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

kafka4-migration/output/01-legacy-kafka3.txt

@@ -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}

kafka4-migration/output/02-boot4-no-starter.txt

@@ -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

kafka4-migration/output/03-starter-added-tests-pass.txt

@@ -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

kafka4-migration/output/04-runtime-classpath-has-no-jackson2.txt

@@ -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

kafka4-migration/output/05-migrated.txt

@@ -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}

kafka4-migration/output/06-serializer-classes.txt

@@ -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 

kafka4-migration/scripts/run-all.sh

@@ -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

protocol-comparison/README.md

@@ -28,7 +28,7 @@ Read from `repo1.maven.org/.../maven-metadata.xml` and Boot's own `spring-boot-d
 ## Quickstart
 
 ```bash
-./scripts/run-all.sh     # five tests, then re-measure docs/output/
+./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
 ```
@@ -63,7 +63,8 @@ and architecture, and all three servers start inside the test JVM.
 | [`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 |
-| [`tests.txt`](docs/output/tests.txt) | 5 tests |
+| [`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
 
@@ -85,5 +86,6 @@ protocols rather than of the link.
    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.** It is binary JSON, not a schema format: 67 bytes against
-   protobuf's 35 and JSON's 83, and the difference is entirely the punctuation.
+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.

protocol-comparison/docs/02-benchmarks.md

@@ -18,7 +18,12 @@ say the opposite of the truth.
 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`.
+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
 

protocol-comparison/docs/03-payload.md

@@ -35,7 +35,19 @@ 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.
+  `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.
 

protocol-comparison/docs/04-backpressure.md

@@ -20,12 +20,13 @@ wait grows means something bounded the producer. A number that keeps climbing me
 ```
 protocol       quiet 0.5s   quiet 1s   quiet 2s
 RSocket               100        100        100
-gRPC              142,000    680,542  1,346,233
-WebSocket          23,693    125,099    311,617
+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.
+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.)
@@ -46,7 +47,7 @@ batch that the others are not doing at all.
 
 **Raw WebSocket has no mechanism.** The frame protocol has no notion of demand. The server writes
 until something underneath refuses — a socket buffer, Spring's `sendBufferSizeLimit`, the
-heap. Every one of those 311 617 messages was serialised and written for a client that had
+heap. Every one of those 320 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.
@@ -58,12 +59,14 @@ citing it ([`grpc-isready.txt`](output/grpc-isready.txt)):
 
 ```
 handler                      quiet 0.5s   quiet 2s
-plain onNext loop               353,996  1,883,443
-isReady() checked before write  1,285,787     63,859
+plain onNext loop               258,145  1,421,778
+isReady() checked before write    667,259  1,044,018
 ```
 
-**It did not bound the producer, and it was not even stable** — the same handler produced
-1.3 million in half a second on one row and 64 thousand in two seconds on the next. I
+**It did not bound the producer.** The `isReady()` row is not lower than the plain one in any
+useful sense — it produced 667 thousand messages in half a second for a client that
+wanted a hundred — 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.

protocol-comparison/docs/output/backpressure.txt

@@ -1,10 +1,10 @@
 === unbounded stream; client takes 100, then goes quiet ===
 protocol       quiet 0.5s   quiet 1s   quiet 2s
 RSocket               100        100        100
-gRPC              142,000    680,542  1,346,233
-WebSocket          23,693    125,099    311,617
+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 9.5x between 0.5 s and 2 s; WebSocket grows 13.2x.
+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.)

protocol-comparison/docs/output/grpc-isready.txt

@@ -1,4 +1,4 @@
 === gRPC server streaming, client takes 100 then stops ===
 handler                      quiet 0.5s   quiet 2s
-plain onNext loop               353,996  1,883,443
-isReady() checked before write  1,285,787     63,859
+plain onNext loop               258,145  1,421,778
+isReady() checked before write    667,259  1,044,018

protocol-comparison/docs/output/request-response.txt

@@ -1,4 +1,4 @@
 === request/response, 5000 sequential calls after 2000 warm-up, loopback, JDK 25.0.4.1 ===
-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         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

protocol-comparison/docs/output/rsocket-codecs.txt

@@ -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]

protocol-comparison/docs/output/stream-throughput.txt

@@ -1,4 +1,4 @@
 === server streaming, 50000 messages on one connection, loopback ===
-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)
+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)

protocol-comparison/docs/output/tests.txt

@@ -1,5 +1,6 @@
-Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.301 s -- in com.ankurm.protocols.PayloadSizeTest
-Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 11.84 s -- in com.ankurm.protocols.RequestResponseBenchmarkTest
-Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 11.31 s -- in com.ankurm.protocols.BackPressureTest
-Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 26.70 s -- in com.ankurm.protocols.GrpcIsReadyTest
-Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 3.055 s -- in com.ankurm.protocols.StreamThroughputBenchmarkTest
+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

protocol-comparison/scripts/run-all.sh

@@ -8,14 +8,22 @@ 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*$|Initializing|Completed initialization'; }
+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'          "$LOG" | clean > docs/output/request-response.txt
-sed -n '/=== server streaming, 50000/,/^WebSocket  /p'    "$LOG" | clean > docs/output/stream-throughput.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"
@@ -24,6 +32,9 @@ sed -n '/=== server streaming, 50000/,/^WebSocket  /p'    "$LOG" | clean > docs/
   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/

protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java

@@ -59,12 +59,19 @@ public class WebSocketQuoteHandler extends TextWebSocketHandler {
             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.
+                // 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"));
         }

protocol-comparison/src/test/java/com/ankurm/protocols/RSocketDefaultsTest.java

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

saga/README.md

@@ -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.

saga/output/00-choreography-happy-path.txt

@@ -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".

saga/output/01-choreography-compensation.txt

@@ -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.

saga/output/02-orchestration-saga.txt

@@ -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.

saga/output/03-idempotent-consumer-duplicate.txt

@@ -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.

saga/pom.xml

@@ -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>

saga/src/main/java/com/ankurm/sagademo/SagaDemoApplication.java

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

saga/src/main/java/com/ankurm/sagademo/Topics.java

@@ -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";
+}

saga/src/main/java/com/ankurm/sagademo/choreography/ChoreographySagaStarter.java

@@ -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;
+    }
+}

saga/src/main/java/com/ankurm/sagademo/choreography/InventoryChoreographyListener.java

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

saga/src/main/java/com/ankurm/sagademo/choreography/OrderChoreographyListener.java

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

saga/src/main/java/com/ankurm/sagademo/choreography/PaymentChoreographyListener.java

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

saga/src/main/java/com/ankurm/sagademo/events/InventoryRejected.java

@@ -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) {
+}

saga/src/main/java/com/ankurm/sagademo/events/InventoryReply.java

@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record InventoryReply(String commandId, Long orderId, boolean success, String reason) {
+}

saga/src/main/java/com/ankurm/sagademo/events/InventoryReserved.java

@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record InventoryReserved(Long orderId) {
+}

saga/src/main/java/com/ankurm/sagademo/events/OrderCreated.java

@@ -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) {
+}

saga/src/main/java/com/ankurm/sagademo/events/PaymentRefunded.java

@@ -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) {
+}

saga/src/main/java/com/ankurm/sagademo/events/PaymentReply.java

@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record PaymentReply(String commandId, Long orderId, boolean success) {
+}

saga/src/main/java/com/ankurm/sagademo/events/PaymentReserved.java

@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record PaymentReserved(Long orderId, String sku, int qty) {
+}

saga/src/main/java/com/ankurm/sagademo/events/RefundPaymentCommand.java

@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record RefundPaymentCommand(String commandId, Long orderId) {
+}

saga/src/main/java/com/ankurm/sagademo/events/RefundReply.java

@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record RefundReply(String commandId, Long orderId) {
+}

saga/src/main/java/com/ankurm/sagademo/events/ReserveInventoryCommand.java

@@ -0,0 +1,4 @@
+package com.ankurm.sagademo.events;
+
+public record ReserveInventoryCommand(String commandId, Long orderId, String sku, int qty) {
+}

saga/src/main/java/com/ankurm/sagademo/events/ReservePaymentCommand.java

@@ -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) {
+}

saga/src/main/java/com/ankurm/sagademo/idempotency/IdempotencyGuard.java

@@ -0,0 +1,34 @@
+package com.ankurm.sagademo.idempotency;
+
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * "Has this exact command already been carried out?" -- answered by trying to insert its id
+ * and seeing whether the database's own UNIQUE constraint lets it through. REQUIRES_NEW so a
+ * duplicate here rolls back only this one insert, never whatever business transaction the
+ * caller is also in the middle of.
+ */
+@Component
+public class IdempotencyGuard {
+
+    private final ProcessedCommandRepository processed;
+
+    public IdempotencyGuard(ProcessedCommandRepository processed) {
+        this.processed = processed;
+    }
+
+    /** Returns true the first time a given commandId is claimed, false on every claim after
+     *  that. Callers are expected to do their real work only when this returns true. */
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public boolean claim(String commandId) {
+        try {
+            processed.save(new ProcessedCommand(commandId));
+            return true;
+        } catch (DataIntegrityViolationException alreadyClaimed) {
+            return false;
+        }
+    }
+}

saga/src/main/java/com/ankurm/sagademo/idempotency/ProcessedCommand.java

@@ -0,0 +1,39 @@
+package com.ankurm.sagademo.idempotency;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import jakarta.persistence.UniqueConstraint;
+
+/** One row per command id this consumer has actually carried out. The UNIQUE constraint on
+ *  commandId, not application code, is what makes the dedupe check safe under concurrent
+ *  delivery -- two threads racing to insert the same commandId can't both win. */
+@Entity
+@Table(name = "processed_commands", uniqueConstraints = @UniqueConstraint(columnNames = "commandId"))
+public class ProcessedCommand {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @Column(nullable = false)
+    private String commandId;
+
+    protected ProcessedCommand() {
+    }
+
+    public ProcessedCommand(String commandId) {
+        this.commandId = commandId;
+    }
+
+    public Long getId() {
+        return id;
+    }
+
+    public String getCommandId() {
+        return commandId;
+    }
+}

saga/src/main/java/com/ankurm/sagademo/idempotency/ProcessedCommandRepository.java

@@ -0,0 +1,8 @@
+package com.ankurm.sagademo.idempotency;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface ProcessedCommandRepository extends JpaRepository<ProcessedCommand, Long> {
+
+    long countByCommandId(String commandId);
+}

saga/src/main/java/com/ankurm/sagademo/inventory/InventoryService.java

@@ -0,0 +1,37 @@
+package com.ankurm.sagademo.inventory;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.stereotype.Service;
+
+/**
+ * Deliberately not a database table. Stock here is in-memory, reset per-test by calling
+ * {@link #setStock}, because the point of this module is saga coordination, not inventory
+ * persistence. The dual-write problem and the transactional outbox that actually fixes it for a
+ * real stock table are the subject of a different post in this series.
+ */
+@Service
+public class InventoryService {
+
+    private final ConcurrentHashMap<String, Integer> stock = new ConcurrentHashMap<>();
+
+    public void setStock(String sku, int quantity) {
+        stock.put(sku, quantity);
+    }
+
+    public int getStock(String sku) {
+        return stock.getOrDefault(sku, 0);
+    }
+
+    /** Returns true and decrements if enough stock was available; returns false and leaves
+     *  stock untouched otherwise. The check and the decrement happen under one lock per SKU so
+     *  two concurrent reservations for the last unit can't both succeed. */
+    public synchronized boolean tryReserve(String sku, int quantity) {
+        int available = stock.getOrDefault(sku, 0);
+        if (available < quantity) {
+            return false;
+        }
+        stock.put(sku, available - quantity);
+        return true;
+    }
+}

saga/src/main/java/com/ankurm/sagademo/orchestration/InventoryOrchestrationHandler.java

@@ -0,0 +1,45 @@
+package com.ankurm.sagademo.orchestration;
+
+import com.ankurm.sagademo.Topics;
+import com.ankurm.sagademo.events.InventoryReply;
+import com.ankurm.sagademo.events.ReserveInventoryCommand;
+import com.ankurm.sagademo.idempotency.IdempotencyGuard;
+import com.ankurm.sagademo.inventory.InventoryService;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * The one listener in this module where a duplicate delivery would be genuinely dangerous: a
+ * command, not a notification, and the action it triggers -- decrementing stock -- is not safe
+ * to run twice. {@link IdempotencyGuard#claim} is checked before touching inventory at all;
+ * a redelivered command with the same commandId is dropped, silently, with no second reply.
+ */
+@Component
+public class InventoryOrchestrationHandler {
+
+    private final InventoryService inventory;
+    private final IdempotencyGuard idempotencyGuard;
+    private final KafkaTemplate<String, Object> kafka;
+
+    public InventoryOrchestrationHandler(InventoryService inventory, IdempotencyGuard idempotencyGuard,
+            KafkaTemplate<String, Object> kafka) {
+        this.inventory = inventory;
+        this.idempotencyGuard = idempotencyGuard;
+        this.kafka = kafka;
+    }
+
+    @KafkaListener(topics = Topics.ORCH_CMD_RESERVE_INVENTORY, groupId = "orch-inventory")
+    public void onReserveInventory(ReserveInventoryCommand command) {
+        if (!idempotencyGuard.claim(command.commandId())) {
+            // Already carried out once. The first delivery already sent the one reply this
+            // command will ever get; stock must not move a second time.
+            return;
+        }
+        boolean reserved = inventory.tryReserve(command.sku(), command.qty());
+        String reason = reserved ? null : "insufficient stock for " + command.sku();
+        kafka.send(Topics.ORCH_REPLY_INVENTORY,
+                new InventoryReply(command.commandId(), command.orderId(), reserved, reason));
+    }
+}

saga/src/main/java/com/ankurm/sagademo/orchestration/OrderSagaOrchestrator.java

@@ -0,0 +1,76 @@
+package com.ankurm.sagademo.orchestration;
+
+import java.util.UUID;
+
+import com.ankurm.sagademo.Topics;
+import com.ankurm.sagademo.events.InventoryReply;
+import com.ankurm.sagademo.events.PaymentReply;
+import com.ankurm.sagademo.events.RefundPaymentCommand;
+import com.ankurm.sagademo.events.RefundReply;
+import com.ankurm.sagademo.events.ReserveInventoryCommand;
+import com.ankurm.sagademo.events.ReservePaymentCommand;
+import com.ankurm.sagademo.orders.Order;
+import com.ankurm.sagademo.orders.OrderRepository;
+import com.ankurm.sagademo.orders.OrderService;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * Every decision in this saga is made here and nowhere else. Payment and inventory never see
+ * each other's events, never decide what happens next, and never know whether the order was
+ * ultimately confirmed or cancelled -- they only ever answer the one command this class sent
+ * them. Contrast with the choreography package, where no single class looks like this one.
+ */
+@Component
+public class OrderSagaOrchestrator {
+
+    private final OrderService orders;
+    private final OrderRepository orderRepository;
+    private final KafkaTemplate<String, Object> kafka;
+
+    public OrderSagaOrchestrator(OrderService orders, OrderRepository orderRepository,
+            KafkaTemplate<String, Object> kafka) {
+        this.orders = orders;
+        this.orderRepository = orderRepository;
+        this.kafka = kafka;
+    }
+
+    public Order startSaga(String sku, int qty, int amountCents) {
+        Order order = orders.createPendingOrder(sku, qty, amountCents, "ORCHESTRATION");
+        kafka.send(Topics.ORCH_CMD_RESERVE_PAYMENT,
+                new ReservePaymentCommand(UUID.randomUUID().toString(), order.getId(), amountCents));
+        return order;
+    }
+
+    @KafkaListener(topics = Topics.ORCH_REPLY_PAYMENT, groupId = "orchestrator")
+    public void onPaymentReply(PaymentReply reply) {
+        if (!reply.success()) {
+            orders.cancel(reply.orderId());
+            return;
+        }
+        Order order = orderRepository.findById(reply.orderId()).orElseThrow();
+        kafka.send(Topics.ORCH_CMD_RESERVE_INVENTORY,
+                new ReserveInventoryCommand(UUID.randomUUID().toString(), order.getId(), order.getSku(),
+                        order.getQty()));
+    }
+
+    @KafkaListener(topics = Topics.ORCH_REPLY_INVENTORY, groupId = "orchestrator")
+    public void onInventoryReply(InventoryReply reply) {
+        if (reply.success()) {
+            orders.confirm(reply.orderId());
+            return;
+        }
+        // Compensation: the orchestrator is the one that notices the failure and decides to
+        // undo payment -- payment itself never finds out inventory said no except by being
+        // told to refund.
+        kafka.send(Topics.ORCH_CMD_REFUND_PAYMENT,
+                new RefundPaymentCommand(UUID.randomUUID().toString(), reply.orderId()));
+    }
+
+    @KafkaListener(topics = Topics.ORCH_REPLY_REFUND, groupId = "orchestrator")
+    public void onRefundReply(RefundReply reply) {
+        orders.cancel(reply.orderId());
+    }
+}

saga/src/main/java/com/ankurm/sagademo/orchestration/PaymentOrchestrationHandler.java

@@ -0,0 +1,38 @@
+package com.ankurm.sagademo.orchestration;
+
+import com.ankurm.sagademo.Topics;
+import com.ankurm.sagademo.events.PaymentReply;
+import com.ankurm.sagademo.events.RefundPaymentCommand;
+import com.ankurm.sagademo.events.RefundReply;
+import com.ankurm.sagademo.events.ReservePaymentCommand;
+import com.ankurm.sagademo.payments.PaymentService;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/** Payment under orchestration does only what it's told: carry out the command, reply with the
+ *  outcome. It never decides what the reply should lead to. */
+@Component
+public class PaymentOrchestrationHandler {
+
+    private final PaymentService payments;
+    private final KafkaTemplate<String, Object> kafka;
+
+    public PaymentOrchestrationHandler(PaymentService payments, KafkaTemplate<String, Object> kafka) {
+        this.payments = payments;
+        this.kafka = kafka;
+    }
+
+    @KafkaListener(topics = Topics.ORCH_CMD_RESERVE_PAYMENT, groupId = "orch-payment")
+    public void onReservePayment(ReservePaymentCommand command) {
+        payments.reserve(command.orderId(), command.amountCents());
+        kafka.send(Topics.ORCH_REPLY_PAYMENT, new PaymentReply(command.commandId(), command.orderId(), true));
+    }
+
+    @KafkaListener(topics = Topics.ORCH_CMD_REFUND_PAYMENT, groupId = "orch-payment-refund")
+    public void onRefundPayment(RefundPaymentCommand command) {
+        payments.refund(command.orderId());
+        kafka.send(Topics.ORCH_REPLY_REFUND, new RefundReply(command.commandId(), command.orderId()));
+    }
+}

saga/src/main/java/com/ankurm/sagademo/orders/Order.java

@@ -0,0 +1,66 @@
+package com.ankurm.sagademo.orders;
+
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+
+@Entity
+@Table(name = "orders")
+public class Order {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    private String sku;
+    private int qty;
+    private int amountCents;
+    private OrderStatus status;
+
+    /** Which coordination style this particular order's saga is running under -- "CHOREOGRAPHY"
+     *  or "ORCHESTRATION". Both styles share this same Order table and the same InventoryService
+     *  stock, which is exactly why each test resets its own sku's stock rather than relying on
+     *  the other style to have left it alone. */
+    private String sagaMode;
+
+    protected Order() {
+    }
+
+    public Order(String sku, int qty, int amountCents, String sagaMode) {
+        this.sku = sku;
+        this.qty = qty;
+        this.amountCents = amountCents;
+        this.sagaMode = sagaMode;
+        this.status = OrderStatus.PENDING;
+    }
+
+    public Long getId() {
+        return id;
+    }
+
+    public String getSku() {
+        return sku;
+    }
+
+    public int getQty() {
+        return qty;
+    }
+
+    public int getAmountCents() {
+        return amountCents;
+    }
+
+    public OrderStatus getStatus() {
+        return status;
+    }
+
+    public void setStatus(OrderStatus status) {
+        this.status = status;
+    }
+
+    public String getSagaMode() {
+        return sagaMode;
+    }
+}

saga/src/main/java/com/ankurm/sagademo/orders/OrderRepository.java

@@ -0,0 +1,6 @@
+package com.ankurm.sagademo.orders;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface OrderRepository extends JpaRepository<Order, Long> {
+}

saga/src/main/java/com/ankurm/sagademo/orders/OrderService.java

@@ -0,0 +1,31 @@
+package com.ankurm.sagademo.orders;
+
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Service
+public class OrderService {
+
+    private final OrderRepository orders;
+
+    public OrderService(OrderRepository orders) {
+        this.orders = orders;
+    }
+
+    @Transactional
+    public Order createPendingOrder(String sku, int qty, int amountCents, String sagaMode) {
+        return orders.save(new Order(sku, qty, amountCents, sagaMode));
+    }
+
+    @Transactional
+    public void confirm(Long orderId) {
+        Order order = orders.findById(orderId).orElseThrow();
+        order.setStatus(OrderStatus.CONFIRMED);
+    }
+
+    @Transactional
+    public void cancel(Long orderId) {
+        Order order = orders.findById(orderId).orElseThrow();
+        order.setStatus(OrderStatus.CANCELLED);
+    }
+}

saga/src/main/java/com/ankurm/sagademo/orders/OrderStatus.java

@@ -0,0 +1,7 @@
+package com.ankurm.sagademo.orders;
+
+public enum OrderStatus {
+    PENDING,
+    CONFIRMED,
+    CANCELLED
+}

saga/src/main/java/com/ankurm/sagademo/payments/PaymentRecord.java

@@ -0,0 +1,49 @@
+package com.ankurm.sagademo.payments;
+
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+
+@Entity
+@Table(name = "payment_records")
+public class PaymentRecord {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    private Long orderId;
+    private int amountCents;
+    private PaymentStatus status;
+
+    protected PaymentRecord() {
+    }
+
+    public PaymentRecord(Long orderId, int amountCents) {
+        this.orderId = orderId;
+        this.amountCents = amountCents;
+        this.status = PaymentStatus.RESERVED;
+    }
+
+    public Long getId() {
+        return id;
+    }
+
+    public Long getOrderId() {
+        return orderId;
+    }
+
+    public int getAmountCents() {
+        return amountCents;
+    }
+
+    public PaymentStatus getStatus() {
+        return status;
+    }
+
+    public void setStatus(PaymentStatus status) {
+        this.status = status;
+    }
+}

saga/src/main/java/com/ankurm/sagademo/payments/PaymentRepository.java

@@ -0,0 +1,9 @@
+package com.ankurm.sagademo.payments;
+
+import java.util.Optional;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface PaymentRepository extends JpaRepository<PaymentRecord, Long> {
+    Optional<PaymentRecord> findByOrderId(Long orderId);
+}

saga/src/main/java/com/ankurm/sagademo/payments/PaymentService.java

@@ -0,0 +1,31 @@
+package com.ankurm.sagademo.payments;
+
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * Payment always succeeds in this demo -- the saga's single point of failure is deliberately
+ * placed in inventory instead, so both coordination styles can be compared against exactly the
+ * same failure, with payment's compensation (a refund) as the thing that actually gets exercised
+ * when inventory says no.
+ */
+@Service
+public class PaymentService {
+
+    private final PaymentRepository payments;
+
+    public PaymentService(PaymentRepository payments) {
+        this.payments = payments;
+    }
+
+    @Transactional
+    public PaymentRecord reserve(Long orderId, int amountCents) {
+        return payments.save(new PaymentRecord(orderId, amountCents));
+    }
+
+    @Transactional
+    public void refund(Long orderId) {
+        PaymentRecord payment = payments.findByOrderId(orderId).orElseThrow();
+        payment.setStatus(PaymentStatus.REFUNDED);
+    }
+}

saga/src/main/java/com/ankurm/sagademo/payments/PaymentStatus.java

@@ -0,0 +1,6 @@
+package com.ankurm.sagademo.payments;
+
+public enum PaymentStatus {
+    RESERVED,
+    REFUNDED
+}

saga/src/main/resources/application.properties

@@ -0,0 +1,12 @@
+spring.application.name=saga
+
+spring.datasource.url=jdbc:h2:mem:sagadb;DB_CLOSE_DELAY=-1
+spring.jpa.hibernate.ddl-auto=update
+
+spring.kafka.consumer.auto-offset-reset=earliest
+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.sagademo.*
+
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

saga/src/test/java/com/ankurm/sagademo/ChoreographyCompensationTest.java

@@ -0,0 +1,76 @@
+package com.ankurm.sagademo;
+
+import java.time.Duration;
+import java.util.UUID;
+
+import com.ankurm.sagademo.choreography.ChoreographySagaStarter;
+import com.ankurm.sagademo.inventory.InventoryService;
+import com.ankurm.sagademo.orders.Order;
+import com.ankurm.sagademo.orders.OrderRepository;
+import com.ankurm.sagademo.orders.OrderStatus;
+import com.ankurm.sagademo.payments.PaymentRepository;
+import com.ankurm.sagademo.payments.PaymentStatus;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Same saga, one fact changed: not enough stock. Nobody told payment to expect a refund --
+ * PaymentChoreographyListener is simply also subscribed to the rejection event, and reacts to
+ * it the same way it reacts to anything else. That is choreography's compensation story in
+ * full: a participant's own subscription list is the only place the "undo" logic lives.
+ */
+@SpringBootTest
+@EmbeddedKafka(partitions = 1, bootstrapServersProperty = "spring.kafka.bootstrap-servers", topics = {
+        Topics.CHOREO_ORDER_CREATED, Topics.CHOREO_PAYMENT_RESERVED, Topics.CHOREO_INVENTORY_RESERVED,
+        Topics.CHOREO_INVENTORY_REJECTED, Topics.CHOREO_PAYMENT_REFUNDED
+})
+class ChoreographyCompensationTest {
+
+    @Autowired
+    ChoreographySagaStarter sagaStarter;
+    @Autowired
+    InventoryService inventory;
+    @Autowired
+    OrderRepository orders;
+    @Autowired
+    PaymentRepository payments;
+
+    @DynamicPropertySource
+    static void kafkaProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.datasource.url",
+                () -> "jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
+    }
+
+    @Test
+    void insufficientStockTriggersPaymentRefundAndOrderCancellation() {
+        inventory.setStock("GADGET-1", 1);
+
+        Order order = sagaStarter.placeOrder("GADGET-1", 5, 2999);
+
+        await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
+            Order reloaded = orders.findById(order.getId()).orElseThrow();
+            assertThat(reloaded.getStatus()).isEqualTo(OrderStatus.CANCELLED);
+        });
+
+        System.out.println("Order " + order.getId() + " status: "
+                + orders.findById(order.getId()).orElseThrow().getStatus());
+        System.out.println("Stock remaining for GADGET-1 (untouched by the rejected reservation): "
+                + inventory.getStock("GADGET-1"));
+        System.out.println("Payment record status: "
+                + payments.findByOrderId(order.getId()).orElseThrow().getStatus());
+
+        // The stock was never decremented -- tryReserve's own check-then-decrement refused
+        // before touching the map.
+        assertThat(inventory.getStock("GADGET-1")).isEqualTo(1);
+        assertThat(payments.findByOrderId(order.getId()).orElseThrow().getStatus())
+                .isEqualTo(PaymentStatus.REFUNDED);
+    }
+}

saga/src/test/java/com/ankurm/sagademo/ChoreographyHappyPathTest.java

@@ -0,0 +1,67 @@
+package com.ankurm.sagademo;
+
+import java.time.Duration;
+import java.util.UUID;
+
+import com.ankurm.sagademo.choreography.ChoreographySagaStarter;
+import com.ankurm.sagademo.inventory.InventoryService;
+import com.ankurm.sagademo.orders.Order;
+import com.ankurm.sagademo.orders.OrderRepository;
+import com.ankurm.sagademo.orders.OrderStatus;
+import com.ankurm.sagademo.payments.PaymentRepository;
+import com.ankurm.sagademo.payments.PaymentStatus;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+@SpringBootTest
+@EmbeddedKafka(partitions = 1, bootstrapServersProperty = "spring.kafka.bootstrap-servers", topics = {
+        Topics.CHOREO_ORDER_CREATED, Topics.CHOREO_PAYMENT_RESERVED, Topics.CHOREO_INVENTORY_RESERVED,
+        Topics.CHOREO_INVENTORY_REJECTED, Topics.CHOREO_PAYMENT_REFUNDED
+})
+class ChoreographyHappyPathTest {
+
+    @Autowired
+    ChoreographySagaStarter sagaStarter;
+    @Autowired
+    InventoryService inventory;
+    @Autowired
+    OrderRepository orders;
+    @Autowired
+    PaymentRepository payments;
+
+    @DynamicPropertySource
+    static void kafkaProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.datasource.url",
+                () -> "jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
+    }
+
+    @Test
+    void orderGoesThroughPaymentAndInventoryAndEndsConfirmed() {
+        inventory.setStock("GADGET-1", 10);
+
+        Order order = sagaStarter.placeOrder("GADGET-1", 2, 2999);
+
+        await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
+            Order reloaded = orders.findById(order.getId()).orElseThrow();
+            assertThat(reloaded.getStatus()).isEqualTo(OrderStatus.CONFIRMED);
+        });
+
+        System.out.println("Order " + order.getId() + " status: "
+                + orders.findById(order.getId()).orElseThrow().getStatus());
+        System.out.println("Stock remaining for GADGET-1: " + inventory.getStock("GADGET-1"));
+        System.out.println("Payment record status: "
+                + payments.findByOrderId(order.getId()).orElseThrow().getStatus());
+
+        assertThat(inventory.getStock("GADGET-1")).isEqualTo(8);
+        assertThat(payments.findByOrderId(order.getId()).orElseThrow().getStatus())
+                .isEqualTo(PaymentStatus.RESERVED);
+    }
+}

saga/src/test/java/com/ankurm/sagademo/IdempotentConsumerTest.java

@@ -0,0 +1,105 @@
+package com.ankurm.sagademo;
+
+import java.time.Duration;
+import java.util.UUID;
+
+import com.ankurm.sagademo.events.InventoryReply;
+import com.ankurm.sagademo.events.ReserveInventoryCommand;
+import com.ankurm.sagademo.idempotency.ProcessedCommandRepository;
+import com.ankurm.sagademo.inventory.InventoryService;
+
+import org.junit.jupiter.api.AfterEach;
+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.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Publishes the exact same {@link ReserveInventoryCommand} -- same commandId -- twice, by
+ * hand, to simulate exactly the kind of redelivery a consumer rebalance or a producer retry
+ * can cause. This is the one listener in the saga where a duplicate is dangerous: it decrements
+ * real stock, not just a log line.
+ */
+@SpringBootTest
+@EmbeddedKafka(partitions = 1, bootstrapServersProperty = "spring.kafka.bootstrap-servers", topics = {
+        Topics.ORCH_CMD_RESERVE_INVENTORY, Topics.ORCH_REPLY_INVENTORY
+})
+class IdempotentConsumerTest {
+
+    @Autowired
+    KafkaTemplate<String, Object> kafka;
+    @Autowired
+    InventoryService inventory;
+    @Autowired
+    ProcessedCommandRepository processedCommands;
+    @Autowired
+    org.springframework.kafka.test.EmbeddedKafkaBroker embeddedKafka;
+
+    Consumer<String, Object> replyConsumer;
+
+    @DynamicPropertySource
+    static void kafkaProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.datasource.url",
+                () -> "jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
+    }
+
+    @AfterEach
+    void closeConsumer() {
+        if (replyConsumer != null) {
+            replyConsumer.close();
+        }
+    }
+
+    @Test
+    void sameCommandIdDeliveredTwiceOnlyReservesStockOnce() {
+        inventory.setStock("GADGET-DUP", 10);
+
+        var consumerProps = KafkaTestUtils.consumerProps("idempotency-test-group", "true",
+                embeddedKafka);
+        consumerProps.put("value.deserializer", "org.springframework.kafka.support.serializer.JsonDeserializer");
+        consumerProps.put("spring.json.trusted.packages", "com.ankurm.sagademo.*");
+        consumerProps.put("auto.offset.reset", "earliest");
+        var cf = new org.springframework.kafka.core.DefaultKafkaConsumerFactory<String, Object>(consumerProps);
+        replyConsumer = cf.createConsumer();
+        embeddedKafka.consumeFromAnEmbeddedTopic(replyConsumer, Topics.ORCH_REPLY_INVENTORY);
+
+        String duplicatedCommandId = UUID.randomUUID().toString();
+        ReserveInventoryCommand command = new ReserveInventoryCommand(duplicatedCommandId, 999L, "GADGET-DUP", 3);
+
+        // The exact same command, same commandId, sent twice -- this is the redelivery.
+        kafka.send(Topics.ORCH_CMD_RESERVE_INVENTORY, command);
+        kafka.send(Topics.ORCH_CMD_RESERVE_INVENTORY, command);
+
+        await().atMost(Duration.ofSeconds(10)).untilAsserted(() ->
+                assertThat(processedCommands.countByCommandId(duplicatedCommandId)).isEqualTo(1));
+
+        // Give any second (wrongly-processed) reply a moment to arrive, if it was going to.
+        ConsumerRecords<String, Object> records = KafkaTestUtils.getRecords(replyConsumer, Duration.ofSeconds(5));
+        int replyCount = records.count();
+
+        System.out.println("ProcessedCommand rows for commandId " + duplicatedCommandId + ": "
+                + processedCommands.countByCommandId(duplicatedCommandId));
+        System.out.println("Stock remaining for GADGET-DUP after two deliveries of the same command: "
+                + inventory.getStock("GADGET-DUP"));
+        System.out.println("InventoryReply messages actually published: " + replyCount);
+        for (ConsumerRecord<String, Object> record : records) {
+            System.out.println("  reply: " + record.value());
+        }
+
+        // 10 - 3 = 7 if the reservation ran exactly once. If the duplicate had also been
+        // processed, this would be 4.
+        assertThat(inventory.getStock("GADGET-DUP")).isEqualTo(7);
+        assertThat(replyCount).isEqualTo(1);
+    }
+}