Real-Time Fraud Detection with Kafka

Build a Kafka Streams fraud detection pipeline with windowed aggregation.

Introduction

30 Second Summary

Think about how your bank instantly texts you when a suspicious purchase happens on your card. Behind the scenes, a system is watching every transaction in real-time and flagging patterns that look wrong.

In this project, you will build a real-time fraud detection pipeline that processes transaction events as they happen, counts how many transactions each user makes within a time window, and automatically flags suspicious activity.

What You'll Build

A Spring Boot application with a Kafka Streams topology that detects fraudulent transaction patterns using windowed aggregation and state stores.

By the end of this project, you'll have:

  • A local Kafka development environment running in Docker with KRaft mode (no ZooKeeper).
  • A Spring Boot application that produces simulated transaction events, processes them through a fraud detection topology, and exposes a REST API for querying live data.
  • A fraud detection pipeline using windowed aggregation with a named state store that flags users exceeding a transaction threshold within a 60-second window.
  • Secret Mission: Add a REST endpoint that lists all currently flagged users by querying the windowed state store with a range scan.

Are there any prerequisites?

You should have basic Java and Spring Boot familiarity (dependency injection, REST controllers, application.properties). No prior Kafka experience is required.

Set Up Kafka and the Spring Boot Project

Every real-time fraud detection system needs two things: a message broker to handle the flow of transaction events, and an application to process them.

In this step, you'll spin up Apache Kafka as your message broker and scaffold a Spring Boot project pre-configured to communicate with it.

In this step, get ready to:
  • Start a Kafka broker in Docker using KRaft mode.
  • Generate a Spring Boot project with Kafka dependencies.
  • Configure the application to connect to your local Kafka broker.
Create a Docker Compose file for Kafka

Kafka traditionally required a separate service called ZooKeeper for coordination. Since Kafka 3.3, a built-in consensus protocol called KRaft replaces ZooKeeper entirely. This means you only need a single container to run a fully functional broker.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir fraud-detection
cd fraud-detection
  • Create a docker-compose.yml file by running:
touch docker-compose.yml
  • Confirm the file exists by running:
ls

You should see docker-compose.yml in the output.

  • Open docker-compose.yml in a text editor and paste the following configuration:
services:
  kafka:
    image: apache/kafka:4.3.0
    container_name: kafka-fraud-detection
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-fraud-detection:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

What does this configuration do?

This file tells Docker to run a single Kafka broker using the official apache/kafka:4.3.0 image.

  • KAFKA_PROCESS_ROLES: broker,controller tells this single node to act as both the message broker and the KRaft controller.
  • KAFKA_LISTENERS exposes port 9092 for your application and port 9093 for internal KRaft consensus.
  • The replication factor settings are all set to 1 because you only have one broker locally.
  • Save docker-compose.yml.
  • Start the Kafka container by running:
docker compose up -d
  • Verify the container is running by running:
docker ps

You should see a container named kafka-fraud-detection with status Up and port 9092 mapped.

Container not starting?

  • Make sure Docker Desktop is running. You can check by looking for the Docker icon in your menu bar or system tray.
  • If port 9092 is already in use, another service may be occupying it. Stop the other service or change the port mapping in docker-compose.yml.
  • Run docker compose logs kafka to see detailed error output from the container.

help me troubleshoot my Kafka Docker container not starting

✔️ Awesome, I've got everything!

Great. Make sure you've saved docker-compose.yml and your container is running before continuing.

ⓧ I'd like to double check the full code

services:
  kafka:
    image: apache/kafka:4.3.0
    container_name: kafka-fraud-detection
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-fraud-detection:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
Generate the Spring Boot project

With Kafka running, you need an application to send events to it and process them. Spring Boot's project generator handles the scaffolding so you can focus on the fraud detection logic.

  • Open your browser and navigate to start.spring.io.
  • Set **Project** to Maven.
  • Set **Language** to Java.
  • Set **Spring Boot** to 4.0.7.
  • Under **Project Metadata**, set Group to com.example.
  • Set Artifact to fraud-detection.
  • Set **Java** to 25.
  • In the Dependencies section, click Add Dependencies.
  • Search for and add Spring Web.
  • Search for and add Spring for Apache Kafka.
  • Search for and add Spring Boot Actuator.
  • Click Generate to download the project zip file.

Why these three dependencies?

  • Spring Web gives you REST endpoints to query the fraud detection state store later.
  • Spring for Apache Kafka provides the Kafka producer, consumer, and Streams integration for Spring Boot.
  • Spring Boot Actuator exposes health check endpoints so you can monitor whether your Kafka connection is healthy.
  • Extract the downloaded zip into your existing fraud-detection folder on your Desktop by running:
unzip ~/Downloads/fraud-detection.zip -o -d ~/Desktop/

What does this command do?

The -o flag overwrites without prompting. The -d ~/Desktop/ flag tells unzip to extract into your Desktop folder. Since the zip contains a fraud-detection/ directory at its root, the project files merge into your existing fraud-detection folder alongside docker-compose.yml.

  • Confirm the project files were extracted by running:
ls ~/Desktop/fraud-detection

You should see docker-compose.yml, pom.xml, src/, mvnw, and other project files listed together.

Now you need to add one more dependency that start.spring.io does not offer directly: the Kafka Streams library.

  • Open pom.xml in a text editor.
  • Find the spring-kafka dependency block. It looks like this:
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
  • Add the following dependency block directly below it:
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
</dependency>

Why add kafka-streams separately?

The spring-kafka starter includes the Kafka client library but not the Streams library. Kafka Streams is an optional module that you add only when building stream processing topologies.

  • Save pom.xml.

Not sure where to add the dependency?

Place it anywhere inside the <dependencies> block. The order of dependencies does not matter to Maven. Check the full code tab below to confirm your file looks correct.

help me find where to add the kafka-streams dependency in my pom.xml

✔️ Awesome, I've got everything!

Great. Make sure pom.xml is saved before continuing.

ⓧ I'd like to double check the full code

<?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.0.7</version>
        <relativePath/>
    </parent>
    <groupId>com.example</groupId>
    <artifactId>fraud-detection</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>fraud-detection</name>
    <description>Real-Time Fraud Detection with Kafka Streams</description>
    <properties>
        <java.version>25</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>
  • Open the fraud-detection project in Google Antigravity IDE.

Opening the project folder

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar. Type Antigravity and press Enter to open the IDE. Then use File > Open Folder and select ~/Desktop/fraud-detection.

Configure the application to connect to Kafka

Spring Boot needs to know where Kafka is running and how to serialize messages. The application.properties file centralizes all of this configuration.

  • In the IDE, navigate to src/main/resources/application.properties and open it.
  • Replace the entire file content with the following configuration:
spring.application.name=fraud-detection

spring.kafka.bootstrap-servers=localhost:9092

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.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.auto-offset-reset=earliest

spring.kafka.streams.application-id=fraud-detection-streams
spring.kafka.streams.properties[default.key.serde]=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties[default.value.serde]=org.apache.kafka.common.serialization.Serdes$StringSerde

management.endpoints.web.exposure.include=health
management.endpoint.health.show-details=always

What does this configuration do?

  • spring.kafka.bootstrap-servers points to the Kafka broker you started in Docker on port 9092.
  • The producer serializers convert Java objects to bytes for Kafka. JsonSerializer means your transaction objects will be stored as JSON.
  • spring.kafka.streams.application-id uniquely identifies your Streams topology so Kafka can track its processing progress.
  • The default Serdes (serializer/deserializer pairs) tell Kafka Streams to treat keys and values as strings unless you specify otherwise.
  • The Actuator health endpoint lets you check whether the application is connected to Kafka.
  • Save application.properties.

✔️ Awesome, I've got everything!

Great. Double check you've saved the file before moving on.

ⓧ I'd like to double check the full code

spring.application.name=fraud-detection

spring.kafka.bootstrap-servers=localhost:9092

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.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.auto-offset-reset=earliest

spring.kafka.streams.application-id=fraud-detection-streams
spring.kafka.streams.properties[default.key.serde]=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.properties[default.value.serde]=org.apache.kafka.common.serialization.Serdes$StringSerde

management.endpoints.web.exposure.include=health
management.endpoint.health.show-details=always

Next, you need to enable scheduling support in the main application class. The transaction producer you'll build in the next step uses a scheduled task to send events at a fixed interval.

  • Open src/main/java/com/example/frauddetection/FraudDetectionApplication.java.
  • Find this line near the top of the class:
@SpringBootApplication
  • Add the @EnableScheduling annotation directly below it, and add its import at the top of the file. The full file should look like this:
package com.example.frauddetection;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class FraudDetectionApplication {

    public static void main(String[] args) {
        SpringApplication.run(FraudDetectionApplication.class, args);
    }
}

What does @EnableScheduling do?

This annotation activates Spring's task scheduling infrastructure. Without it, any @Scheduled methods in your application would be silently ignored. You'll use scheduling in the next step to produce transaction events at a fixed rate.

  • Save FraudDetectionApplication.java.

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file.

ⓧ I'd like to double check the full code

package com.example.frauddetection;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class FraudDetectionApplication {

    public static void main(String[] args) {
        SpringApplication.run(FraudDetectionApplication.class, args);
    }
}

Now verify the application starts and connects to Kafka successfully.

  • Open the terminal in Google Antigravity IDE.
  • Make sure you are inside the fraud-detection folder, then start the application by running:
./mvnw spring-boot:run

What does this command do?

The mvnw file is the Maven Wrapper. It downloads the correct version of Maven automatically, then builds and runs your Spring Boot application. No separate Maven installation required.

You should see Spring Boot start up in the terminal logs. Look for a line containing Started FraudDetectionApplication near the end of the output.

Seeing connection errors to Kafka?

  • Make sure your Kafka container is still running. Run docker ps in a separate terminal and confirm kafka-fraud-detection shows Up status.
  • If you see warnings about Kafka broker not available, wait a few seconds and restart the app. Kafka can take 10-15 seconds to fully initialize after container start.
  • Verify spring.kafka.bootstrap-servers in your application.properties is exactly localhost:9092.

help me fix Kafka connection errors in my Spring Boot application

Your local Kafka development environment is live and your Spring Boot application is connected. Next up, you'll build the transaction producer that floods Kafka with simulated payment events.

Build the Transaction Producer

Your Spring Boot application starts and connects to Kafka, but it is sitting idle. A fraud detection system needs a constant stream of transaction events to analyze.

In this step, you will create a Kafka producer that simulates payment transactions. It will generate randomized events every 500 milliseconds and publish them to a Kafka topic, giving the Streams topology (built in the next step) something to process.

In this step, get ready to:
  • Create a Transaction data model and a scheduled producer component.
  • Configure Kafka topics that auto-create on startup.
  • Verify that transaction events are flowing into Kafka.
Create the Transaction model and producer

Every event in your pipeline needs a consistent shape. You will define a Java record that holds three pieces of data about each payment: who made it, how much it was, and when it happened.

  • In Google Antigravity IDE, create a new file called Transaction.java inside src/main/java/com/example/frauddetection/.
  • Copy the following code into Transaction.java:
package com.example.frauddetection;

// A record is a compact class that holds immutable data
public record Transaction(String userId, double amount, long timestamp) {
}
  • Now create a second file called TransactionProducer.java in the same com.example.frauddetection package.
  • Copy the following code into TransactionProducer.java:
package com.example.frauddetection;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Random;

// Spring automatically creates an instance of this class
@Component
public class TransactionProducer {

    // KafkaTemplate handles serialization and sending to Kafka
    private final KafkaTemplate<String, Transaction> kafkaTemplate;
    private final Random random = new Random();
    // Simulated pool of users making transactions
    private final String[] userIds = {"user-1", "user-2", "user-3", "user-4", "user-5"};

    public TransactionProducer(KafkaTemplate<String, Transaction> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // Runs every 500 milliseconds to simulate continuous payment traffic
    @Scheduled(fixedRate = 500)
    public void produceTransaction() {
        String userId = userIds[random.nextInt(userIds.length)];
        double amount = Math.round(random.nextDouble() * 1000.0) / 100.0;
        Transaction transaction = new Transaction(userId, amount, System.currentTimeMillis());
        // Send to "transactions" topic, keyed by userId for partitioning
        kafkaTemplate.send("transactions", userId, transaction);
    }
}

What does this code do?

The Transaction record is a compact Java class that holds three fields: userId, amount, and timestamp. Records automatically generate constructors, getters, and equality methods.

The TransactionProducer uses Spring's @Scheduled annotation to call produceTransaction() every 500ms. Each call picks a random user from the pool, generates a random dollar amount, and sends the resulting Transaction object to the transactions topic.

Using userId as the message key ensures all transactions for the same user land on the same Kafka partition. This is critical for the windowed counting you will build in Step 3.

  • Save both Transaction.java and TransactionProducer.java.

You won't see a change yet. The next substep configures the Kafka topics and starts the application.

Seeing red underlines in your editor?

If KafkaTemplate or @Scheduled show import errors, make sure your pom.xml includes spring-kafka and that your IDE has refreshed its Maven dependencies. In Google Antigravity IDE, open the Command Palette with Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows) and run Java: Force Java Compilation or restart the Java language server.

Help me fix my import errors

✔️ Awesome, I've got everything!

Great. Double check you have saved both files before moving on.

ⓧ I'd like to double check the full code

Here is Transaction.java:

package com.example.frauddetection;

public record Transaction(String userId, double amount, long timestamp) {
}

Here is TransactionProducer.java:

package com.example.frauddetection;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Random;

@Component
public class TransactionProducer {

    private final KafkaTemplate<String, Transaction> kafkaTemplate;
    private final Random random = new Random();
    private final String[] userIds = {"user-1", "user-2", "user-3", "user-4", "user-5"};

    public TransactionProducer(KafkaTemplate<String, Transaction> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedRate = 500)
    public void produceTransaction() {
        String userId = userIds[random.nextInt(userIds.length)];
        double amount = Math.round(random.nextDouble() * 1000.0) / 100.0;
        Transaction transaction = new Transaction(userId, amount, System.currentTimeMillis());
        kafkaTemplate.send("transactions", userId, transaction);
    }
}
Configure topics and verify message flow

Kafka topics are the named channels that messages flow through. Your producer sends to a transactions topic, and later your fraud detection logic will write alerts to a fraud-alerts topic.

Why create topics with a bean?

Kafka can auto-create topics when a producer first sends to them, but that uses default settings (1 partition, no control over replication). Defining NewTopic beans gives you explicit control over partition count and replication factor from day one.

  • Create a new file called KafkaTopicConfig.java in the same src/main/java/com/example/frauddetection/ package.
  • Copy the following code into KafkaTopicConfig.java:
package com.example.frauddetection;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    // Input topic: receives raw transaction events
    @Bean
    public NewTopic transactionsTopic() {
        return TopicBuilder.name("transactions")
                .partitions(3)
                .replicas(1)
                .build();
    }

    // Output topic: receives fraud alert messages
    @Bean
    public NewTopic fraudAlertsTopic() {
        return TopicBuilder.name("fraud-alerts")
                .partitions(1)
                .replicas(1)
                .build();
    }
}

What does this code do?

  • transactionsTopic() creates a topic with 3 partitions. Multiple partitions allow Kafka to distribute load and process messages in parallel.
  • fraudAlertsTopic() creates a topic with 1 partition. This is the output channel where the Streams topology will publish fraud alerts in Step 3.
  • Both topics use 1 replica because you are running a single-node Kafka broker locally.
  • Save KafkaTopicConfig.java.
  • Return to the terminal where your application was running.
  • Stop the application by pressing Ctrl+C.
  • Restart the application by running:
./mvnw spring-boot:run

Watch the terminal output. Within a few seconds you should see log lines confirming that transactions are being sent to Kafka every 500 milliseconds.

Not seeing any transaction logs?

Make sure your Kafka container is still running. Run docker ps and confirm kafka-fraud-detection is listed with status Up. If it is not, restart it with docker compose up -d from your project root and try running the application again.

If you see a connection refused error on port 9092, Kafka may still be starting up. Wait 10-15 seconds and restart the application.

Help me debug my Kafka producer

✔️ Awesome, I've got everything!

Great. Make sure you saved KafkaTopicConfig.java before moving on.

ⓧ I'd like to double check the full code

Here is KafkaTopicConfig.java:

package com.example.frauddetection;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    @Bean
    public NewTopic transactionsTopic() {
        return TopicBuilder.name("transactions")
                .partitions(3)
                .replicas(1)
                .build();
    }

    @Bean
    public NewTopic fraudAlertsTopic() {
        return TopicBuilder.name("fraud-alerts")
                .partitions(1)
                .replicas(1)
                .build();
    }
}

Your producer is now firing transaction events into Kafka every half second. Next up, you will build the Kafka Streams topology that watches this stream and flags users who transact too frequently.

Build the Kafka Streams Fraud Detection Topology

Your producer is sending a steady stream of transaction events to Kafka. Now you need something to process those events in real-time and detect suspicious patterns.

In this step, you will build a Kafka Streams topology that watches the transaction stream, counts how many transactions each user makes within a 60-second window, and fires an alert when a user exceeds the threshold. This is the core fraud detection logic.

In this step, get ready to:
  • Create a Kafka Streams configuration class with a windowed aggregation topology.
  • Add a count-based fraud detection pipeline with a named state store.
  • Create a consumer that logs fraud alerts from the alerts topic.
Create the Kafka Streams configuration class

A Kafka Streams topology defines how events flow from input topics through transformations to output topics. Spring Boot auto-configures the streams infrastructure when it sees @EnableKafkaStreams on a configuration class.

  • Create a new file called KafkaStreamsConfig.java in src/main/java/com/example/frauddetection/.
  • Add the following code to define the fraud detection topology:
package com.example.frauddetection;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.state.WindowStore;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;

import java.time.Duration;

@Configuration(proxyBeanMethods = false)
@EnableKafkaStreams
public class KafkaStreamsConfig {

    private static final int FRAUD_THRESHOLD = 5;
    private static final Duration WINDOW_SIZE = Duration.ofSeconds(60);
    private static final Duration GRACE_PERIOD = Duration.ofSeconds(5);

    @Bean
    public KStream<String, String> fraudDetectionStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> transactions = streamsBuilder.stream("transactions");

        transactions
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeAndGrace(WINDOW_SIZE, GRACE_PERIOD))
                .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("transaction-counts")
                        .withKeySerde(Serdes.String())
                        .withValueSerde(Serdes.Long()))
                .toStream()
                .filter((windowedKey, count) -> count != null && count >= FRAUD_THRESHOLD)
                .selectKey((windowedKey, count) -> windowedKey.key())
                .mapValues((userId, count) ->
                        "FRAUD ALERT: User " + userId + " made " + count + " transactions in 60 seconds")
                .to("fraud-alerts");

        return transactions;
    }
}

What does this topology do?

This is a lot of code in one file. Here is what each section handles:

  • @Configuration(proxyBeanMethods = false) and @EnableKafkaStreams tell Spring Boot to auto-configure the Kafka Streams infrastructure using the properties you set in application.properties.
  • The three constants define your detection rules: flag any user who makes 5 or more transactions within a 60-second window. The 5-second grace period allows slightly late-arriving events to still be counted.
  • streamsBuilder.stream("transactions") opens a stream from the input topic. groupByKey() groups events by their Kafka key (the userId you set in the producer).
  • windowedBy(TimeWindows.ofSizeAndGrace(...)) creates a 60-second tumbling window. Events are bucketed into non-overlapping 60-second intervals.

What does the aggregation pipeline do?

After the windowing step, the topology counts and filters:

  • .count(Materialized.as("transaction-counts")) counts events per user per window and stores the result in a named state store. This store persists locally and is queryable via REST (you will do this in the next step).
  • .toStream() converts the KTable back into a KStream so you can continue processing.
  • .filter(...) only passes through entries where the count meets or exceeds the threshold of 5.
  • .selectKey(...) unwraps the windowed key back to the plain userId string.
  • .mapValues(...) creates a human-readable alert message.
  • .to("fraud-alerts") sends the alert to the output topic.

Why Kafka Streams instead of a plain consumer?

A plain Kafka consumer can read messages one by one, but it has no built-in concept of time windows or state. You would need to manually manage counters, handle rebalancing, and persist state across restarts.

Kafka Streams gives you windowed aggregation, fault-tolerant state stores, and exactly-once semantics out of the box. That is the core reason it exists as a separate library.

  • Save KafkaStreamsConfig.java.

You will not see fraud alerts yet because nothing is listening to the fraud-alerts topic. The next substep adds that consumer.

Application fails to start?

  • Make sure your Kafka Docker container is still running. The streams topology needs a live broker to initialize.
  • Check that spring.kafka.streams.application-id is set in your application.properties. Without it, the streams configuration fails on startup.
  • Verify the kafka-streams dependency is in your pom.xml. Spring Boot only auto-configures streams when this jar is on the classpath.

Help me debug why my Kafka Streams application won't start.

✔️ Awesome, I've got everything!

Great. Make sure you have saved KafkaStreamsConfig.java before moving on.

ⓧ I'd like to double check the full code

package com.example.frauddetection;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.state.WindowStore;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;

import java.time.Duration;

@Configuration(proxyBeanMethods = false)
@EnableKafkaStreams
public class KafkaStreamsConfig {

    private static final int FRAUD_THRESHOLD = 5;
    private static final Duration WINDOW_SIZE = Duration.ofSeconds(60);
    private static final Duration GRACE_PERIOD = Duration.ofSeconds(5);

    @Bean
    public KStream<String, String> fraudDetectionStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> transactions = streamsBuilder.stream("transactions");

        transactions
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeAndGrace(WINDOW_SIZE, GRACE_PERIOD))
                .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("transaction-counts")
                        .withKeySerde(Serdes.String())
                        .withValueSerde(Serdes.Long()))
                .toStream()
                .filter((windowedKey, count) -> count != null && count >= FRAUD_THRESHOLD)
                .selectKey((windowedKey, count) -> windowedKey.key())
                .mapValues((userId, count) ->
                        "FRAUD ALERT: User " + userId + " made " + count + " transactions in 60 seconds")
                .to("fraud-alerts");

        return transactions;
    }
}
Create the fraud alert consumer

The topology is routing alerts to the fraud-alerts topic, but you need a consumer listening on that topic to see them. A simple @KafkaListener will log each alert as it arrives.

  • Create a new file called FraudAlertConsumer.java in src/main/java/com/example/frauddetection/.
  • Add the following code to listen for fraud alerts:
package com.example.frauddetection;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class FraudAlertConsumer {

    private static final Logger log = LoggerFactory.getLogger(FraudAlertConsumer.class);

    @KafkaListener(topics = "fraud-alerts", groupId = "fraud-alert-consumer")
    public void consumeFraudAlert(String message) {
        log.warn("*** {} ***", message);
    }
}

What does this code do?

  • @KafkaListener(topics = "fraud-alerts", groupId = "fraud-alert-consumer") subscribes this method to the fraud-alerts topic. Spring manages the consumer lifecycle automatically.
  • log.warn("*** {} ***", message) logs each alert at WARN level so it stands out in your console output among the other INFO messages.
  • Save FraudAlertConsumer.java.
  • Stop the running application if it is still active, then restart it by running:
./mvnw spring-boot:run
  • Watch the console output. Within 30-60 seconds, you should see WARN-level log lines containing messages like:
*** FRAUD ALERT: User user-3 made 5 transactions in 60 seconds ***

The exact user and count will vary because the producer picks users randomly. The important thing is that alert messages appear, proving your full pipeline is working end-to-end.

Not seeing any fraud alerts?

  • Give it at least 60 seconds. The tumbling window needs to fill before any counts can exceed the threshold.
  • Check that TransactionProducer is still sending messages. Look for log lines showing transactions being produced. If the producer is not running, make sure @EnableScheduling is on your main application class.
  • Verify that spring.kafka.consumer.auto-offset-reset=earliest is set in application.properties. Without this, the consumer may skip older messages.

Why aren't my fraud alerts showing up in the logs?

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

package com.example.frauddetection;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class FraudAlertConsumer {

    private static final Logger log = LoggerFactory.getLogger(FraudAlertConsumer.class);

    @KafkaListener(topics = "fraud-alerts", groupId = "fraud-alert-consumer")
    public void consumeFraudAlert(String message) {
        log.warn("*** {} ***", message);
    }
}

Your fraud detection pipeline is fully operational. Transactions flow in, get counted per user per window, and trigger alerts when someone exceeds the threshold. Next up, you will expose the state store through a REST endpoint so you can query live transaction counts on demand.

Query the State Store via REST

Your fraud detection topology is processing transactions and firing alerts. But right now, the only way to see what's happening is by watching the logs scroll by.

In this step, you'll expose the Kafka Streams state store through a REST API. This lets you query the current transaction count for any user on demand, directly from the running stream processing topology.

In this step, get ready to:
  • Create a REST controller that queries the Kafka Streams state store.
  • Test the endpoint with live traffic flowing through the pipeline.
Create the FraudController

Kafka Streams supports a feature called interactive queries. This lets you read directly from a running state store without going through a Kafka topic. You'll use Spring's StreamsBuilderFactoryBean to get a handle on the underlying KafkaStreams instance and query the store.

  • Create a new file called FraudController.java inside src/main/java/com/example/frauddetection/.
  • Paste the following code into the file:
package com.example.frauddetection;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyWindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Instant;
import java.util.Map;

@RestController
@RequestMapping("/fraud")
public class FraudController {

    // Spring injects the factory bean that manages the Kafka Streams lifecycle
    private final StreamsBuilderFactoryBean streamsBuilderFactoryBean;

    public FraudController(StreamsBuilderFactoryBean streamsBuilderFactoryBean) {
        this.streamsBuilderFactoryBean = streamsBuilderFactoryBean;
    }

    @GetMapping("/count/{userId}")
    public ResponseEntity<Map<String, Object>> getTransactionCount(@PathVariable String userId) {
        // Get the running KafkaStreams instance from the factory bean
        KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();
        if (kafkaStreams == null) {
            return ResponseEntity.serviceUnavailable().build();
        }

        // Query the named windowed state store
        ReadOnlyWindowStore<String, Long> windowStore = kafkaStreams.store(
                StoreQueryParameters.fromNameAndType("transaction-counts", QueryableStoreTypes.windowStore()));

        // Fetch counts for this user within the last 60 seconds
        Instant now = Instant.now();
        Instant from = now.minusSeconds(60);
        long latestCount = 0;

        // Iterate through all windows and find the highest count
        try (WindowStoreIterator<Long> iterator = windowStore.fetch(userId, from, now)) {
            while (iterator.hasNext()) {
                Long count = iterator.next().value;
                if (count != null && count > latestCount) {
                    latestCount = count;
                }
            }
        }

        // Return the count along with context about the detection window
        return ResponseEntity.ok(Map.of(
                "userId", userId,
                "transactionCount", latestCount,
                "window", "60 seconds",
                "threshold", 5
        ));
    }
}

What does this code do?

This controller exposes a GET endpoint at /fraud/count/{userId} that lets you query the state store directly.

  • StreamsBuilderFactoryBean is Spring's wrapper around the Kafka Streams lifecycle. Calling getKafkaStreams() gives you the running instance so you can issue queries.
  • StoreQueryParameters.fromNameAndType locates the state store by the same name you used in KafkaStreamsConfig ("transaction-counts").
  • The WindowStoreIterator fetches all windowed entries for that user within the last 60 seconds. The code finds the highest count across any overlapping windows.
  • The try-with-resources block ensures the iterator is closed after use, preventing resource leaks.

What are interactive queries?

Kafka Streams state stores are local to the application instance. Interactive queries let you read from those stores without publishing data to another topic first.

This is the pattern that makes Kafka Streams a full-fledged queryable database, not just a stream processor.

  • Save FraudController.java.
  • Stop the running application (press Ctrl+C in the terminal) and restart it by running:
./mvnw spring-boot:run
  • Watch the logs. You should see the application start up successfully with no errors, and transaction production should resume within a few seconds.

Seeing a compilation error?

  • Make sure the package declaration at the top of FraudController.java is package com.example.frauddetection; and the file is in the correct directory.
  • Check that all import statements are present. Missing imports for StoreQueryParameters or QueryableStoreTypes will cause unresolved symbol errors.

Why won't my FraudController compile in my Kafka Streams Spring Boot project?

Test the endpoint with live traffic

Your producer has been sending transactions every 500 milliseconds since the app started. The state store is accumulating counts in real-time. Now you can query those counts on demand.

  • Open a new terminal tab (keep the Spring Boot application running in the first one).
  • Query the transaction count for user-1 by running this command:
curl http://localhost:8080/fraud/count/user-1
  • You should see a JSON response containing the current transaction count for that user. The transactionCount value will be a number reflecting how many transactions that user made in the last 60 seconds.

What should I see?

The response looks something like this:

{"userId":"user-1","transactionCount":7,"window":"60 seconds","threshold":5}

Your transactionCount will differ each time you call the endpoint because the producer is continuously sending random transactions.

  • Run the same command again a few seconds later to see the count change:
curl http://localhost:8080/fraud/count/user-1
  • The transactionCount should be different from the first call, confirming the state store is updating in real-time.
  • Try querying a different user to see their count by running:
curl http://localhost:8080/fraud/count/user-3

Getting an empty response or zero count?

  • If you get a 503 response, the Kafka Streams instance has not finished starting yet. Wait a few seconds and try again.
  • If you see transactionCount: 0 for every user, check that the producer is still running by looking at the application logs. You should see scheduled transaction messages.
  • Make sure the state store name in your controller ("transaction-counts") matches the name you used in KafkaStreamsConfig.java.

Why is my Kafka Streams state store query returning zero for all users?

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete FraudController.java file:

package com.example.frauddetection;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyWindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Instant;
import java.util.Map;

@RestController
@RequestMapping("/fraud")
public class FraudController {

    private final StreamsBuilderFactoryBean streamsBuilderFactoryBean;

    public FraudController(StreamsBuilderFactoryBean streamsBuilderFactoryBean) {
        this.streamsBuilderFactoryBean = streamsBuilderFactoryBean;
    }

    @GetMapping("/count/{userId}")
    public ResponseEntity<Map<String, Object>> getTransactionCount(@PathVariable String userId) {
        KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();
        if (kafkaStreams == null) {
            return ResponseEntity.serviceUnavailable().build();
        }

        ReadOnlyWindowStore<String, Long> windowStore = kafkaStreams.store(
                StoreQueryParameters.fromNameAndType("transaction-counts", QueryableStoreTypes.windowStore()));

        Instant now = Instant.now();
        Instant from = now.minusSeconds(60);
        long latestCount = 0;

        try (WindowStoreIterator<Long> iterator = windowStore.fetch(userId, from, now)) {
            while (iterator.hasNext()) {
                Long count = iterator.next().value;
                if (count != null && count > latestCount) {
                    latestCount = count;
                }
            }
        }

        return ResponseEntity.ok(Map.of(
                "userId", userId,
                "transactionCount", latestCount,
                "window", "60 seconds",
                "threshold", 5
        ));
    }
}

You now have a complete fraud detection pipeline: events flow in, Kafka Streams counts them in real-time windows, alerts fire automatically, and you can query the state on demand via REST. Next up, you'll take on a secret mission to extend this API even further.

Secret mission

List All Flagged Users

Your /fraud/count/{userId} endpoint is useful, but it only checks one user at a time. In a real fraud operations dashboard, analysts need to see ALL currently flagged users at a glance. Can you build an endpoint that performs a range scan over the windowed state store and returns every user who has exceeded the fraud threshold?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker container kafka-fraud-detection running Apache Kafka.
  • Spring Boot project files in the fraud-detection folder on your Desktop.

Keep everything running

No action needed. Choose this if you're still actively building or want to keep experimenting with your Kafka pipeline.

  • Your kafka-fraud-detection Docker container stays running and ready for future Kafka experiments.
  • Your Spring Boot project files remain on your Desktop, ready to reopen in Google Antigravity IDE.

Pause - I'll come back to this later

Shut down running processes to free up memory, but keep your files and container data intact for later.

  • Stop your Spring Boot application if it is still running by pressing Ctrl+C in the terminal.
  • Pause the Kafka container by running this command from the folder containing your docker-compose.yml:
docker compose stop
  • To resume later, run docker compose start from the same folder. Your Kafka data and topics will still be there.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your Kafka container, all topic data, and your local project files.

  • Stop your Spring Boot application if it is still running by pressing Ctrl+C in the terminal.
  • Remove the Kafka container and all its data by running this command from the folder containing your docker-compose.yml:
docker compose down -v
  • Delete your project folder from the Desktop by running:
cd ~/Desktop
rm -rf fraud-detection

On Windows, run this instead:

cd ~\Desktop
rmdir /s /q fraud-detection
  • Confirm the folder is gone by running ls (macOS/Linux) or dir (Windows) on your Desktop. You should no longer see fraud-detection listed.

That's a Wrap!

That's a Wrap!

Nice work! You just built a real-time fraud detection pipeline using Kafka Streams and Spring Boot that processes transaction events, detects suspicious patterns through windowed aggregation, and exposes live state through a REST API.

You've learned how to:

  • Set up a local Kafka development environment with Docker and KRaft mode (no ZooKeeper).
  • Implemented a Kafka Streams topology with windowed aggregation and state stores for real-time fraud detection.
  • Exposed Kafka Streams state store data through a REST API using interactive queries.
  • Secret Mission: Built a range-scan endpoint that lists all currently flagged users from the windowed state store.

Ready to quiz yourself?