Building a Real-Time Fraud Detection System with Kafka Streams

Payment fraud is a latency game: the window to block a fraudulent authorization is measured in tens of milliseconds, not seconds. For platform and SRE engineers running Apache Kafka in production, a fraud pipeline is less an application-development exercise than an operational one—it lives or dies on state-store sizing, rebalance behavior, and predictable processing latency. This guide walks through architecting, implementing, and operating such a system with Kafka Streams, keeping the emphasis on the operational details that keep the pipeline healthy.

The system ingests a high-velocity stream of payment transactions, enriches them against per-account behavioural profiles, evaluates rules and anomaly scores, and emits alerts—all in the low-latency critical path. We cover state-store retention and sizing, rebalance tuning that actually applies to Streams (rather than to plain consumers), interactive queries for operational dashboards, and the testing patterns that prevent 3 a.m. pages.

Architectural Overview and Data Flow

A production fraud pipeline on Kafka Streams follows a staged topology. Each stage maps to a set of topics and processors, giving operations teams independently observable and tunable stages.

The data flow has five stages:

  1. Ingestion and normalisation — Raw transaction events arrive on a source topic, potentially from multiple gateways. The first processor normalises schemas, enriches with geo-IP metadata, and drops malformed records.
  2. Behavioural profiling — A stateful processor maintains per-account aggregates: rolling transaction counts, average amounts, velocity metrics, and device-fingerprint histories.
  3. Rule evaluation — A stateless (or lightly stateful) processor applies deterministic rules: amount thresholds, velocity checks, high-risk merchant category codes, and time-of-day anomalies.
  4. Anomaly scoring — A processor computes a composite risk score, optionally calling an external model-serving endpoint for ML-based scoring.
  5. Alert emission and feedback — High-scoring transactions are written to an alerts topic and to a compacted topic for case management. A feedback loop updates behavioural profiles once alerts are dispositioned as fraud or false positive.

This is fundamentally a Stream Processing workload that leans heavily on stateful operations. The stateless/stateful split drives your capacity planning: the behavioural-profiling stage dominates state-store disk and off-heap memory usage, so plan it first. One design decision dominates the rest—keep the model-scoring call (stage 4) out of the synchronous path if its tail latency is unbounded. An external endpoint that occasionally takes 200 ms will blow your per-record budget and stall the consumer; either bound it with a strict timeout and a deterministic fallback score, or move ML scoring to an asynchronous side-path and gate authorization on the deterministic rules alone.

Implementing Stateful Behavioural Profiling

The heart of fraud detection is comparing the current transaction against historical patterns. In Kafka Streams this means Stateful Stream Processing with Kafka Streams—specifically windowed aggregations keyed by account. The profiling processor maintains rolling windows of transaction counts, summed amounts, and distinct device fingerprints.

A common operational pitfall is underestimating state-store growth. A windowed store keeps a separate aggregate per (key, window) pair, and with overlapping retention you hold multiple windows per key at once. Make the math concrete: with 20 million active accounts, a 5-minute window, and a 10-minute retention, you hold roughly two live windows per key—on the order of 40 million windowed entries at steady state, before RocksDB overhead. Size disk for the serialised payload plus RocksDB overhead—index/filter blocks, SST file metadata, and space amplification from pending compactions—which can add a meaningful multiple over the raw payload, especially under write-heavy load. Treat any rule-of-thumb multiplier as a starting hypothesis and measure it for your workload (see the sizing procedure under monitoring, below).

// Behavioural profiling: 5-minute tumbling windows for velocity metrics
KStream<String, Transaction> transactions = builder.stream(
    "transactions-normalised",
    Consumed.with(Serdes.String(), transactionSerde)
);

KTable<Windowed<String>, ProfileAggregate> profiles = transactions
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)))
    .aggregate(
        ProfileAggregate::new,
        (accountId, txn, aggregate) -> aggregate.add(txn),
        Materialized.<String, ProfileAggregate, WindowStore<Bytes, byte[]>>as("profile-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(profileAggregateSerde)
            .withRetention(Duration.ofMinutes(10))
    );

Two correctness points. First, use ofSizeAndGrace to declare grace explicitly; ofSizeWithNoGrace accepts late records up to retention but emits no closing semantics for late data, which is rarely what you want for velocity rules. Second, withRetention must be at least the window size plus the grace period—otherwise records that arrive late are silently dropped because their window has already been purged. Keep retention tight: oversizing it inflates the store without improving accuracy.

There is a subtle timing trap here: windows close on event time, advanced by the stream’s observed timestamps, not wall-clock. A quiet account that stops transacting will not advance the stream time on its partition, so its final window can stay open indefinitely until a later record arrives. For a velocity rule, that means “no transactions for 5 minutes” is not something the windowed aggregation will tell you on its own—you detect absence with a punctuator or a separate idle-timeout, not by waiting for a window to close.

To detect when late arrivals are pressing against the retention boundary, watch the end-to-end latency metrics introduced in KIP-613. These are reported under kafka.streams:type=stream-processor-node-metrics (e.g. record-e2e-latency-max) at source and terminal nodes, and are recorded at the INFO recording level or higher, which is the default (metrics.recording.level=INFO)—they disappear if someone has lowered the recording level.

Tuning for Production Resilience

Kafka Streams sets sensible defaults, but a latency-sensitive pipeline needs deliberate tuning. The configs below apply to the embedded consumer and producer that Streams manages on your behalf.

# Consumer-side: bound poll work and keep liveness generous for stateful tasks
max.poll.records=500
max.poll.interval.ms=600000
session.timeout.ms=30000
heartbeat.interval.ms=3000

# Producer-side: low-latency, safe writes
linger.ms=5
batch.size=32768
compression.type=lz4
max.in.flight.requests.per.connection=5
enable.idempotence=true

Do not set partition.assignment.strategy in a Kafka Streams app. Streams installs its own StreamsPartitionAssignor, which already performs incremental cooperative rebalancing (KIP-429). Overriding the assignment strategy with CooperativeStickyAssignor—a setting that belongs to plain consumers—will not improve Streams and can break task assignment. Streams gets cooperative rebalancing for free.

The expensive event in a stateful Streams app is not the rebalance itself but the state restore that can follow it: a reassigned task with no local copy of its store must replay the entire changelog topic from the start, which can take minutes for a large store while that task processes nothing. Three mechanisms attack this, and you usually want all three:

  • Static group membership (KIP-345). Set a stable group.instance.id per pod or container so a restart within session.timeout.ms rejoins with the same identity and skips reassignment entirely—no rebalance, no restore. This is the single biggest win for rolling restarts and pod reschedules.
  • Standby replicas. Set num.standby.replicas (default 0) to 1 so a hot copy of every store is kept warm on another instance. If the active instance dies, the standby is already caught up to within acceptable.recovery.lag (default 10000 records) and takes over with minimal restore. The cost is a second copy of all state and doubled changelog read traffic—budget for it.
  • Warmup replicas (KIP-441). When you scale out, the assignor keeps the task active on the caught-up instance and restores onto the new one in the background as a warmup replica, switching over only once it is within acceptable.recovery.lag. This is on by default; max.warmup.replicas (default 2) throttles how many move at once, and probing.rebalance.interval.ms (default 10 min, minimum 1 min) controls how often the leader checks whether warmups have caught up. The practical consequence: scaling out a stateful Streams app is not instant—plan for capacity changes to settle over several probing intervals, not seconds.
# Survive a pod restart or single-node loss without a full changelog replay
group.instance.id=fraud-streams-pod-3   # unique & stable per instance
num.standby.replicas=1

State-store caching is a topology-wide budget, not a per-store one. Since KIP-770, the control is statestore.cache.max.bytes (the renamed cache.max.bytes.buffering), with a default of 10 MB total for the whole topology, divided evenly across stream threads. There is no per-store withCacheSizeBytes method; raise the global budget instead:

# Topology-wide record cache (replaces the old cache.max.bytes.buffering)
statestore.cache.max.bytes=104857600   # 100 MB total, split across threads

A larger record cache deduplicates updates and reduces downstream/changelog traffic, but it trades against heap and delays emission until the cache flushes (bounded by commit.interval.ms). That emission delay matters here: for a latency-critical fraud path you may want a small cache so alerts emit promptly, accepting more changelog writes as the price of lower latency. The RocksDB block cache is separate and lives off-heap; tune it with a RocksDBConfigSetter. Monitor RocksDB directly through the metrics from KIP-471 and KIP-607, which are exposed under kafka.streams:type=stream-state-metrics (for example block-cache-usage and write-stall-duration-total).

Interactive Queries for Operational Dashboards

Fraud-ops teams need live visibility into account profiles without adding latency to the critical path. Kafka Streams interactive queries expose state stores for read-only lookups, which you typically wrap in your own thin HTTP layer.

// Your own REST handler wrapping a read-only windowed store lookup.
public ProfileAggregate getProfile(KafkaStreams streams, String accountId) {
    ReadOnlyWindowStore<String, ProfileAggregate> store =
        streams.store(
            StoreQueryParameters.fromNameAndType(
                "profile-store",
                QueryableStoreTypes.windowStore()
            )
        );

    Instant to = Instant.now();
    Instant from = to.minus(Duration.ofMinutes(5));

    // fetch() bounds are on window start time; iterate newest-relevant windows.
    try (WindowStoreIterator<ProfileAggregate> it = store.fetch(accountId, from, to)) {
        ProfileAggregate latest = null;
        while (it.hasNext()) {
            latest = it.next().value;
        }
        return latest != null ? latest : new ProfileAggregate();
    }
}

Three operational details. WindowStoreIterator must be closed (it holds a RocksDB iterator), and fetch(key, from, to) ranges over window start times, so it can return several windows—take the most recent. Across multiple instances, the queried key may live on another host: use KafkaStreams.queryMetadataForKey(...), which returns a KeyQueryMetadata exposing the activeHost and standbyHosts for that key, to locate the owning instance and forward the request. This routing is mandatory for any horizontally scaled deployment.

The third detail is what happens during a rebalance. A query for a key whose active task is mid-restore will otherwise throw (the store isn’t available), which surfaces as dashboard errors exactly when operators most want visibility. KIP-535 lets you serve stale reads from a standby or restoring replica during that window: route to a standbyHosts entry from KeyQueryMetadata, and use allLocalStorePartitionLags() to decide whether the staleness is acceptable for a dashboard. For fraud operations dashboards (as opposed to the blocking decision path) a few seconds of staleness is almost always preferable to an outage.

Testing Strategies That Prevent Midnight Pages

A misfiring fraud pipeline either misses fraud or floods downstream systems with false positives—both page someone. Layered testing is the defence.

TopologyTestDriver runs the full topology in-process with no broker. It processes records synchronously and gives you direct control over event time through record timestamps (it does not advance wall-clock time on its own, and commit.interval.ms/cache settings behave as if zero). That makes window advancement and late-record handling fully deterministic—and it’s the only practical way to assert the idle-window behavior described earlier, since you can feed timestamps that leave a window open and confirm nothing fires.

@Test
public void testVelocityCheckTriggersAlert() {
    try (TopologyTestDriver testDriver = new TopologyTestDriver(topology, props)) {
        TestInputTopic<String, Transaction> inputTopic = testDriver.createInputTopic(
            "transactions-normalised",
            Serdes.String().serializer(),
            transactionSerde.serializer()
        );
        TestOutputTopic<String, Alert> outputTopic = testDriver.createOutputTopic(
            "alerts",
            Serdes.String().deserializer(),
            alertSerde.deserializer()
        );

        // 6 transactions within a 5-minute window for one account.
        Instant baseTime = Instant.parse("2025-01-15T10:00:00Z");
        for (int i = 0; i < 6; i++) {
            inputTopic.pipeInput(
                "account-123",
                new Transaction("account-123", 100.00, baseTime.plusSeconds(i * 50L)),
                baseTime.plusSeconds(i * 50L)   // record timestamp drives event time
            );
        }

        List<KeyValue<String, Alert>> alerts = outputTopic.readKeyValuesToList();
        assertEquals(1, alerts.size());
        assertEquals("VELOCITY_EXCEEDED", alerts.get(0).value.getRuleId());
    }
}

For integration tests against real brokers, containerise a single-broker cluster with Testcontainers (the reliable, supported path), or use Kafka’s internal EmbeddedKafkaCluster—note that class lives in Kafka’s test sources and is not a stable public API, so pin it carefully if you depend on it. Validate exactly-once behavior (processing.guarantee=exactly_once_v2) by injecting failures mid-stream and asserting no duplicate alerts downstream. One failure mode TopologyTestDriver cannot catch, because it has no real changelog or rebalance, is incorrect state after a restore—so reserve at least one integration test that kills and restarts an instance and asserts the recovered profiles match.

Operating at Scale: Monitoring and Capacity Planning

Production observability is layered: JVM metrics, Streams metrics, and business metrics. At the thread level, kafka.streams:type=stream-thread-metrics exposes process-latency-avg and process-rate; at the processor-node level, kafka.streams:type=stream-processor-node-metrics exposes process-rate and the record-e2e-latency-* family. Together they tell you whether the pipeline is keeping pace with input throughput. (process-latency-avg and process-rate are thread-scoped; the older kafka.streams:type=stream-metrics name is not where these live in current releases.)

Alerts worth configuring:

  • Consumer lag on source topics exceeding a fraud-relevant threshold (for an intervention SLA in the tens of ms, even a few seconds of lag is an incident).
  • State-store restore time during rebalances exceeding your restart budget (often single-digit minutes). If num.standby.replicas >= 1 and you still see long restores on failover, your standbys are not keeping up—check changelog read lag, not just the active task.
  • RocksDB write stalls (write-stall-duration-total climbing), indicating disk I/O saturation or compaction pressure. This is a leading indicator of latency regression: stalls back-pressure the processing thread before process-latency-avg moves.
  • Alert emission rate deviating sharply from the rolling baseline (e.g. > 3σ), in either direction. A sudden drop to zero usually means an upstream stall, not a quiet day for fraudsters.

For state-store sizing, the only reliable figure is one you measure. The procedure: provision a representative key population at production cardinality, run production-like load for at least one full retention period (so windows turn over and compaction settles), then read actual on-disk SST size plus block-cache-usage and size-all-mem-tables from the RocksDB metrics. Project per-instance disk from that, divide by partition count to find per-task footprint, and use it to decide when to scale out—since state is partitioned by key, adding instances spreads both disk and I/O, but only down to one task per instance. The Kafka Streams operations guide and the Confluent monitoring reference document the full metric set and restore/troubleshooting steps; keep them open during incident response.

Closing the Loop: Feedback and Model Retraining

A fraud system improves only when outcomes feed back. The alerts topic is consumed by a case-management system that records dispositions; a separate processor reads the disposition topic and updates behavioural profiles—for example, relaxing velocity thresholds for accounts with high false-positive rates.

This loop is itself a Stream Processing pattern: a KStream–KTable join between the transaction stream and a compacted disposition table. The join enriches transactions with their eventual fraud labels, producing a labelled dataset you can materialise to a data lake or feed an online-learning pipeline.

Operationally, the disposition topic must be a compacted log with a retention floor on uncompacted records. Set cleanup.policy=compact and min.compaction.lag.ms=86400000, which guarantees each disposition stays in the uncompacted log head for at least 24 hours before it is eligible for compaction—long enough for every consumer (including the training job) to see it. A frequent silent failure is dispositions being compacted before downstream consumers read them, which degrades model accuracy over weeks with no obvious alarm; detect it by alerting on training-job consumer lag against min.compaction.lag.ms, not just on the lag value alone.

Building a real-time fraud detection system with Kafka Streams is an exercise in operational maturity. The topology code is short; the discipline to keep it correct and low-latency at scale—right retention, the right restore strategy, the right rebalance model—is what separates a demo from a system your on-call rotation can trust.

Back to Stateful Stream Processing With Kafka Streams