Kafka Streams with Spring Boot 4: Real-Time Fraud Detection with Vector Similarity
A fraud-detection topology on Kafka Streams 4.2.1 and Spring Boot 4.1.1: cosine-similarity scoring against known-fraud vectors, a 5-minute windowed count, and a blocklist KTable join, all tested with TopologyTestDriver and no broker.
Imagine a card transaction arrives and you have a few milliseconds to say “this looks wrong”. A nightly batch job is too slow, and a request-time database query per rule gets expensive. Kafka Streams is a Java library that lets you write that logic as a continuously running program that reads a stream of events, remembers a little state, and writes new events out. There is no cluster to deploy: it is a library inside your Spring Boot application.
This post builds a small fraud detector with three rules and, more importantly, shows how to test it with no Kafka broker at all. The three rules are: score each transaction by how similar it looks to known fraud, alert when one card makes too many transactions in five minutes, and alert when a card on a blocklist appears.
Versions this was tested against. JDK 25 (Temurin 25.0.4.1+1), Spring Boot 4.1.1, Spring for Apache Kafka 4.1.1, kafka-streams and kafka-streams-test-utils4.2.1 (the version Boot 4.1.1 manages), JUnit 6.1.3. Code: the kafka-streams module of spring-messaging-demo. Six tests, all green.
Honest framing of “vector similarity”. The vectors here are six numbers I designed by hand (log of the amount, the hour as a sine and cosine, and three category flags). They are not a learned embedding, and the 0.98 threshold was picked by me to separate five sample transactions, not tuned on data. What is real is the shape of the solution: turn an event into a vector, compare with cosine similarity, act on the score. Replacing the six-number vector with a model’s embedding would change one method.
What goes wrong first: a tombstone that will not compile
Kafka has a convention for “delete this key”: send a record whose value is null, called a tombstone. The test for the blocklist needs one, so the obvious line is blocked.pipeInput("c5", null). It does not compile:
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
That is the compiler’s own message, captured by compile-broken.sh from AmbiguousNullTombstone.java into output/07-ambiguous-null-tombstone.txt. There are two overloads, one taking (key, value) and one taking (value, timestamp), and a bare null fits the second argument of both. The fix is a cast that says which one you mean: blocked.pipeInput("c5", (String) null).
The bigger failure to carry forward. Everything below runs in TopologyTestDriver, which is excellent for logic and says nothing about a real cluster: partition counts, rebalances, record caching, commit intervals, or broker behaviour. I did not run a broker for this post. Treat the tests as proof of what the topology computes, not of how it deploys.
The mental model: streams, tables, windows, stores
Four terms cover almost everything here.
A KStream is a never-ending list of events: every record matters. A KTable is a “latest value per key” view: a new record for the same key replaces the old one, and a null value deletes it. A window chops a stream into time slices so you can count “per five minutes”. A state store is the small local database Kafka Streams keeps so that counts and tables survive between records.
Three branches from one input topic, one output topic. All three branches really run. Notice the two orange and green boxes: the count and the blocklist are the only places that remember anything between records. The scoring branch is stateless, which is why it is the easiest to scale and to test.
The whole detector is one static method that takes a StreamsBuilder. Keeping Spring out of it is the design decision that makes everything testable without a broker.
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(),
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) -> 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()));
}
Source: FraudTopology.java (imports trimmed). Transactions travel as the text card,amount,hour,category so every serde is the built-in String one and the tests are readable; a real system would use JSON or Avro.
From Embedder.java. The hour becomes a sine and cosine pair so that 23:00 and 01:00 end up close together instead of 22 apart. Cosine similarity then measures the angle between two vectors: 1.0 means the same direction, 0 means unrelated, negative means opposite. The score is the best match against two stored “confirmed fraud” prototypes. Five sample transactions:
Transcript: output/01-similarity-scores.txt. Large, middle-of-the-night, risky-category transactions score 1.0000 (rounded to four places); a daytime crypto purchase of 2500 scores only 0.2136, so the category flag alone does not trigger anything. Run through the topology, only the fraud-shaped one produces an alert:
With real embeddings you would normally keep the prototype vectors in a vector store and query nearest neighbours; here two prototypes sit in a list, which is the same idea at toy scale. For the vector-store side, see VectorStore.upsert() in Spring AI 2.1.
The threshold is a precision/recall dial. Choose it from labelled data, not from five examples.
Rule 2: count per card in a five-minute window
A tumbling window is a fixed time slice with no overlap: 10:00–10:05, 10:05–10:10, and so on. count() keeps one number per card per window in the state store named txn-count-5m. Six transactions in two minutes, then one more ten minutes later:
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
Transcript: output/03-velocity-window.txt. Read it in order. Six transactions produced exactly one alert, because the filter is count == 4 and not count >= 4: the 5th and 6th transactions also exceed the limit, but we only want one alert per window. The ten-minute-later transaction lands in a new window whose count starts again at one, and the state store dump shows both windows side by side. The last two lines read the store directly, which is a neat way to assert on state and not only on output.
The trap hiding in == 4. This rule relies on every intermediate count being emitted. Under TopologyTestDriver it is: one input, one output update. In a running application Kafka Streams can cache and merge updates to the same key before forwarding them, so in principle a downstream == 4 check could see 3 and then 5 and never 4. I did not reproduce that, because it needs a running application with caching and a broker; I am flagging it from how the Kafka documentation describes record caching. If exactly-once alerting matters, test it on a real cluster, or alert on >= 4 and de-duplicate downstream.
Rule 3: join each transaction to a blocklist table
The blocklist is a KTable: the latest reason per card. A stream-table join looks up the table at the moment each transaction arrives and emits only when the key is present:
before blocking, alerts: 0
alerts <- c5 BLOCKED:reported stolen
after the tombstone, alerts: 0
Transcript: output/04-blocklist-table-join.txt. Three phases of the same card: a transaction before it was blocked passes silently; after blocked-cards receives “reported stolen” the same transaction alerts; after the tombstone (the cast from the first section) the table forgets the card and the next transaction is quiet again. The join is not retroactive — blocking a card does not re-alert on earlier transactions.
Going deeper: windows and joins
I used TimeWindows.ofSizeWithNoGrace, which drops records that arrive after the window closes. Late-arriving data needs a grace period; I did not test it.
A stream-table join requires both topics to be keyed identically and to have the same number of partitions (the “co-partitioning” rule in the Kafka docs). TopologyTestDriver does not enforce partition counts, so a mismatch will only appear against a real cluster.
StreamsBuilder b = new StreamsBuilder();
FraudTopology.wire(b);
Topology 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());
alerts = driver.createOutputTopic(FraudTopology.OUT, new StringDeserializer(), new StringDeserializer());
Source: FraudTopologyTest.java (setup, abridged). TopologyTestDriver runs the topology in-process, synchronously. You pipe a record into an input topic and read what came out of the output topic immediately, with a clock you control — the third constructor argument and the per-record timestamps are how the window tests move time without waiting five real minutes. The bootstrap server dummy:1 is never contacted.
Spring contributes very little, which is the point. @EnableKafkaStreams provides a StreamsBuilder bean and a lifecycle bean that starts and stops the streams instance; your configuration class hands the builder to the plain-Java topology.
@Configuration
@EnableKafkaStreams
public class FraudStreamsConfig {
@Autowired
public void wire(StreamsBuilder builder) {
FraudTopology.wire(builder);
}
}
Source: FraudStreamsConfig.java. The module’s application.properties sets spring.kafka.streams.auto-startup=false so the context can load without a broker; flip it to true when you have one. A test proves the wiring:
Transcript: output/06-spring-context.txt, from ContextLoadsTest.java. The factory bean exists and holds the topology; it is simply not running. That is the most this environment can show about the Spring side: starting it for real needs a broker, which I did not have.
Reference: the topology as Kafka Streams describes itTopology.describe() prints every processor, store and topic. It is the quickest way to confirm that your DSL code built what you meant. Here it shows one sub-topology, one windowed state store (txn-count-5m), one table store (blocked-cards-STATE-STORE-0000000010), and three sinks all writing to alerts. Full text: output/05-topology-description.txt. The first lines:
The three arrows out of the source are the three branches from the diagram. Auto-generated names such as KSTREAM-AGGREGATE-0000000005 encode the position in the DSL; adding or reordering operators can renumber them, which matters for state-store compatibility between deployments. Name the operators and stores you care about (as score and txn-count-5m are named here).
Reference: run it yourself
$ git clone https://ankurm.com/git.app/asmhatre/spring-messaging-demo.git
$ cd spring-messaging-demo/kafka-streams
$ ./scripts/run-all.sh
run-all.sh rewrites every file under output/. The resolved dependency versions on this machine were kafka-streams, kafka-clients and kafka-streams-test-utils 4.2.1 and spring-kafka 4.1.1.
Should you even do this? Kafka Streams earns its keep when decisions must be made per event, with state, at stream speed — velocity checks and blocklist joins are good examples. If you only need to score each transaction independently, a plain @KafkaListener calling a scoring service is simpler. If your “vector similarity” means thousands of prototypes, put them in a vector store and query it rather than looping in memory. And whatever you choose, do not ship on the strength of these tests alone: they prove the logic, not caching behaviour, rebalances or partitioning. Run the topology against a real cluster with realistic traffic first.
No Comments yet!