Part 6: "The Real-Time Pulse" — Redis Streams and Consumer Groups

Story Opening

The distributed Redis Cluster was humming along, effortlessly serving GlobalMart’s massive read traffic. Maya, the Staff Engineer, had solved the caching, the memory limits, and the high availability. But the architecture was still fundamentally request-driven. When a user clicked “Checkout,” the monolithic application synchronously called the Inventory API, then the Billing API, and finally the Shipping API. If any of those services were slow or down, the entire checkout process failed.

The engineering team decided to move to an event-driven architecture. When an order was placed, the checkout service would simply publish an “OrderCreated” event and return a success message to the user immediately. The Inventory, Billing, and Shipping services would then consume that event asynchronously at their own pace.

The team’s first instinct was to deploy Apache Kafka. Kafka is the undisputed king of distributed event streaming. However, deploying and managing a production Kafka cluster (with its ZooKeeper or KRaft dependencies) is a massive operational burden. GlobalMart already had a highly available, blazingly fast Redis Cluster running.

“Can’t we just use Redis?” a junior engineer asked.

Maya knew Redis had a Pub/Sub feature, but she also knew its fatal flaw: it’s a “fire and forget” system. If the Billing service is restarting when the “OrderCreated” event is published via Pub/Sub, that event is lost forever. The message is not stored anywhere.

She needed a durable, append-only log. She needed consumer groups to allow multiple services to process the same stream of events independently. She needed acknowledgment semantics to ensure no message was lost if a consumer crashed mid-processing.

She needed Redis Streams, the data structure introduced in Redis 5.0 that brought Kafka-like semantics to the in-memory world.

Conceptual Deep-Dive

To understand Redis Streams, it’s helpful to compare them to the older Redis Pub/Sub and the industry-standard Apache Kafka.

1. Redis Pub/Sub: Fire and Forget

Redis Pub/Sub is incredibly fast and lightweight. A publisher sends a message to a channel, and Redis immediately broadcasts it to all clients currently subscribed to that channel.

The problem? It has zero memory.

If a subscriber disconnects for 5 seconds due to a network blip, any messages published during those 5 seconds are gone. Pub/Sub is perfect for ephemeral notifications (like updating a live dashboard or a chat room), but it is entirely unsuitable for critical business events like processing orders.

2. Redis Streams: The Append-Only Log

Redis Streams solves the durability problem. A Stream is an append-only data structure. When you add a message to a stream (using the XADD command), it is permanently stored in memory (and persisted to disk via RDB/AOF, as we learned in Part 3).

A message in a stream is not just a simple string; it’s a collection of key-value pairs (like a Redis Hash). Furthermore, every message is automatically assigned a unique, monotonically increasing ID based on the Redis server’s timestamp (e.g., 1698765432100-0).

Because the stream is persistent, consumers can read messages from the beginning, read from a specific ID, or block and wait for new messages to arrive (using XREAD). If a consumer crashes and restarts, it can simply resume reading from the last ID it processed.

3. Consumer Groups: Scaling the Work

If you have 10,000 orders per second, a single instance of the Billing service cannot process them fast enough. You need to run 5 instances of the Billing service and distribute the load among them.

If all 5 instances simply XREAD the stream, they will all receive every message, meaning the customer gets billed 5 times.

This is where Consumer Groups (a concept heavily inspired by Kafka) come in.

A Consumer Group is a logical entity that sits between the Stream and the consumers. It tracks which messages have been delivered to which consumers, and more importantly, which messages have been successfully processed.

When “Billing Instance A” asks the group for a new message (using XREADGROUP), the group gives it message ID: 1 and marks it as “Pending” for Instance A. When “Billing Instance B” asks, it gets ID: 2. The load is automatically distributed.

Crucially, the message remains in the “Pending Entries List” (PEL) until the consumer explicitly acknowledges it using the XACK command.

If Instance A crashes while processing ID: 1, it will never send the XACK. Another consumer can inspect the PEL, notice that ID: 1 has been pending for too long, and claim ownership of it to reprocess it. This provides robust at-least-once delivery semantics.

Redis Streams vs. Kafka

Maya had to justify using Redis Streams over Kafka to the architecture review board.

For GlobalMart’s order processing, the events were small, and they didn’t need to replay orders from 6 months ago. They only needed a reliable buffer to decouple the microservices. Redis Streams, already running in their highly available cluster, was the perfect fit.

Technical Explanation

Maya architected the order flow:

1. Publish: The Checkout Service executes XADD orders * order_id 12345 amount 99.99. The * tells Redis to auto-generate the ID.
2. Group Creation: Maya pre-created two consumer groups on the orders stream: billing_group and inventory_group. Both groups read from the same stream, meaning every order goes to both Billing and Inventory independently.
3. Consume: The Billing Service instances execute XREADGROUP GROUP billing_group consumer_1 BLOCK 5000 STREAMS orders >. The > symbol is special; it means “give me new messages that have never been delivered to anyone in this group.”
4. Acknowledge: After successfully charging the credit card, the Billing Service executes XACK orders billing_group <message_id>.
5. Trim: To prevent the stream from consuming all of Redis’s RAM (Part 4), the Checkout Service periodically executes XTRIM orders MAXLEN ~ 1000000 to keep only the most recent 1 million orders.

Step-by-Step Hands-On

Let’s look at how Maya implemented the Billing Service consumer using Spring Data Redis.

Step 1: The Stream Listener

Spring Data Redis provides a powerful StreamMessageListenerContainer that acts similarly to a Kafka @KafkaListener or a JMS @JmsListener. It continuously polls the stream in a background thread and invokes your code when a message arrives.

Maya created a listener class implementing StreamListener.

import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Service;

@Service
public class OrderBillingListener implements StreamListener<String, MapRecord<String, String, String>> {

    private final RedisTemplate<String, String> redisTemplate;
    private static final String STREAM_KEY = "orders";
    private static final String GROUP_NAME = "billing_group";

    public OrderBillingListener(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        try {
            // 1. Extract the data
            String messageId = message.getId().getValue();
            String orderId = message.getValue().get("order_id");
            String amount = message.getValue().get("amount");

            System.out.println("Processing Order: " + orderId + " for $" + amount);

            // 2. Simulate business logic (charging credit card)
            processBilling(orderId, amount);

            // 3. Explicitly Acknowledge the message (XACK)
            // If this line is not reached (e.g., exception thrown or app crashes),
            // the message remains in the Pending Entries List (PEL) for reprocessing.
            redisTemplate.opsForStream().acknowledge(STREAM_KEY, GROUP_NAME, messageId);

            System.out.println("Successfully acknowledged message: " + messageId);

        } catch (Exception e) {
            // Log the error. Do NOT acknowledge.
            // A background task will claim and retry it later.
            System.err.println("Failed to process billing for message: " + message.getId());
        }
    }

    private void processBilling(String orderId, String amount) {
        // Complex third-party API call here...
    }
}

Step 2: Configuring the Listener Container

Maya then configured the Spring container to bind the listener to the specific stream and consumer group.

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.UUID;

@Configuration
public class StreamConfig {

    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(
            RedisConnectionFactory connectionFactory,
            OrderBillingListener orderBillingListener) {

        // 1. Configure the polling behavior
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                        .pollTimeout(Duration.ofMillis(100))
                        .build();

        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
                StreamMessageListenerContainer.create(connectionFactory, options);

        // 2. Define the Consumer Group and Consumer Name
        // In production, the consumer name should be unique per instance (e.g., hostname or UUID)
        String consumerName = "billing_instance_" + UUID.randomWindow();

        // 3. Bind the listener to the stream
        // ReadOffset.lastConsumed() is equivalent to the '>' symbol in XREADGROUP
        container.receive(
                Consumer.from("billing_group", consumerName),
                StreamOffset.create("orders", ReadOffset.lastConsumed()),
                orderBillingListener);

        // 4. Start the background polling thread
        container.start();

        return container;
    }
}

Debugging/Troubleshooting Tips

• The Poison Pill: If a message contains invalid data that consistently crashes the processBilling method, it will never be XACKed. It will remain in the PEL forever, and every time a consumer claims it, it will crash again. Solution: Implement a Dead Letter Queue (DLQ). If a message is claimed and fails more than $N$ times (check the delivery count in the PEL), move it to a separate orders_dlq stream and XACK it from the main stream so an engineer can inspect it manually.
• Memory Exhaustion: Streams are persistent. If you publish 10,000 messages a second and never trim the stream, Redis will eventually hit maxmemory and either OOM or start evicting your cached products (Part 4). Solution: Always use XADD ... MAXLEN ~ 1000000 to automatically cap the stream size. The ~ means “approximately,” allowing Redis to trim entire macro-nodes efficiently rather than exactly counting 1 million items.
• Group Creation: Spring Data Redis does not automatically create the stream or the consumer group. If you try to listen to a group that doesn’t exist, it throws an error. Solution: Write a startup script (using CommandLineRunner) that executes XGROUP CREATE orders billing_group 0 MKSTREAM before the listener container starts.

Key Takeaways

• Pub/Sub vs. Streams: Pub/Sub is ephemeral (fire and forget). Streams are durable, append-only logs.
• Consumer Groups: Allow multiple instances of a service to scale out message processing by automatically distributing the load and tracking state.
• At-Least-Once Delivery: The XACK command ensures messages are not lost if a consumer crashes. Unacknowledged messages remain in the PEL for reprocessing.
• Trimming: Always use MAXLEN when adding to a stream to prevent unbounded memory growth.

Story Closing + Teaser

The transition to an event-driven architecture using Redis Streams was a resounding success. The monolithic checkout process was decoupled into resilient microservices. If the Billing API went down for 10 minutes, the orders simply queued up in the Redis Stream. When Billing came back online, it processed the backlog instantly. No data was lost, and the customer experience was flawless.

GlobalMart’s backend was now a modern, distributed, highly available, and event-driven powerhouse.

But the e-commerce world was changing. The product team had a vision for the future: they didn’t just want users to search for “blue running shoes.” They wanted users to upload a photo of a shoe and find similar products. They wanted users to ask a chatbot, “I’m running a marathon in the rain next week, what should I wear?” and get a highly personalized, context-aware recommendation.

Traditional keyword search (SQL LIKE '%shoe%') was completely inadequate for this. Maya needed to bridge the gap between structured data and human semantics. She needed to understand vector embeddings, distance metrics, and how to turn her caching layer into a high-dimensional AI memory bank. The era of Redis Stack and Vector Search awaited in the next chapter.

References

[1] Redis Streams Tutorial. https://redis.io/docs/latest/develop/data-types/streams/ [2] Spring Data Redis: Redis Streams. https://docs.spring.io/spring-data/redis/reference/redis/redis-streams.html