diff --git a/README.md b/README.md index 9b9f8bc..1e3db42 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ directly under `output/` instead, with the equivalent depth in the article's own | [`broker-comparison/`](broker-comparison/README.md) | [Kafka vs RabbitMQ vs Pulsar for Java Teams: A Decision Framework with Benchmarks](https://ankurm.com/kafka-vs-rabbitmq-vs-pulsar-java-decision-framework/) | The three brokers measured side by side on ordering, replay, consumer scaling and operational footprint, ending in a decision table where every row has a transcript behind it | | [`saga/`](saga/README.md) | Saga Pattern in Spring Boot: Orchestration vs Choreography with Kafka | The same order/payment/inventory saga built twice against a real broker — choreography and orchestration — driven through the same inventory failure so the compensating transaction can be compared side by side, plus a real duplicate-delivery test for the idempotent-consumer problem | | [`share-groups/`](share-groups/README.md) | [Queues for Kafka: Share Groups with Spring for Apache Kafka 4](https://ankurm.com/kafka-share-groups-spring-kafka-4-queues-for-kafka/) | Share groups against a real Kafka 4.3.1 broker: six consumers on two partitions, the acquire mode that decides whether they really share, the three acknowledgement types, a group that starts at the end of the topic, and the ordering you give up | +| [`kafka-streams/`](kafka-streams/README.md) | Kafka Streams with Spring Boot 4: Real-Time Fraud Detection with Vector Similarity | A fraud topology tested entirely under `TopologyTestDriver`: cosine-similarity scoring, a 5-minute windowed count that alerts exactly once, and a KTable blocklist join with a tombstone, plus the Spring wiring that needs no broker to load | The brokers make an instructive set. Kafka's consumer holds an offset and the broker remembers nothing about individual records; RabbitMQ's broker owns the message until it is acknowledged and diff --git a/kafka-streams/README.md b/kafka-streams/README.md new file mode 100644 index 0000000..85a9219 --- /dev/null +++ b/kafka-streams/README.md @@ -0,0 +1,43 @@ +# 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 | + +The article carries the explanations; there is deliberately no `docs/` folder. diff --git a/kafka-streams/output/01-similarity-scores.txt b/kafka-streams/output/01-similarity-scores.txt new file mode 100644 index 0000000..2ebb48a --- /dev/null +++ b/kafka-streams/output/01-similarity-scores.txt @@ -0,0 +1,6 @@ +c1,4700.0,3,crypto score=1.0000 +c1,3000.0,2,electronics score=1.0000 +c1,45.0,18,grocery score=-0.1185 +c1,120.0,13,electronics score=0.1387 +c1,2500.0,14,crypto score=0.2136 +threshold: 0.98 diff --git a/kafka-streams/output/02-similarity-alerts.txt b/kafka-streams/output/02-similarity-alerts.txt new file mode 100644 index 0000000..b0458fe --- /dev/null +++ b/kafka-streams/output/02-similarity-alerts.txt @@ -0,0 +1,2 @@ +alerts <- c1 SIMILAR:1.0000:c1,4700.0,3,crypto +alerts emitted: 1 diff --git a/kafka-streams/output/03-velocity-window.txt b/kafka-streams/output/03-velocity-window.txt new file mode 100644 index 0000000..738a02d --- /dev/null +++ b/kafka-streams/output/03-velocity-window.txt @@ -0,0 +1,5 @@ +alerts <- c9 VELOCITY:4 in 5m +alerts emitted for 6 transactions in 2 minutes: 1 +alerts after one more transaction 10 minutes later: 0 +state store window 2026-10-14T10:00:00Z count=6 +state store window 2026-10-14T10:10:00Z count=1 diff --git a/kafka-streams/output/04-blocklist-table-join.txt b/kafka-streams/output/04-blocklist-table-join.txt new file mode 100644 index 0000000..2975916 --- /dev/null +++ b/kafka-streams/output/04-blocklist-table-join.txt @@ -0,0 +1,3 @@ +before blocking, alerts: 0 +alerts <- c5 BLOCKED:reported stolen +after the tombstone, alerts: 0 diff --git a/kafka-streams/output/05-topology-description.txt b/kafka-streams/output/05-topology-description.txt new file mode 100644 index 0000000..7cdb895 --- /dev/null +++ b/kafka-streams/output/05-topology-description.txt @@ -0,0 +1,39 @@ +Topologies: + Sub-topology: 0 + Source: KSTREAM-SOURCE-0000000000 (topics: [transactions]) + --> KSTREAM-AGGREGATE-0000000005, KSTREAM-JOIN-0000000013, KSTREAM-MAPVALUES-0000000001 + Processor: KSTREAM-AGGREGATE-0000000005 (stores: [txn-count-5m]) + --> KTABLE-TOSTREAM-0000000006 + <-- KSTREAM-SOURCE-0000000000 + Processor: KSTREAM-MAPVALUES-0000000001 (stores: []) + --> score + <-- KSTREAM-SOURCE-0000000000 + Processor: KTABLE-TOSTREAM-0000000006 (stores: []) + --> KSTREAM-FILTER-0000000007 + <-- KSTREAM-AGGREGATE-0000000005 + Processor: KSTREAM-FILTER-0000000007 (stores: []) + --> KSTREAM-MAP-0000000008 + <-- KTABLE-TOSTREAM-0000000006 + Processor: score (stores: []) + --> KSTREAM-FILTER-0000000003 + <-- KSTREAM-MAPVALUES-0000000001 + Processor: KSTREAM-FILTER-0000000003 (stores: []) + --> KSTREAM-SINK-0000000004 + <-- score + Processor: KSTREAM-JOIN-0000000013 (stores: [blocked-cards-STATE-STORE-0000000010]) + --> KSTREAM-SINK-0000000014 + <-- KSTREAM-SOURCE-0000000000 + Processor: KSTREAM-MAP-0000000008 (stores: []) + --> KSTREAM-SINK-0000000009 + <-- KSTREAM-FILTER-0000000007 + Source: KSTREAM-SOURCE-0000000011 (topics: [blocked-cards]) + --> KTABLE-SOURCE-0000000012 + Sink: KSTREAM-SINK-0000000004 (topic: alerts) + <-- KSTREAM-FILTER-0000000003 + Sink: KSTREAM-SINK-0000000009 (topic: alerts) + <-- KSTREAM-MAP-0000000008 + Sink: KSTREAM-SINK-0000000014 (topic: alerts) + <-- KSTREAM-JOIN-0000000013 + Processor: KTABLE-SOURCE-0000000012 (stores: [blocked-cards-STATE-STORE-0000000010]) + --> none + <-- KSTREAM-SOURCE-0000000011 diff --git a/kafka-streams/output/06-spring-context.txt b/kafka-streams/output/06-spring-context.txt new file mode 100644 index 0000000..a1fd290 --- /dev/null +++ b/kafka-streams/output/06-spring-context.txt @@ -0,0 +1,3 @@ +StreamsBuilderFactoryBean present: true +auto-startup: false +running: false diff --git a/kafka-streams/pom.xml b/kafka-streams/pom.xml new file mode 100644 index 0000000..ae74809 --- /dev/null +++ b/kafka-streams/pom.xml @@ -0,0 +1,56 @@ + + 4.0.0 + + + + org.springframework.boot + spring-boot-starter-parent + 4.1.1 + + + + com.ankurm + kafka-streams-fraud + 1.0 + jar + + + 25 + UTF-8 + + + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-kafka + + + org.apache.kafka + kafka-streams + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.apache.kafka + kafka-streams-test-utils + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/kafka-streams/scripts/run-all.sh b/kafka-streams/scripts/run-all.sh new file mode 100755 index 0000000..1b47ac3 --- /dev/null +++ b/kafka-streams/scripts/run-all.sh @@ -0,0 +1,7 @@ +#!/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 +ls output diff --git a/kafka-streams/src/main/java/com/ankurm/streams/Embedder.java b/kafka-streams/src/main/java/com/ankurm/streams/Embedder.java new file mode 100644 index 0000000..c95a22d --- /dev/null +++ b/kafka-streams/src/main/java/com/ankurm/streams/Embedder.java @@ -0,0 +1,51 @@ +package com.ankurm.streams; + +import java.util.List; + +/** + * Turns a transaction into a vector and scores it by cosine similarity to known-fraud vectors. + * + *

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 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; + } +} diff --git a/kafka-streams/src/main/java/com/ankurm/streams/FraudStreamsConfig.java b/kafka-streams/src/main/java/com/ankurm/streams/FraudStreamsConfig.java new file mode 100644 index 0000000..828d254 --- /dev/null +++ b/kafka-streams/src/main/java/com/ankurm/streams/FraudStreamsConfig.java @@ -0,0 +1,20 @@ +package com.ankurm.streams; + +import org.apache.kafka.streams.StreamsBuilder; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafkaStreams; + +/** + * The Spring side is only this: {@code @EnableKafkaStreams} supplies a {@link StreamsBuilder} + * bean, and this method hands it to the plain-Java topology before the streams instance starts. + */ +@Configuration +@EnableKafkaStreams +public class FraudStreamsConfig { + + @Autowired + public void wire(StreamsBuilder builder) { + FraudTopology.wire(builder); + } +} diff --git a/kafka-streams/src/main/java/com/ankurm/streams/FraudTopology.java b/kafka-streams/src/main/java/com/ankurm/streams/FraudTopology.java new file mode 100644 index 0000000..755be44 --- /dev/null +++ b/kafka-streams/src/main/java/com/ankurm/streams/FraudTopology.java @@ -0,0 +1,64 @@ +package com.ankurm.streams; + +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.kstream.Consumed; +import org.apache.kafka.streams.kstream.Grouped; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.Produced; +import org.apache.kafka.streams.kstream.TimeWindows; + +import java.time.Duration; + +/** + * The topology, as plain Kafka Streams code with no Spring in it. That is deliberate: a topology + * that only needs a {@link StreamsBuilder} can be tested with TopologyTestDriver in milliseconds, + * with no broker. + * + * 

+ * 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:..."
+ *
+ */ +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 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 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())); + } +} diff --git a/kafka-streams/src/main/java/com/ankurm/streams/StreamsApplication.java b/kafka-streams/src/main/java/com/ankurm/streams/StreamsApplication.java new file mode 100644 index 0000000..5645070 --- /dev/null +++ b/kafka-streams/src/main/java/com/ankurm/streams/StreamsApplication.java @@ -0,0 +1,11 @@ +package com.ankurm.streams; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class StreamsApplication { + public static void main(String[] args) { + SpringApplication.run(StreamsApplication.class, args); + } +} diff --git a/kafka-streams/src/main/java/com/ankurm/streams/Txn.java b/kafka-streams/src/main/java/com/ankurm/streams/Txn.java new file mode 100644 index 0000000..e5ff1c3 --- /dev/null +++ b/kafka-streams/src/main/java/com/ankurm/streams/Txn.java @@ -0,0 +1,15 @@ +package com.ankurm.streams; + +/** A card transaction. */ +public record Txn(String card, double amount, int hour, String category) { + + /** Wire format: {@code card,amount,hour,category}. Plain text keeps the serde trivial to read. */ + public String encode() { + return card + "," + amount + "," + hour + "," + category; + } + + public static Txn decode(String s) { + String[] p = s.split(",", 4); + return new Txn(p[0], Double.parseDouble(p[1]), Integer.parseInt(p[2]), p[3]); + } +} diff --git a/kafka-streams/src/main/resources/application.properties b/kafka-streams/src/main/resources/application.properties new file mode 100644 index 0000000..e571b0d --- /dev/null +++ b/kafka-streams/src/main/resources/application.properties @@ -0,0 +1,4 @@ +spring.kafka.bootstrap-servers=localhost:9092 +spring.kafka.streams.application-id=fraud-streams +# Off by default here so the context can load without a broker; set true when you have one. +spring.kafka.streams.auto-startup=false diff --git a/kafka-streams/src/test/java/com/ankurm/streams/ContextLoadsTest.java b/kafka-streams/src/test/java/com/ankurm/streams/ContextLoadsTest.java new file mode 100644 index 0000000..8924eb6 --- /dev/null +++ b/kafka-streams/src/test/java/com/ankurm/streams/ContextLoadsTest.java @@ -0,0 +1,27 @@ +package com.ankurm.streams; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.ApplicationContext; +import org.springframework.kafka.config.StreamsBuilderFactoryBean; + +import static org.junit.jupiter.api.Assertions.*; + +/** Proves the Spring wiring: the factory bean exists and holds a topology, without a broker. */ +@SpringBootTest +class ContextLoadsTest { + + @Autowired ApplicationContext ctx; + + @Test + void streamsFactoryBeanIsWiredAndNotStarted() throws Exception { + var t = new Transcript("06-spring-context.txt"); + var fb = ctx.getBean(StreamsBuilderFactoryBean.class); + t.line("StreamsBuilderFactoryBean present: " + (fb != null)); + t.line("auto-startup: " + fb.isAutoStartup()); + t.line("running: " + fb.isRunning()); + t.write(); + assertFalse(fb.isRunning()); + } +} diff --git a/kafka-streams/src/test/java/com/ankurm/streams/FraudTopologyTest.java b/kafka-streams/src/test/java/com/ankurm/streams/FraudTopologyTest.java new file mode 100644 index 0000000..6afc3d0 --- /dev/null +++ b/kafka-streams/src/test/java/com/ankurm/streams/FraudTopologyTest.java @@ -0,0 +1,140 @@ +package com.ankurm.streams; + +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.TestInputTopic; +import org.apache.kafka.streams.TestOutputTopic; +import org.apache.kafka.streams.Topology; +import org.apache.kafka.streams.TopologyTestDriver; +import org.apache.kafka.streams.state.WindowStore; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Properties; + +import static org.junit.jupiter.api.Assertions.*; + +class FraudTopologyTest { + + private TopologyTestDriver driver; + private TestInputTopic txns; + private TestInputTopic blocked; + private TestOutputTopic 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 store = driver.getWindowStore(FraudTopology.COUNT_STORE); + try (KeyValueIterator, 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)); + } +} diff --git a/kafka-streams/src/test/java/com/ankurm/streams/Transcript.java b/kafka-streams/src/test/java/com/ankurm/streams/Transcript.java new file mode 100644 index 0000000..facfac0 --- /dev/null +++ b/kafka-streams/src/test/java/com/ankurm/streams/Transcript.java @@ -0,0 +1,23 @@ +package com.ankurm.streams; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +/** Collects lines and writes them to output/NN-name.txt, so the test IS the transcript. */ +final class Transcript { + private final List 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); + } +}