Message Ordering

Global vs Partition Ordering

Partition Ordering

Partition 0:
[msg0] → [msg1] → [msg2] → [msg3]

Ordering guaranteed: msg0 before msg1 before msg2 before msg3

Within same partition = Ordered

Global Ordering

Partition 0: [msg0] → [msg2] → [msg4]
Partition 1: [msg1] → [msg3] → [msg5]

NO guarantee: msg1 may be processed before msg0

Across partitions = Unordered

Comparison

Type	Guarantee	Parallelism	Use Case
Partition	Within partition	Yes	Entity-level ordering
Global	All messages	No (single partition)	Strict sequence
None	No ordering	Maximum	Independent events

Achieving Global Order

Option 1: Single Partition
- All messages in one partition
- No parallelism
- Max throughput limited

Option 2: Sequence Numbers
- Include sequence in message
- Consumer reorders
- Complex consumer logic

Option 3: Single Producer
- One producer serializes
- Producer handles ordering
- Bottleneck at producer

Partition Ordering Use Cases

# User events - order matters per user
{
    "user_id": "123",  # Key
    "events": [
        {"type": "signup", "time": "10:00"},
        {"type": "purchase", "time": "10:05"},
        {"type": "refund", "time": "10:10"}
    ]
}
# All user events go to same partition
# Guarantees: signup before purchase before refund

Global Ordering Use Cases

# Financial transactions - strict ordering
{
    "account_id": "456",
    "transactions": [
        {"type": "deposit", "amount": 1000},
        {"type": "withdrawal", "amount": 500},
        {"type": "transfer", "amount": 200}
    ]
}
# May need global order for audit/compliance

Ordering Tradeoffs

Tradeoff Matrix

Requirement	Parallelism	Throughput	Complexity
Global Order	Low	Low	High
Partition Order	Medium	Medium	Medium
No Order	High	High	Low

Impact on System Design

Global Ordering:
- Single partition = bottleneck
- Producer must serialize
- Consumer must reorder
- Latency: Higher
- Throughput: Limited

Partition Ordering:
- Multiple partitions = parallelism
- Key-based routing
- Consumer per partition
- Latency: Lower
- Throughput: Scales with partitions

No Ordering:
- Maximum parallelism
- Simple implementation
- Best throughput
- May need application-level ordering

Design Decisions

# Decision framework
ordering_requirements = {
    'global_strict': {
        'solution': 'single_partition',
        'parallelism': 'none',
        'throughput': 'limited'
    },
    'entity_level': {
        'solution': 'key_by_entity',
        'parallelism': 'per_entity',
        'throughput': 'good'
    },
    'no_ordering': {
        'solution': 'round_robin',
        'parallelism': 'maximum',
        'throughput': 'best'
    }
}

Common Patterns

1. Entity-level ordering: Key by entity ID
2. Causal ordering: Dependency chain in messages
3. Temporal ordering: Timestamp-based (but not guaranteed)
4. Priority ordering: Priority queue per partition

Ordering Implementation

Partition-Level Ordering

// Producer: Key-based partitioning
ProducerRecord<String, String> record = new ProducerRecord<>(
    "user-events",
    userId,  // Key ensures same partition
    eventJson
);

// Consumer: One consumer per partition
consumer.subscribe(Arrays.asList("user-events"));
// Partitions assigned to consumers
// Each consumer processes ordered stream

Global Ordering with Sequence Numbers

class GlobalOrderProducer:
    def __init__(self, kafka_producer):
        self.producer = kafka_producer
        self.sequence = 0
        self.lock = threading.Lock()
    
    def send(self, topic, value):
        with self.lock:
            self.sequence += 1
            record = {
                'sequence': self.sequence,
                'timestamp': time.time(),
                'value': value
            }
            self.producer.send(topic, value=json.dumps(record))

# Consumer reorders
class GlobalOrderConsumer:
    def __init__(self):
        self.expected_sequence = 0
        self.buffer = {}
    
    def process(self, message):
        seq = message['sequence']
        
        if seq == self.expected_sequence:
            self.handle(message)
            self.expected_sequence += 1
            # Process buffered messages
            while self.expected_sequence in self.buffer:
                self.handle(self.buffer.pop(self.expected_sequence))
                self.expected_sequence += 1
        else:
            self.buffer[seq] = message  # Buffer out-of-order

Idempotent Consumer

class IdempotentConsumer:
    def __init__(self, db):
        self.db = db
    
    def process(self, message):
        # Check if already processed
        if self.db.is_processed(message['id']):
            return  # Skip duplicate
        
        # Process and mark as processed
        with self.db.transaction():
            self.handle(message)
            self.db.mark_processed(message['id'])

Best Practices

1. Use entity-level ordering when possible
2. Implement idempotent consumers for retry safety
3. Document ordering guarantees clearly
4. Test with out-of-order scenarios
5. Monitor processing order metrics