Patrón de Convoy Secuencial

Resumen

El Patrón de Convoy Secuencial preserva el orden de los mensajes relacionados en un sistema de mensajería distribuido. Muchos brokers garantizan orden dentro de una partición o cola, pero procesan mensajes no relacionados en paralelo para ganar rendimiento. Cuando los mensajes tienen una relación causal — por ejemplo, “crear usuario” seguido de “actualizar usuario” — procesarlos fuera de orden produce estado inconsistente.

Este patrón agrupa los mensajes relacionados en un convoy (una secuencia) y enruta cada convoy a un único consumidor que lo procesa en orden. Los convoys no relacionados siguen procesándose en paralelo, así conservas corrección y throughput.

Cuándo usarlo

• Los mensajes de la misma entidad deben procesarse en el orden en que se produjeron
• Event sourcing donde los eventos de un agregado deben aplicarse secuencialmente
• Pipelines de pedidos donde las transiciones de estado dependen de estados previos
• Sistemas de inventario donde los movimientos de stock deben aplicarse cronológicamente
• Workflows multi-paso donde el paso N no puede empezar hasta que el paso N-1 termine

Cuándo evitarlo

• Los mensajes no tienen relación causal — el procesamiento paralelo es más simple y rápido
• El orden estricto no es necesario (ej. eventos de analytics independientes)
• El sistema tolera consistencia eventual sin garantías de orden
• El volumen por convoy es tan alto que un solo consumidor crea un cuello de botella
• Necesitas sobrevivir a entregas duplicadas más que al orden — ese es el trabajo del Patrón de Consumidor Idempotente
• Los productores con ráfagas necesitan buffer más que orden — mira Queue-Based Load Leveling

Solución

Python (Kafka con Partition Key)

from kafka import KafkaProducer, KafkaConsumer
import json
import time

class OrderedMessageProducer:
    """Produce mensajes que mantienen ordenamiento por entidad"""

    def __init__(self, bootstrap_servers):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            # Particionar por ID de entidad asegura que todos los mensajes
            # de la misma entidad caigan en la misma partición, en orden
            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):
        """Envía el evento con entity_id como clave de partición"""
        message = {
            'entity_id': entity_id,
            'event_type': event_type,
            'payload': payload,
            'timestamp': time.time(),
            'sequence_number': self._get_next_sequence(entity_id)
        }
        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:
        # En producción usa un contador en base de datos o Redis;
        # esto es un ejemplo simplificado
        return int(time.time() * 1000)

class SequentialConvoyConsumer:
    """Procesa mensajes en orden por entidad"""

    def __init__(self, bootstrap_servers, topic, group_id):
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            auto_offset_reset='earliest',
            # Crítico: máximo un consumidor por partición para preservar el orden
            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']

            # Procesamiento estrictamente en orden
            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:
                # Fuera de orden: se bufferea hasta que lleguen los que faltan
                self.pending_sequences.setdefault(entity_id, {})[seq_num] = event
                print(f"Buffering mensaje fuera de orden {seq_num} para {entity_id}")
            else:
                # Duplicado o ya procesado
                print(f"Saltando duplicado {seq_num} para {entity_id}")

            self.consumer.commit()

    def _process_event(self, event):
        print(f"Procesando {event['event_type']} para {event['entity_id']}")
        # Lógica de negocio aquí

    def _check_pending(self, entity_id):
        """Procesa los mensajes buffereados que ya están en orden"""
        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

# Uso
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";

    // Productor: usa session ID para agrupar los mensajes ordenados
    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))
                );
                // El session ID asegura que los mensajes con el mismo ID
                // los procese el mismo consumidor, en orden
                message.setSessionId(entityId);
                message.setApplicationProperty("sequenceNumber", i);

                sender.sendMessage(message);
            }
        }
    }

    // Consumidor: procesa una sesión a la vez, preservando el orden
    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 1 garantiza orden estricto dentro de la sesión
                .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");

            // Service Bus garantiza el orden porque solo un procesador
            // atiende una sesión a la vez
            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 con 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) {
        // INCR por entidad = número de secuencia monótono y detectable
        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);

        // Verificación de orden
        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) {
            // Llegó antes de tiempo: se reintentará cuando la secuencia se complete
            console.log(`Fuera de orden: esperado ${expected}, recibido ${sequence}`);
            return null;
        } else {
            // Duplicado
            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(`Procesando ${event.eventType} para ${event.entityId}`);
        // Lógica de negocio
    }
}

Explicación

El patrón se apoya en dos mecanismos:

1. Particionado por ID de entidad: los mensajes de la misma entidad se enrutan a la misma partición, cola o sesión — con una partition key (Kafka), un session ID (Service Bus) o un campo de entidad (Redis).

2. Un solo consumidor por partición: una partición es atendida por un único consumidor a la vez, así que nadie más puede tomar el siguiente mensaje de la misma entidad en paralelo.

El coste es menos paralelismo por entidad: todos los mensajes de user-123 se procesan uno tras otro. A cambio, los de user-456 van en paralelo por otra partición.

Dentro de cada partición el convoy marcha en fila india; entre particiones, los convoys corren en paralelo.

Variantes

Variante	Mecanismo	Ideal para
Partition key de Kafka	Asignación de partición por hash	Alto throughput, orden simple
Sesiones de Service Bus	Balanceo consciente de sesiones	Nativo cloud, exactly-once por sesión
Single active consumer de RabbitMQ	Consumidor exclusivo por cola	Orden simple basado en cola
Tabla de secuencia en base de datos	Bloqueo optimista sobre números de secuencia	Sistemas sin ordering del broker
Sagas con orquestación	Orden explícito de pasos en un motor de workflows	Procesos de negocio multi-paso complejos

Si los mensajes son demasiado grandes para el broker, combina este patrón con Claim Check: ordena una referencia ligera y deja que el consumidor descargue el payload.

Lo que funciona

• Clave de partición determinista. El ID de entidad debe mapear siempre a la misma partición; cambiar la clave invalida el orden.
• Vigila el skew de particiones. Si una entidad genera el 90 % de los mensajes, su partición se convierte en cuello de botella. Divide las entidades calientes.
• Gestiona los mensajes perdidos. Si la secuencia N nunca llega, el convoy se frena. Implementa timeouts y alertas.
• Convoys cortos. Las secuencias largas retienen mensajes nuevos; diseña para convoys acotados.
• Idempotencia dentro del convoy. Incluso con orden, los reintentos producen duplicados; haz cada operación idempotente.
• Agrupa por partición, no por poll. Un poll() de Kafka puede devolver registros de varias particiones; agrúpalos por partición y aplica cada grupo en orden — hacer batch entre entidades rompe el orden.

Errores comunes

• Cambiar claves de partición. Rebalancear particiones de Kafka cambia qué consumidor atiende cada entidad y rompe los supuestos de orden.
• Dos consumidores en una partición. Dos consumidores leyendo la misma partición procesan en paralelo mensajes de la misma entidad — la garantía de orden desaparece.
• No manejar los gaps de secuencia. Un mensaje perdido bloquea todos los mensajes siguientes para siempre.
• Convoys gigantes. Un convoy de miles de mensajes para una entidad crea un hotspot.
• Ignorar los reintentos del productor. Un mensaje reintentado puede quedar reordenado respecto a uno más nuevo si no comparten clave de partición.

Ejemplos del mundo real

Particionado de Kafka

Kafka garantiza orden dentro de una partición, así que usando el ID de usuario como partition key cada usuario mantiene su historia ordenada. Una app de ride-sharing, por ejemplo, puede usar el trip ID como clave: trip-created, driver-assigned, trip-started, trip-completed caen en la misma partición y el cálculo de tarifa los ve siempre en secuencia.

Sesiones de Azure Service Bus

Las sesiones de Service Bus dan orden FIFO dentro de una sesión. Una plataforma de e-commerce usa sesiones por carrito: item-added, quantity-changed, checkout-initiated, payment-received se procesan en secuencia para mantener el carrito consistente.

Event Store DB

Event Store DB usa control de concurrencia optimista sobre streams. Cada agregado (por ejemplo, un pedido) es un stream y los eventos se agregan con números de versión esperados; los escritores concurrentes fallan si el stream cambió, preservando el orden.

Código companion: el repo companion de sequential-convoy-pattern empaqueta las tres implementaciones como proyectos ejecutables.