Redis Streams as Your Startup's Event Bus

---
title: "Redis Streams as Your Startup's Event Bus: A Ktor Workshop"
published: true
description: "Build a production event bus with Redis Streams, consumer groups, and Ktor coroutines — delaying your Kafka migration until you actually need it."
tags: kotlin, architecture, api, backend
canonical_url: https://blog.mvpfactory.co/redis-streams-as-your-startups-event-bus
---

## What We're Building

In this workshop, I'll walk you through building a production event bus using Redis Streams with Ktor and Exposed. By the end, you'll have a working order-processing consumer with exactly-once semantics, automatic dead-letter routing, and webhook fan-out — all without touching Kafka.

Redis Streams, added in Redis 5.0, give you append-only log semantics with consumer groups. Those are the exact primitives you need for an event bus, and you're almost certainly already running Redis. Let me show you a pattern I use in every project.

## Prerequisites

- Kotlin + Ktor project with coroutines
- Exposed ORM configured with your database
- Redis 6.2+ (for `XAUTOCLAIM`)
- Lettuce Redis client with coroutine support

## Step 1: Understand the Consumer Group Mental Model

Before writing code, here is the mapping that makes everything click:

| Concept | Kafka | Redis Streams |
|---|---|---|
| Log append | `producer.send()` | `XADD` |
| Consumer group read | `poll()` | `XREADGROUP` |
| Offset commit | `commitSync()` | `XACK` |
| Uncommitted offsets | Consumer lag | Pending Entry List (PEL) |
| Rebalancing | Automatic partition reassignment | Manual via `XCLAIM` / `XAUTOCLAIM` |
| Retention | Time/size-based | `MAXLEN` / `MINID` |

The key difference: Redis Streams gives you per-message acknowledgment out of the box. No batching offset commits, no rebalancing storms.

## Step 2: Build the Coroutine Consumer

Here is the minimal setup to get this working. This consumer reads from a group with exactly-once semantics via idempotency keys:

kotlin
class StreamConsumer(
private val redis: RedisCoroutinesCommands,
private val db: Database,
private val stream: String,
private val group: String,
private val consumerId: String
) {
suspend fun consume() = coroutineScope {
try { redis.xgroupCreate(
XReadArgs.StreamOffset.from(stream, "0-0"), group,
XGroupCreateArgs.Builder.mkstream()
) } catch (_: Exception) { /* group exists */ }

    while (isActive) {
        val messages = redis.xreadgroup(
            Consumer.from(group, consumerId),
            XReadArgs.Builder.count(50).block(Duration.ofSeconds(2)),
            XReadArgs.StreamOffset.lastConsumed(stream)
        )
        messages?.forEach { msg ->
            processWithIdempotency(msg)
        }
    }
}

private suspend fun processWithIdempotency(msg: StreamMessage<String, String>) {
    val idempotencyKey = msg.id
    transaction(db) {
        val exists = ProcessedEvents
            .select { ProcessedEvents.eventId eq idempotencyKey }
            .count() > 0
        if (!exists) {
            ProcessedEvents.insert { it[eventId] = idempotencyKey }
            handleEvent(msg.body)
        }
    }
    redis.xack(stream, group, msg.id)  // ACK only after DB commit
}

}

The ordering here matters: DB commit happens *before* `XACK`. If the process crashes between commit and ACK, the message stays in the PEL and gets redelivered, but the idempotency check prevents double processing. This is exactly-once *effective* processing.

## Step 3: Add Backpressure with XPENDING and Dead-Letter Routing

The Pending Entry List is your backpressure signal. This monitor coroutine detects stuck consumers and reclaims their messages:

kotlin
launch {
while (isActive) {
delay(30_000)
val stale = redis.xautoclaim(
stream, XAutoClaimArgs.Builder
.xautoclaim(group, consumerId, 60_000, "0-0")
)
stale.messages.forEach { msg ->
val deliveryCount = redis.xpending(stream, group, msg.id, msg.id, 1)
.firstOrNull()?.nacks ?: 0
if (deliveryCount > 3) {
redis.xadd("${stream}:dlq", msg.body)
redis.xack(stream, group, msg.id)
}
}
}
}

| Signal | Action |
|---|---|
| PEL size > threshold | Pause new reads via `XPENDING` summary check |
| Message idle > 60s | Reclaim with `XAUTOCLAIM` |
| Delivery count > 3 | Route to dead-letter queue |
| Consumer lag growing | Add consumer instances to the group |

## Step 4: Webhook Fan-Out with Multiple Consumer Groups

For webhook fan-out, use multiple consumer groups on the same stream. Each group gets every message independently:

kotlin
val groups = listOf("webhook-delivery", "analytics-pipeline", "notification-sender")
groups.forEach { group ->
launch { StreamConsumer(redis, db, "orders", group, "$group-${hostId}").consume() }
}

Each group maintains its own PEL and offset. Zero coordination between them.

## Gotchas

Here is the gotcha that will save you hours:

- **ACK after commit, always.** If you `XACK` before your DB transaction commits and the process crashes, you lose the event silently. The PEL exists to protect you — let it.
- **`XAUTOCLAIM` requires Redis 6.2+.** On older versions you need manual `XCLAIM` loops. Check your version before deploying.
- **RAM is your retention ceiling.** Redis Streams live in memory. You get hours to days of retention, not weeks. Use `MAXLEN` or `MINID` trimming aggressively.
- **The docs do not mention this, but** the crossover point to Kafka is concrete: sustained throughput above ~80K msg/s, retention needs beyond 48 hours, or more than ~20 independent consumer groups. A single Redis node sustains ~120K msg/s sustained (~200K burst) at <2ms p99 — but storage efficiency is low compared to Kafka's disk-based approach.

## Conclusion

You can build a production event bus in an afternoon with Redis Streams instead of a quarter-long Kafka rollout. Wire up `XREADGROUP` + `XACK`, add idempotency keys in your Ktor consumers, and monitor backpressure through `XPENDING`. Don't plan a Kafka migration *timeline* — plan a Kafka migration *trigger* with concrete thresholds. Migrate when you hit them. Not before.