Sequential Convoy Pattern

Overview

The Sequential Convoy Pattern preserves the order of related messages in a distributed messaging system. While many message brokers guarantee ordering within a single partition or queue, they typically process unrelated messages in parallel for throughput. When messages have a causal relationship — for example, “create user” followed by “update user” — processing them out of order leads to inconsistent state.

This pattern groups related messages into a convoy (a sequence) and ensures they are processed by a single consumer in order. Unrelated convoys can still be processed in parallel, preserving both correctness and throughput.

When to Use

• For alternatives, see Idempotent Consumer Pattern.

• 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 cannot begin until step N-1 completes

When to Avoid

• Messages have no causal relationship — parallel processing is simpler and faster
• Strict ordering is not 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

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: Messages for the same entity are routed to the same partition/queue/session. This is done using a partition key (Kafka), session ID (Service Bus), or entity field (Redis).

2. Single consumer per partition: Only one consumer processes messages from a given partition at a time. This prevents two consumers from handling different messages for the same entity simultaneously, which would violate ordering.

The trade-off is reduced parallelism per entity — all messages for user-123 must be processed sequentially. However, messages for user-456 can be processed in parallel on a different partition.

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

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.

Common Mistakes

• Changing partition keys. Rebalancing Kafka partitions changes which consumer handles which entity, violating ordering assumptions.
• Multiple consumers per partition. Two consumers reading the same partition will process messages in parallel for the same entity.
• Not handling sequence gaps. A lost message in a sequence blocks all subsequent messages forever.
• Overly large convoys. A convoy that processes thousands of messages for one entity creates 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. By using the user ID as the partition key, all events for a user are ordered. Uber uses this for trip events: trip-created, driver-assigned, trip-started, trip-completed must be processed in order for fare calculation.

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.