Sequential Convoy Pattern

Overview

The Sequential Convoy Pattern keeps related messages in order inside a distributed messaging system. Most brokers already order messages within one partition or queue, but they fan unrelated messages out across consumers for throughput. That works until messages depend on each other: process “update user” before “create user” and the state is wrong.

This pattern groups related messages into a convoy (a sequence) and routes each convoy to a single consumer that processes it in order. Unrelated convoys still run in parallel, so you keep both correctness and throughput.

When to Use

• Messages for the same entity must be processed in the order they were produced
• Event sourcing where events for an aggregate must be applied sequentially
• Order processing pipelines where status transitions depend on prior states
• Inventory systems where stock movements must be applied chronologically
• Multi-step workflows where step N can’t begin until step N-1 completes

When to Avoid

• Messages have no causal relationship — parallel processing is simpler and faster
• Strict ordering isn’t required (e.g., independent analytics events)
• The system can tolerate eventual consistency without ordering guarantees
• Message volumes per convoy are so high that single-consumer processing creates a bottleneck
• You need to survive duplicate deliveries rather than ordering — that’s the Idempotent Consumer Pattern’s job
• Bursty producers need buffering more than ordering — see Queue-Based Load Leveling

Solution

Python (Kafka with Partition Key)

from kafka import KafkaProducer, KafkaConsumer
from kafka.partitioner import DefaultPartitioner
import json
import time

class OrderedMessageProducer:
    """Produce messages that maintain ordering per entity"""

    def __init__(self, bootstrap_servers):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            # Partition by entity ID ensures all messages for the same entity
            # go to the same partition, preserving order
            partitioner=lambda key, partitions, topic: (
                hash(key) % len(partitions) if key else 0
            )
        )

    def send_event(self, entity_id: str, event_type: str, payload: dict):
        """Send event with entity_id as partition key"""
        message = {
            'entity_id': entity_id,
            'event_type': event_type,
            'payload': payload,
            'timestamp': time.time(),
            'sequence_number': self._get_next_sequence(entity_id)
        }
        # Using entity_id as key ensures all events for this entity
        # land in the same partition, preserving order
        self.producer.send(
            'entity-events',
            key=entity_id.encode('utf-8'),
            value=message
        )
        self.producer.flush()

    def _get_next_sequence(self, entity_id: str) -> int:
        # In production, use a database or Redis counter
        # This is a simplified example
        return int(time.time() * 1000)

class SequentialConvoyConsumer:
    """Process messages in order per entity"""

    def __init__(self, bootstrap_servers, topic, group_id):
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            auto_offset_reset='earliest',
            # Critical: max one consumer per partition to preserve order
            max_poll_records=1,
            enable_auto_commit=False
        )
        self.pending_sequences: dict = {}
        self.last_processed: dict = {}

    def process_messages(self):
        for message in self.consumer:
            event = json.loads(message.value)
            entity_id = event['entity_id']
            seq_num = event['sequence_number']

            # Ensure in-order processing
            expected_seq = self.last_processed.get(entity_id, 0) + 1

            if seq_num == expected_seq:
                self._process_event(event)
                self.last_processed[entity_id] = seq_num
                self._check_pending(entity_id)
            elif seq_num > expected_seq:
                # Out of order — buffer and wait for missing messages
                self.pending_sequences.setdefault(entity_id, {})[seq_num] = event
                print(f"Buffering out-of-order message {seq_num} for {entity_id}")
            else:
                # Duplicate or already processed
                print(f"Skipping duplicate/late message {seq_num} for {entity_id}")

            self.consumer.commit()

    def _process_event(self, event):
        print(f"Processing {event['event_type']} for {event['entity_id']}")
        # Business logic here

    def _check_pending(self, entity_id):
        """Process any buffered messages that are now in order"""
        pending = self.pending_sequences.get(entity_id, {})
        expected = self.last_processed.get(entity_id, 0) + 1

        while expected in pending:
            event = pending.pop(expected)
            self._process_event(event)
            self.last_processed[entity_id] = expected
            expected += 1

# Usage
producer = OrderedMessageProducer(['localhost:9092'])
producer.send_event('user-123', 'created', {'name': 'Alice'})
producer.send_event('user-123', 'updated', {'name': 'Alice Smith'})
producer.send_event('user-123', 'deleted', {})

consumer = SequentialConvoyConsumer(['localhost:9092'], 'entity-events', 'convoy-group')
consumer.process_messages()

Java (Azure Service Bus Sessions)

import com.azure.messaging.servicebus.ServiceBusClientBuilder;
import com.azure.messaging.servicebus.ServiceBusMessage;
import com.azure.messaging.servicebus.ServiceBusSenderClient;
import com.azure.messaging.servicebus.ServiceBusProcessorClient;
import com.azure.messaging.servicebus.models.ServiceBusReceiveMode;

import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

public class SequentialConvoyServiceBus {

    private static final String CONNECTION_STRING = "<connection-string>";
    private static final String QUEUE_NAME = "ordered-queue";

    // Producer: uses session ID to group ordered messages
    public static class OrderedProducer {
        private final ServiceBusSenderClient sender;

        public OrderedProducer() {
            this.sender = new ServiceBusClientBuilder()
                .connectionString(CONNECTION_STRING)
                .sender()
                .queueName(QUEUE_NAME)
                .buildClient();
        }

        public void sendOrderedEvents(String entityId, List<DomainEvent> events) {
            for (int i = 0; i < events.size(); i++) {
                ServiceBusMessage message = new ServiceBusMessage(
                    serializeEvent(events.get(i))
                );
                // Session ID ensures all messages with the same ID
                // are processed by the same consumer in order
                message.setSessionId(entityId);
                message.setApplicationProperty("sequenceNumber", i);

                sender.sendMessage(message);
            }
        }
    }

    // Consumer: processes one session at a time, preserving order
    public static class OrderedConsumer {

        public void startProcessing() {
            ServiceBusProcessorClient processor = new ServiceBusClientBuilder()
                .connectionString(CONNECTION_STRING)
                .processor()
                .queueName(QUEUE_NAME)
                .receiveMode(ServiceBusReceiveMode.PEEK_LOCK)
                .processMessage(this::processMessage)
                .processError(this::handleError)
                // Prefetch count 1 ensures strict ordering within session
                .prefetchCount(1)
                .buildProcessorClient();

            processor.start();
        }

        private void processMessage(ServiceBusReceivedMessageContext context) {
            ServiceBusReceivedMessage message = context.getMessage();
            String sessionId = message.getSessionId();
            int sequenceNumber = (int) message.getApplicationProperties()
                .get("sequenceNumber");

            // Process in order — Service Bus guarantees this
            // because only one processor handles a session at a time
            DomainEvent event = deserializeEvent(message.getBody().toString());
            applyEvent(sessionId, event);

            context.complete();
        }

        private void handleError(ServiceBusErrorContext context) {
            System.err.println("Error: " + context.getException().getMessage());
        }
    }
}

JavaScript (Redis Streams with Consumer Groups)

const Redis = require('ioredis');

class SequentialConvoyProcessor {
    constructor(redis, streamKey) {
        this.redis = redis;
        this.streamKey = streamKey;
        this.groupName = 'convoy-processors';
    }

    async initialize() {
        try {
            await this.redis.xgroup('CREATE', this.streamKey,
                this.groupName, '0', 'MKSTREAM');
        } catch (err) {
            if (!err.message.includes('already exists')) throw err;
        }
    }

    async produceEvent(entityId, eventType, payload) {
        const sequence = await this.redis.incr(`seq:${entityId}`);

        await this.redis.xadd(this.streamKey, '*',
            'entityId', entityId,
            'sequence', sequence.toString(),
            'eventType', eventType,
            'payload', JSON.stringify(payload)
        );
    }

    async consumeOrdered(consumerId) {
        const results = await this.redis.xreadgroup(
            'GROUP', this.groupName, consumerId,
            'COUNT', 1,
            'BLOCK', 5000,
            'STREAMS', this.streamKey, '>'
        );

        if (!results || results.length === 0) return null;

        const [[, messages]] = results;
        const [id, fields] = messages[0];

        const event = this.parseFields(fields);
        const entityId = event.entityId;
        const sequence = parseInt(event.sequence);

        // Check ordering
        const lastProcessed = await this.redis.get(`last:${entityId}`);
        const expected = lastProcessed ? parseInt(lastProcessed) + 1 : 1;

        if (sequence === expected) {
            await this.processEvent(event);
            await this.redis.set(`last:${entityId}`, sequence);
            await this.redis.xack(this.streamKey, this.groupName, id);
            return event;
        } else if (sequence > expected) {
            // Requeue — will be retried when sequence is complete
            console.log(`Out of order: expected ${expected}, got ${sequence}`);
            return null;
        } else {
            // Duplicate
            await this.redis.xack(this.streamKey, this.groupName, id);
            return null;
        }
    }

    parseFields(fields) {
        const obj = {};
        for (let i = 0; i < fields.length; i += 2) {
            obj[fields[i]] = fields[i + 1];
        }
        return obj;
    }

    async processEvent(event) {
        console.log(`Processing ${event.eventType} for ${event.entityId}`);
        // Business logic
    }
}

Explanation

The pattern relies on two key mechanisms:

1. Partitioning by entity ID: Everything for the same entity lands on the same partition/queue/session — via a partition key (Kafka), a session ID (Service Bus), or an entity field (Redis).

2. Single consumer per partition: One consumer owns a partition at a time. Nobody else can pull the next message for the same entity in parallel — that’s exactly what would break the ordering.

The cost: user-123’s messages go one by one. Meanwhile user-456 flows through its own partition at full speed.

Inside a partition the convoy marches single file; across partitions, convoys run side by side.

Variants

Variant	Mechanism	Best For
Kafka partition key	Hash-based partition assignment	High throughput, simple ordering
Service Bus sessions	Session-aware load balancing	Cloud-native, exactly-once per session
RabbitMQ single active consumer	Exclusive consumer per queue	Simple queue-based ordering
Database sequence table	Optimistic locking on sequence numbers	Systems without message broker ordering
Sagas with orchestration	Explicit step ordering in workflow engine	Complex multi-step business processes

When the messages are too large to carry through the broker, combine this pattern with Claim Check: order a lightweight reference and let consumers fetch the payload.

What Works

• Use a deterministic partition key. The entity ID should consistently map to the same partition. Changing the key invalidates ordering.
• Monitor partition skew. If one entity generates 90% of messages, its partition becomes a bottleneck. Consider splitting hot entities.
• Handle missing messages gracefully. If sequence N never arrives, the convoy stalls. Implement timeouts and alerts.
• Keep convoys small. Long-running convoys hold up new messages. Design for short, bounded sequences.
• Idempotent processing within convoys. Even with ordering, retries can cause duplicates. Make individual operations idempotent.
• Batch per partition, not per poll. A Kafka poll() can return records for several partitions; group them by partition and apply each group in sequence order — batching across entities breaks ordering.

Common Mistakes

• Changing partition keys. Rebalancing Kafka partitions changes which consumer handles which entity, violating ordering assumptions.
• Two consumers on one partition. Let a second consumer join the same partition and both end up racing through an entity’s messages — ordering gone.
• Not handling sequence gaps. A lost message in a sequence blocks all subsequent messages forever.
• Overly large convoys. Thousands of queued messages for one entity turns its partition into a hotspot.
• Ignoring producer retries. A retried message may be reordered relative to a newer message if not using the same partition key.

Real-World Examples

Kafka Partitioning

Kafka guarantees order within a partition, so keying events by user ID keeps every user’s history ordered. A ride-sharing app, for instance, can key trip events by trip ID: trip-created, driver-assigned, trip-started, trip-completed land in the same partition and fare calculation always sees them in sequence.

Azure Service Bus Sessions

Service Bus sessions provide FIFO ordering within a session. An e-commerce platform uses sessions per shopping cart: item-added, quantity-changed, checkout-initiated, payment-received must be processed sequentially to maintain cart consistency.

Event Store DB

Event Store DB uses optimistic concurrency control on streams. Each aggregate (e.g., an order) is a stream, and events are appended with expected version numbers. Concurrent writers fail if the stream has been modified, preserving ordering.

Companion code: the sequential-convoy-pattern companion repo packages all three implementations as runnable projects.