kafka-streams: capture the ambiguous null tombstone compile error

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

							`public record OrderEvent(String id, int qty) { }`
		`@@ -0,0 +1,2 @@`
							`$ jar tf spring-kafka-4.1.1.jar \| grep support/serializer/(Json\|Jackson)`
							`JacksonJsonDeserializer.class JacksonJsonSerde.class JacksonJsonSerializer.class JacksonJsonTypeResolver.class JsonDeserializer.class JsonSerde.class JsonSerializer.class JsonTypeResolver.class`