Add the protocol-comparison module

2026-09-04 00:52:18 +05:30
123 changed files with 40 additions and 3161 deletions
@@ -1,10 +1,9 @@
 # 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` and its own captured
+self-contained Maven project for one article, with its own `pom.xml`, its own numbered
-output, regenerated from the test suite rather than typed by hand. Most modules keep that
+documentation chapters, and its own captured output under `docs/output/` — regenerated by that
-output under `docs/output/` alongside numbered documentation chapters; `saga/` and `share-groups/` keep it
+module's `scripts/run-all.sh`, never typed by hand.
 directly under `output/` instead, with the equivalent depth in the article's own accordion sections.
 | Module | Article | What it demonstrates |
 |---|---|---|
@@ -14,9 +13,6 @@ directly under `output/` instead, with the equivalent depth in the article's own
 | [`sse-websocket/`](sse-websocket/README.md) | [Server-Sent Events and WebSocket on Spring Boot 4: SseEmitter, STOMP, and Which to Pick](https://ankurm.com/spring-boot-4-sse-websocket-stomp/) | The two browser-facing options side by side: the SSE lifecycle and the three ways a stream ends, STOMP's routing defaults, and the size limit that is enforced by Tomcat rather than by Spring |
 | [`protocol-comparison/`](protocol-comparison/README.md) | [RSocket vs gRPC vs WebSocket on Spring Boot 4.1: When Each One Wins](https://ankurm.com/rsocket-vs-grpc-vs-websocket-spring-boot-4-1/) | One application serving the same two operations over all three, benchmarked in one JVM, ending in the only measurement that transfers off the box: what each server produces for a consumer that asked for a hundred |
 | [`broker-comparison/`](broker-comparison/README.md) | [Kafka vs RabbitMQ vs Pulsar for Java Teams: A Decision Framework with Benchmarks](https://ankurm.com/kafka-vs-rabbitmq-vs-pulsar-java-decision-framework/) | The three brokers measured side by side on ordering, replay, consumer scaling and operational footprint, ending in a decision table where every row has a transcript behind it |
 | [`saga/`](saga/README.md) | Saga Pattern in Spring Boot: Orchestration vs Choreography with Kafka | The same order/payment/inventory saga built twice against a real broker &mdash; choreography and orchestration &mdash; driven through the same inventory failure so the compensating transaction can be compared side by side, plus a real duplicate-delivery test for the idempotent-consumer problem |
 | [`share-groups/`](share-groups/README.md) | [Queues for Kafka: Share Groups with Spring for Apache Kafka 4](https://ankurm.com/kafka-share-groups-spring-kafka-4-queues-for-kafka/) | Share groups against a real Kafka 4.3.1 broker: six consumers on two partitions, the acquire mode that decides whether they really share, the three acknowledgement types, a group that starts at the end of the topic, and the ordering you give up |
 | [`kafka-streams/`](kafka-streams/README.md) | Kafka Streams with Spring Boot 4: Real-Time Fraud Detection with Vector Similarity | A fraud topology tested entirely under `TopologyTestDriver`: cosine-similarity scoring, a 5-minute windowed count that alerts exactly once, and a KTable blocklist join with a tombstone, plus the Spring wiring that needs no broker to load |
 The brokers make an instructive set. Kafka's consumer holds an offset and the broker remembers
 nothing about individual records; RabbitMQ's broker owns the message until it is acknowledged and
@@ -1,44 +0,0 @@
 # 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.
@@ -1,6 +0,0 @@
 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
@@ -1,2 +0,0 @@
 alerts <- c1 SIMILAR:1.0000:c1,4700.0,3,crypto
 alerts emitted: 1
@@ -1,5 +0,0 @@
 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
@@ -1,3 +0,0 @@
 before blocking, alerts: 0
 alerts <- c5 BLOCKED:reported stolen
 after the tombstone, alerts: 0
@@ -1,39 +0,0 @@
 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
@@ -1,3 +0,0 @@
 StreamsBuilderFactoryBean present: true
 auto-startup: false
 running: false
@@ -1,8 +0,0 @@
 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
@@ -1,56 +0,0 @@
 <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>
@@ -1,9 +0,0 @@
 #!/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
@@ -1,8 +0,0 @@
 #!/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
@@ -1,11 +0,0 @@
 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);
    }
 }
@@ -1,51 +0,0 @@
 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;
    }
 }
@@ -1,20 +0,0 @@
 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);
    }
 }
@@ -1,64 +0,0 @@
 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()));
    }
 }
@@ -1,11 +0,0 @@
 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);
    }
 }
@@ -1,15 +0,0 @@
 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]);
    }
 }
@@ -1,4 +0,0 @@
 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
@@ -1,27 +0,0 @@
 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());
    }
 }
@@ -1,140 +0,0 @@
 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));
    }
 }
@@ -1,23 +0,0 @@
 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);
    }
 }
@@ -1 +0,0 @@
 target/
@@ -1,13 +0,0 @@
 # 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.
@@ -1,21 +0,0 @@
 <?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>
@@ -1,9 +0,0 @@
 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); }
 }
@@ -1,3 +0,0 @@
 package com.ankurm.kafka;
 public record OrderEvent(String id, int qty) { }
@@ -1,18 +0,0 @@
 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());
    }
 }
@@ -1,14 +0,0 @@
 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));
        }
    }
 }
@@ -1,8 +0,0 @@
 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
@@ -1,27 +0,0 @@
 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"));
    }
 }
@@ -1,21 +0,0 @@
 <?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>
@@ -1,9 +0,0 @@
 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); }
 }
@@ -1,3 +0,0 @@
 package com.ankurm.kafka;
 public record OrderEvent(String id, int qty) { }
@@ -1,18 +0,0 @@
 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());
    }
 }
@@ -1,14 +0,0 @@
 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));
        }
    }
 }
@@ -1,8 +0,0 @@
 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
@@ -1,27 +0,0 @@
 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"));
    }
 }
@@ -1,12 +0,0 @@
 $ 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}
@@ -1,5 +0,0 @@
 $ 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
@@ -1,15 +0,0 @@
 $ 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
@@ -1,3 +0,0 @@
 $ 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
@@ -1,13 +0,0 @@
 $ 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}
@@ -1,2 +0,0 @@
 $ 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 
@@ -1,44 +0,0 @@
 #!/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
@@ -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     # six tests, then re-measure docs/output/
+./scripts/run-all.sh     # five 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,8 +63,7 @@ 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 |
-| [`rsocket-codecs.txt`](docs/output/rsocket-codecs.txt) | what Boot configures for RSocket, and why CBOR wins |
+| [`tests.txt`](docs/output/tests.txt) | 5 tests |
 | [`tests.txt`](docs/output/tests.txt) | 6 tests |
 ## Read the benchmark honestly
@@ -86,6 +85,5 @@ 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, and it wins by list order.** It is binary JSON, not a schema
+4. **CBOR keeps the field names.** It is binary JSON, not a schema format: 67 bytes against
-   format: 67 bytes against protobuf's 35 and JSON's 83. Boot puts `JacksonCborEncoder` ahead of
+   protobuf's 35 and JSON's 83, and the difference is entirely the punctuation.
   `JacksonJsonEncoder`, and a bare `RSocketStrategies.create()` has neither.
@@ -18,12 +18,7 @@ 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`, and re-measured **in their own JVM invocation**.
+Re-measured on every `./scripts/run-all.sh`.
 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
@@ -35,19 +35,7 @@ Two practical notes:
 - **Spring's RSocket default is CBOR**, not JSON, which is why
  `spring-boot-starter-rsocket` brings `jackson-dataformat-cbor`. If you are debugging with
-  `tcpdump` and expecting to read your payloads, that is why you cannot. The mechanism is codec
+  `tcpdump` and expecting to read your payloads, that is why you cannot.
  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.
@@ -20,13 +20,12 @@ 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              103,425    449,318  1,208,219
+gRPC              142,000    680,542  1,346,233
-WebSocket          19,664    143,870    320,255
+WebSocket          23,693    125,099    311,617
 ```
 RSocket produced **exactly one hundred**, at every quiet period. gRPC produced over a million in
-two seconds. Raw WebSocket, three hundred thousand. These are re-measured on every run and the
+two seconds. Raw WebSocket, three hundred thousand.
 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.)
@@ -47,7 +46,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 &mdash; a socket buffer, Spring's `sendBufferSizeLimit`, the
-heap. Every one of those 320&nbsp;255 messages was serialised and written for a client that had
+heap. Every one of those 311&nbsp;617 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.
@@ -59,14 +58,12 @@ citing it ([`grpc-isready.txt`](output/grpc-isready.txt)):
 ```
 handler                      quiet 0.5s   quiet 2s
-plain onNext loop               258,145  1,421,778
+plain onNext loop               353,996  1,883,443
-isReady() checked before write    667,259  1,044,018
+isReady() checked before write  1,285,787     63,859
 ```
-**It did not bound the producer.** The `isReady()` row is not lower than the plain one in any
+**It did not bound the producer, and it was not even stable** &mdash; the same handler produced
-useful sense &mdash; it produced 667&nbsp;thousand messages in half a second for a client that
+1.3&nbsp;million in half a second on one row and 64&nbsp;thousand in two seconds on the next. I
 wanted a hundred &mdash; and across runs it has come out both far above and far below the plain
 loop, which is its own kind of answer. I
 have not established whether the surplus accumulates in the server's outbound queue, in the client
 transport, or is discarded after the deframer, and I am not going to assert a mechanism I have not
 read the source for.
@@ -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              103,425    449,318  1,208,219
+gRPC              142,000    680,542  1,346,233
-WebSocket          19,664    143,870    320,255
+WebSocket          23,693    125,099    311,617
 RSocket is flat: the server produced exactly what request(n) asked for.
-gRPC grows 11.7x between 0.5 s and 2 s; WebSocket grows 16.3x.
+gRPC grows 9.5x between 0.5 s and 2 s; WebSocket grows 13.2x.
 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.)
@@ -1,4 +1,4 @@
 === gRPC server streaming, client takes 100 then stops ===
 handler                      quiet 0.5s   quiet 2s
-plain onNext loop               258,145  1,421,778
+plain onNext loop               353,996  1,883,443
-isReady() checked before write    667,259  1,044,018
+isReady() checked before write  1,285,787     63,859
@@ -1,4 +1,4 @@
 === request/response, 5000 sequential calls after 2000 warm-up, loopback, JDK 25.0.4.1 ===
-gRPC         n=5000   p50=   302.0 us  p99=    994.7 us  mean=   335.1 us  ~2,984 calls/s
+gRPC         n=5000   p50=   291.3 us  p99=   1015.3 us  mean=   324.4 us  ~3,083 calls/s
-RSocket      n=5000   p50=   265.9 us  p99=   1166.2 us  mean=   311.9 us  ~3,206 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=   122.8 us  p99=   1893.9 us  mean=   184.8 us  ~5,410 calls/s
+WebSocket    n=5000   p50=   124.3 us  p99=   1447.8 us  mean=   172.8 us  ~5,786 calls/s
@@ -1,4 +0,0 @@
 === 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]
@@ -1,4 +1,4 @@
 === server streaming, 50000 messages on one connection, loopback ===
-gRPC          50,000 msgs in  1.341 s  =      37,285 msgs/s  ( 26.82 us each)
+gRPC          50,000 msgs in  1.116 s  =      44,801 msgs/s  ( 22.32 us each)
-RSocket       50,000 msgs in  0.900 s  =      55,580 msgs/s  ( 17.99 us each)
+RSocket       50,000 msgs in  0.702 s  =      71,175 msgs/s  ( 14.05 us each)
-WebSocket     50,000 msgs in  0.839 s  =      59,630 msgs/s  ( 16.77 us each)
+WebSocket     50,000 msgs in  0.459 s  =     109,022 msgs/s  (  9.17 us each)
@@ -1,6 +1,5 @@
-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: 0.301 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.84 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: 11.31 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: 26.70 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: 3.055 s -- in com.ankurm.protocols.StreamThroughputBenchmarkTest
 Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 7.244 s -- in com.ankurm.protocols.StreamThroughputBenchmarkTest
@@ -8,22 +8,14 @@ set -eu
 cd "$(dirname "$0")/.."
 LOG=/tmp/protocol-comparison-test.log
 # Everything, for the test count and the non-timing transcripts.
 MAVEN_OPTS=${MAVEN_OPTS:--Xmx900m} mvn -B test > "$LOG" 2>&1
 # The two timing benchmarks are then re-run ALONE. This is not tidiness: the back-pressure and
 # isReady tests leave unbounded producers spinning, and on a two-core box that halves every
 # throughput number measured after them. Benchmarks share a JVM with nothing.
 BENCH=/tmp/protocol-comparison-bench.log
 MAVEN_OPTS=${MAVEN_OPTS:--Xmx900m} mvn -B test \
  -Dtest=RequestResponseBenchmarkTest,StreamThroughputBenchmarkTest > "$BENCH" 2>&1
 # Interleaved log lines from other contexts have to come out, or the transcripts are unreadable.
-clean() { grep -vE '^2026-|^\s*$|^WARNING|^[[:space:]]+at |^Caused by|^java\.|^org\.|Initializing|Completed initialization'; }
+clean() { grep -vE '^2026-|^\s*$|Initializing|Completed initialization'; }
 sed -n '/=== one Quote, five fields/,/^JSON is /p'        "$LOG" | clean > docs/output/payload-sizes.txt
-sed -n '/=== request\/response/,/^WebSocket  /p'       "$BENCH" | clean > docs/output/request-response.txt
+sed -n '/=== request\/response/,/^WebSocket  /p'          "$LOG" | clean > docs/output/request-response.txt
-sed -n '/=== server streaming, 50000/,/^WebSocket  /p' "$BENCH" | clean > docs/output/stream-throughput.txt
+sed -n '/=== server streaming, 50000/,/^WebSocket  /p'    "$LOG" | 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"
@@ -32,9 +24,6 @@ sed -n '/=== server streaming, 50000/,/^WebSocket  /p' "$BENCH" | clean > docs/o
  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/
@@ -59,17 +59,10 @@ 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. Catching the write
+                // socket, the send buffer or the send-time limit stops it.
-                // failure keeps an abandoned stream from filling the log with one stack trace
+                while (session.isOpen()) {
-                // per run -- and note that catching it is the ONLY notification that arrives.
+                    send(session, source.at(parts[1], PRODUCED.get()));
-                try {
+                    PRODUCED.incrementAndGet();
                    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())));
@@ -1,38 +0,0 @@
 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();
    }
 }
@@ -1,61 +0,0 @@
 # 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.
@@ -1,17 +0,0 @@
 $ 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".
@@ -1,15 +0,0 @@
 $ 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.
@@ -1,20 +0,0 @@
 $ 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.
@@ -1,16 +0,0 @@
 $ 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.
@@ -1,74 +0,0 @@
 <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>
@@ -1,12 +0,0 @@
 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);
    }
 }
@@ -1,34 +0,0 @@
 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";
 }
@@ -1,30 +0,0 @@
 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;
    }
 }
@@ -1,40 +0,0 @@
 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()));
        }
    }
 }
@@ -1,34 +0,0 @@
 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());
    }
 }
@@ -1,47 +0,0 @@
 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()));
    }
 }
@@ -1,6 +0,0 @@
 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) {
 }
@@ -1,4 +0,0 @@
 package com.ankurm.sagademo.events;
 public record InventoryReply(String commandId, Long orderId, boolean success, String reason) {
 }
@@ -1,4 +0,0 @@
 package com.ankurm.sagademo.events;
 public record InventoryReserved(Long orderId) {
 }
@@ -1,7 +0,0 @@
 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) {
 }
@@ -1,6 +0,0 @@
 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) {
 }
@@ -1,4 +0,0 @@
 package com.ankurm.sagademo.events;
 public record PaymentReply(String commandId, Long orderId, boolean success) {
 }
@@ -1,4 +0,0 @@
 package com.ankurm.sagademo.events;
 public record PaymentReserved(Long orderId, String sku, int qty) {
 }
@@ -1,4 +0,0 @@
 package com.ankurm.sagademo.events;
 public record RefundPaymentCommand(String commandId, Long orderId) {
 }
@@ -1,4 +0,0 @@
 package com.ankurm.sagademo.events;
 public record RefundReply(String commandId, Long orderId) {
 }
@@ -1,4 +0,0 @@
 package com.ankurm.sagademo.events;
 public record ReserveInventoryCommand(String commandId, Long orderId, String sku, int qty) {
 }
@@ -1,8 +0,0 @@
 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) {
 }
@@ -1,34 +0,0 @@
 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;
        }
    }
 }
@@ -1,39 +0,0 @@
 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;
    }
 }
@@ -1,8 +0,0 @@
 package com.ankurm.sagademo.idempotency;
 import org.springframework.data.jpa.repository.JpaRepository;
 public interface ProcessedCommandRepository extends JpaRepository<ProcessedCommand, Long> {
    long countByCommandId(String commandId);
 }
@@ -1,37 +0,0 @@
 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;
    }
 }
@@ -1,45 +0,0 @@
 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));
    }
 }
@@ -1,76 +0,0 @@
 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());
    }
 }
@@ -1,38 +0,0 @@
 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()));
    }
 }
@@ -1,66 +0,0 @@
 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;
    }
 }
@@ -1,6 +0,0 @@
 package com.ankurm.sagademo.orders;
 import org.springframework.data.jpa.repository.JpaRepository;
 public interface OrderRepository extends JpaRepository<Order, Long> {
 }
@@ -1,31 +0,0 @@
 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);
    }
 }
@@ -1,7 +0,0 @@
 package com.ankurm.sagademo.orders;
 public enum OrderStatus {
    PENDING,
    CONFIRMED,
    CANCELLED
 }
@@ -1,49 +0,0 @@
 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;
    }
 }
@@ -1,9 +0,0 @@
 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);
 }
@@ -1,31 +0,0 @@
 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);
    }
 }
@@ -1,6 +0,0 @@
 package com.ankurm.sagademo.payments;
 public enum PaymentStatus {
    RESERVED,
    REFUNDED
 }
@@ -1,12 +0,0 @@
 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
@@ -1,76 +0,0 @@
 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);
    }
 }
@@ -1,67 +0,0 @@
 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);
    }
 }
@@ -1,105 +0,0 @@
 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);
    }
 }
--- a/Show More
+++ b/Show More
		`@@ -1,2 +0,0 @@`
			`alerts <- c1 SIMILAR:1.0000:c1,4700.0,3,crypto`
			`alerts emitted: 1`
		`@@ -1,3 +0,0 @@`
			`package com.ankurm.kafka;`

			`public record OrderEvent(String id, int qty) { }`
		`@@ -1,2 +0,0 @@`
			`$ 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`