Redis Streams with Consumer Groups: Reliable Event Processing

1. Overview

Redis Pub/Sub has a major drawback. If a consumer is offline when a message is published, the message is lost. Redis Streams solves this problem. Messages are durably stored in the stream, and each consumer group tracks which messages have been fully processed.

This article guides you through building an order event processing pipeline with Spring Boot 3 and Spring Data Redis. The pipeline consists of: writing events, reading through consumer groups, manual ACK, reclaiming pending messages, and forwarding failed messages to a dead-letter stream.

2. How Redis Streams and Consumer Groups Work

Before writing Java code, you should try the raw commands in redis-cli:

XADD orders * orderId 1001 amount 250000
XGROUP CREATE orders shipping-group 0
XREADGROUP GROUP shipping-group worker-1 COUNT 1 STREAMS orders >
XPENDING orders shipping-group
XACK orders shipping-group 1728640000000-0

What each command does:

• XADD adds an entry to the orders stream. The * lets Redis auto-generate an ID in the form <millis>-<seq>.
• XGROUP CREATE ... 0 creates the shipping-group group and starts reading from the beginning of the stream. If you use $, the group only reads new messages.
• XREADGROUP ... > fetches messages not yet assigned to anyone in the group. Messages are assigned to worker-1 and simultaneously recorded in the group’s Pending Entries List (PEL).
• XPENDING shows how many messages have been delivered but not yet ACK’d.
• XACK removes the message from the PEL. From this point on, the group considers the message fully processed.

Output of XPENDING before ACK looks like this:

1) (integer) 1
2) "1728640000000-0"
3) "1728640000000-0"
4) 1) 1) "worker-1"
      2) "1"

The PEL is the key to reliability. Suppose a consumer crashes after receiving a message but before ACK’ing it. The message remains in the PEL, and another consumer can claim it for further processing. This is the at-least-once semantic.

3. Setup

Add the Spring Data Redis dependency. This starter uses Lettuce as the default client:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

Configure the connection and stream names:

spring:
  data:
    redis:
      host: localhost
      port: 6379

app:
  stream:
    key: orders
    group: shipping-group
    consumer: ${HOSTNAME:worker-local}

The consumer name is taken from HOSTNAME, so each instance (pod) has a unique name. This is important. If two instances share the same name, Redis treats them as one consumer, and their PELs become mixed together.

4. Writing Events to the Stream

@Service
public class OrderEventPublisher {

    private final StringRedisTemplate redisTemplate;
    private final String streamKey;

    public OrderEventPublisher(StringRedisTemplate redisTemplate,
                               @Value("${app.stream.key}") String streamKey) {
        this.redisTemplate = redisTemplate;
        this.streamKey = streamKey;
    }

    public RecordId publishOrderCreated(String orderId, long amount) {
        MapRecord<String, String, String> record = StreamRecords.newRecord()
                .in(streamKey)
                .ofMap(Map.of(
                        "type", "ORDER_CREATED",
                        "orderId", orderId,
                        "amount", String.valueOf(amount)));

        return redisTemplate.opsForStream().add(record);
    }
}

StreamRecords.newRecord().in(...).ofMap(...) creates a MapRecord with string-valued fields. The opsForStream().add() call is equivalent to XADD orders * ... and returns the RecordId generated by Redis, for example 1728640000000-0.

Using StringRedisTemplate makes the data in Redis readable via redis-cli. You don’t have to deal with JDK serialization byte strings.

5. Creating Consumer Groups Safely

The group must exist before reading. The application might start when the stream has no messages yet, so you need the MKSTREAM parameter. The createGroup() method of StreamOperations doesn’t enable this parameter. Therefore, we call down to the connection layer:

@Component
public class StreamGroupInitializer {

    private final StringRedisTemplate redisTemplate;

    public StreamGroupInitializer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void ensureGroup(String streamKey, String group) {
        try {
            redisTemplate.execute((RedisCallback<String>) connection ->
                    connection.streamCommands().xGroupCreate(
                            streamKey.getBytes(StandardCharsets.UTF_8),
                            group,
                            ReadOffset.from("0"),
                            true)); // MKSTREAM
        } catch (DataAccessException e) {
            String msg = NestedExceptionUtils.getMostSpecificCause(e).getMessage();
            if (msg == null || !msg.contains("BUSYGROUP")) {
                throw e;
            }
            // Group already exists: normal when restarting the application
        }
    }
}

The last parameter true enables MKSTREAM. Redis will create an empty stream if it doesn’t exist yet. When the application restarts, Redis returns the error BUSYGROUP Consumer Group name already exists. We ignore this specific error and rethrow any other error.

6. Reading Messages with StreamMessageListenerContainer

6.1. Listener with Manual ACK

@Component
public class OrderEventListener
        implements StreamListener<String, MapRecord<String, String, String>> {

    private final StringRedisTemplate redisTemplate;
    private final ShippingService shippingService;
    private final String streamKey;
    private final String group;

    public OrderEventListener(StringRedisTemplate redisTemplate,
                              ShippingService shippingService,
                              @Value("${app.stream.key}") String streamKey,
                              @Value("${app.stream.group}") String group) {
        this.redisTemplate = redisTemplate;
        this.shippingService = shippingService;
        this.streamKey = streamKey;
        this.group = group;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> record) {
        Map<String, String> body = record.getValue();
        shippingService.createShipment(body.get("orderId"));

        redisTemplate.opsForStream().acknowledge(streamKey, group, record.getId());
    }
}

The ACK command only runs if createShipment succeeds. If the method throws an exception, the message remains in the PEL for reprocessing later. Don’t ACK before processing. Doing so would turn at-least-once into at-most-once.

ShippingService in this article is a simple service. It tracks which orderIds have had shipments created:

@Service
public class ShippingService {

    private final Set<String> shippedOrders = ConcurrentHashMap.newKeySet();

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // already processed, skip (idempotent)
        }
        // call the shipping provider's API...
    }

    public Set<String> shippedOrders() {
        return Set.copyOf(shippedOrders);
    }
}

6.2. Container Configuration

@Configuration
public class StreamConfig {

    @Bean(destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>>
    orderStreamContainer(RedisConnectionFactory connectionFactory,
                         StreamGroupInitializer initializer,
                         OrderEventListener listener,
                         @Value("${app.stream.key}") String streamKey,
                         @Value("${app.stream.group}") String group,
                         @Value("${app.stream.consumer}") String consumerName) {

        initializer.ensureGroup(streamKey, group);

        var options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(2))
                .batchSize(10)
                .build();

        var container = StreamMessageListenerContainer.create(connectionFactory, options);

        var request = StreamMessageListenerContainer.StreamReadRequest
                .builder(StreamOffset.create(streamKey, ReadOffset.lastConsumed()))
                .consumer(Consumer.from(group, consumerName))
                .autoAcknowledge(false)
                .cancelOnError(t -> false)
                .build();

        container.register(request, listener);
        container.start();
        return container;
    }
}

Key points in the configuration:

• ReadOffset.lastConsumed() corresponds to > in XREADGROUP, meaning it only fetches new messages for the group.
• pollTimeout is the blocking time per read attempt. batchSize is the maximum number of messages in a single read.
• autoAcknowledge(false) disables auto-ACK because we manually ACK in the listener.
• cancelOnError(t -> false) is the most critical line. By default, when the listener throws an exception, the container cancels the subscription. The consumer will silently stop reading without a clear warning.

Note: Don’t use receiveAutoAck(...) for critical business logic. This mode ACK’s immediately upon reading, before the listener finishes. If the application crashes midway, the message is lost permanently.

7. Handling Pending Messages and Dead-Letter Streams

The listener only reads new messages (>). A message that errors or belongs to a dead consumer will remain in the PEL forever if no one handles it. We need a periodic job for this:

@Component
public class PendingOrderReclaimer {

    private static final Duration MIN_IDLE = Duration.ofMinutes(1);
    private static final long MAX_DELIVERIES = 5;

    private final StringRedisTemplate redisTemplate;
    private final ShippingService shippingService;
    private final String streamKey;
    private final String group;
    private final String consumerName;

    public PendingOrderReclaimer(StringRedisTemplate redisTemplate,
                                 ShippingService shippingService,
                                 @Value("${app.stream.key}") String streamKey,
                                 @Value("${app.stream.group}") String group,
                                 @Value("${app.stream.consumer}") String consumerName) {
        this.redisTemplate = redisTemplate;
        this.shippingService = shippingService;
        this.streamKey = streamKey;
        this.group = group;
        this.consumerName = consumerName;
    }

    @Scheduled(fixedDelay = 30_000)
    public void reclaim() {
        StreamOperations<String, String, String> ops = redisTemplate.opsForStream();
        PendingMessages pending = ops.pending(streamKey, group, Range.unbounded(), 50);

        for (PendingMessage pm : pending) {
            if (pm.getElapsedTimeSinceLastDelivery().compareTo(MIN_IDLE) < 0) {
                continue;
            }
            List<MapRecord<String, String, String>> claimed =
                    ops.claim(streamKey, group, consumerName, MIN_IDLE, pm.getId());

            for (MapRecord<String, String, String> record : claimed) {
                if (pm.getTotalDeliveryCount() >= MAX_DELIVERIES) {
                    moveToDeadLetter(record);
                } else {
                    shippingService.createShipment(record.getValue().get("orderId"));
                }
                ops.acknowledge(streamKey, group, record.getId());
            }
        }
    }

    private void moveToDeadLetter(MapRecord<String, String, String> record) {
        Map<String, String> body = new HashMap<>(record.getValue());
        body.put("originalId", record.getId().getValue());
        redisTemplate.opsForStream().add(
                StreamRecords.newRecord().in(streamKey + ":dlq").ofMap(body));
    }
}

The job’s processing flow:

1. pending(...) is equivalent to XPENDING orders shipping-group - + 50. Each PendingMessage tells you which consumer holds the message, its idle time, and how many times it’s been delivered.
2. claim(...) is equivalent to XCLAIM. This command transfers ownership to the current consumer. Redis only transfers if the message has been idle for at least MIN_IDLE, so two instances running the job simultaneously won’t claim the same message. Each claim also increments the delivery count.
3. Messages delivered more than MAX_DELIVERIES times are copied to the orders:dlq stream and then ACK’d. Permanently failing messages (poison messages) thus don’t get retried indefinitely.

If createShipment throws an exception, the ACK command doesn’t run. The message stays in the PEL and gets retried in the next cycle. Remember to enable @EnableScheduling on a @Configuration class.

Note: From Redis 6.2, there’s the XAUTOCLAIM command. This command combines XPENDING and XCLAIM into one step and is more efficient with large PELs. You can call it via redisTemplate.execute(...) if your current Spring Data Redis version doesn’t have a convenient API.

8. Writing Tests with Testcontainers

Add the test dependencies:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-testcontainers</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

The test below publishes an event. Then it verifies that the shipment was created and the PEL is empty:

@SpringBootTest
@Testcontainers
class OrderStreamIntegrationTest {

    @Container
    @ServiceConnection(name = "redis")
    static GenericContainer<?> redis =
            new GenericContainer<>("redis:7-alpine").withExposedPorts(6379);

    @Autowired OrderEventPublisher publisher;
    @Autowired ShippingService shippingService;
    @Autowired StringRedisTemplate redisTemplate;

    @Test
    void whenOrderPublished_thenShipmentCreatedAndMessageAcked() {
        publisher.publishOrderCreated("1001", 250_000);

        await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
            assertThat(shippingService.shippedOrders()).contains("1001");

            PendingMessagesSummary summary = redisTemplate.opsForStream()
                    .pending("orders", "shipping-group");
            assertThat(summary.getTotalPendingMessages()).isZero();
        });
    }
}

@ServiceConnection(name = "redis") lets Spring Boot automatically pick up the host and port from the container, so you don’t need @DynamicPropertySource. await() is necessary because the listener runs on a separate thread. An assertion that getTotalPendingMessages() equals 0 proves the message was ACK’d, not just read.

9. Common Pitfalls and When Not to Use

Non-idempotent processing. At-least-once means a message can arrive twice. For example: a consumer finishes processing but crashes before ACK’ing. Therefore, the business logic must tolerate retries. A common approach is to check if orderId was already processed or use a unique constraint in the database.

Stream grows unbounded. ACK doesn’t remove entries from the stream. You need to limit the size at write time:

redisTemplate.opsForStream().add(record, XAddOptions.maxlen(100_000).approximateTrimming(true));

This option is equivalent to XADD orders MAXLEN ~ 100000 * .... The ~ allows Redis to trim at internal node boundaries, much faster than exact trimming.

Forgetting cancelOnError. The first exception will silently stop the consumer. The symptom is messages backing up in the stream while logs show no clear errors.

Phantom consumers. Each time you deploy with a new consumer name (e.g., pod name), the old name lingers in the group. Clean up periodically with XGROUP DELCONSUMER, but only after claiming all pending messages from that consumer.

When NOT to use Redis Streams:

• You need to retain events for many days and replay large volumes. Redis keeps all data in RAM, so the cost is very high.
• You need absolute data loss guarantees. With AOF everysec or asynchronous replica failover, you can still lose a few recently written messages.
• Very high throughput requiring partitioning and scaling to hundreds of consumers. Kafka is a better fit in this scenario.

10. Conclusion

Redis Streams combined with consumer groups give you a lightweight and reliable event queue running on the Redis you already have. You must remember four key points:

• Manual ACK after processing finishes.
• Set cancelOnError(t -> false) so the consumer doesn’t silently stop.
• Have a job that claims pending messages, with a dead-letter stream for poison messages.
• Write idempotent processing logic.

Add MAXLEN to control memory and integration tests with Testcontainers, and you’ll have a solid at-least-once event processing pipeline for most Spring Boot applications.