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
parent 5db56e70a0
commit 59fcdbe079
19 changed files with 520 additions and 0 deletions
@@ -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);
+    }
+}