Causal Consistency in Distributed Event Streams: Building Idempotent Consumers with Vector Clocks in Rust and Apache Kafka
Causal Consistency in Distributed Event Streams: Building Idempotent Consumers with Vector Clocks in Rust and Apache Kafka
The Problem Nobody Talks About
You've built your microservices. Events flow through Kafka. Consumers process messages idempotently. You're comfortable. Then one day, a user reports that their order was created twice — or worse, that a refund arrived before the original charge event. Your system is technically correct — exactly-once delivery per partition — but it's causally wrong. This is the silent plague of distributed systems that operate over partitioned, asynchronous event streams. Most teams solve for at-least-once or exactly-once semantics and call it a day. But when the causal ordering of events across partitions, consumers, and services breaks down, you get phantom state transitions that are nearly impossible to trace and even harder to reproduce.
In this post, I'll walk through building a causal consistency layer for Kafka consumers in Rust, using vector clocks to preserve happens-before relationships across a multi-region event streaming topology.
Why "Exactly-One" Isn't Enough
Let's set up the scenario:
User Service ──► Kafka (Partition 0) ──► Order Service ──► Kafka (Partition 3)
│ │ └──► Notification Service ◄───────────────────┘
Two events:
- E1:
OrderCreated { order_id: "abc", total: 49.99 }— produced to Partition 0 at t=100ms - E2:
PaymentConfirmed { order_id: "abc", amount: 49.99 }— produced to Partition 3 at t=105ms
In a single-broker, single-partition world, Kafka guarantees offset ordering. E1 arrives before E2. No problem. But in a multi-broker, multi-partition layout with producers writing concurrently, and consumers reading from different partitions on different threads or instances, Kafka makes no guarantees about cross-partition ordering. E2 could be processed by the Order Service's consumer before E1 is even read.
Traditional solutions:
- Single partition per entity: Works for simple cases but kills throughput and creates hot partitions.
- Timestamp-based ordering: Fails under clock skew.
- Two-phase commit: Adds latency and complexity that doesn't scale.
None of these solve the fundamental problem: you need a logical clock that captures the happens-before relation across decentralized producers and consumers.
Vector Clocks: The Right Abstraction
A Lamport clock gives you a total ordering but is too coarse — it inflates the "happened-before" relation, marking concurrent events as causally related. A vector clock gives you a partial order that correctly identifies true concurrency. A vector clock is a map from node ID to logical timestamp:
VC = { node_A: 5, node_B: 3, node_C: 7 }
When a node produces an event, it increments its own counter. When a node receives an event, it takes the element-wise maximum of its current clock and the incoming clock, then increments its own counter.
Comparison of two vector clocks V1 and V2:
- V1 ≤ V2 (V1 happened before V2) if for all i, V1[i] ≤ V2[i], and at least one strict inequality holds.
- V1 ∥ V2 (concurrent) if neither V1 ≤ V2 nor V2 ≤ V1.
The Rust Implementation
1. Vector Clock Data Structure
use std::collections::HashMap;
use std::cmp;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VectorClock {
counters: HashMap<String, u64>,
}
impl VectorClock {
pub fn new() -> Self {
VectorClock { counters: HashMap::new() }
}
pub fn tick(&mut self, node_id: &str) {
let counter = self.counters.entry(node_id.to_string()).or_insert(0);
*counter += 1;
}
pub fn merge(&mut self, other: &VectorClock) {
for (node, ×tamp) in &other.counters {
let entry = self.counters.entry(node.clone()).or_insert(0);
*entry = cmp::max(*entry, timestamp);
}
}
pub fn compare(&self, other: &VectorClock) -> Option<std::cmp::Ordering> {
let mut self_less_or_equal = true;
let mut other_less_or_equal = true;
let all_nodes: std::collections::HashSet<_> = self.counters.keys()
.chain(other.counters.keys())
.cloned()
.collect();
for node in all_nodes {
let a = *self.counters.get(node).unwrap_or(&0);
let b = *other.counters.get(node).unwrap_or(&0);
if a > b {
other_less_or_equal = false;
}
if b > a {
self_less_or_equal = false;
}
}
if self_less_or_equal && other_less_or_equal {
Some(std::cmp::Ordering::Equal)
} else if self_less_or_equal {
Some(std::cmp::Ordering::Less)
} else if other_less_or_equal {
Some(std::cmp::Ordering::Greater)
} else {
None // Concurrent
}
}
}
2. Envelope Wrapper for Kafka Messages
use serde::{Serialize, Deserialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CausalEnvelope<T> {
pub event_id: String,
pub producer_node: String,
pub vector_clock: HashMap<String, u64>,
pub payload: T,
pub timestamp_ns: u128,
}
impl<T> CausalEnvelope<T> {
pub fn new(producer_node: String, payload: T, event_id: String) -> Self {
CausalEnvelope {
event_id,
producer_node,
vector_clock: HashMap::new(),
payload,
timestamp_ns: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos(),
}
}
}
3. The Causal Kafka Consumer
use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::message::BorrowedMessage;
use futures::stream::StreamExt;
use serde_json;
use std::collections::{HashMap, BTreeMap};
use tokio::sync::RwLock;
use std::sync::Arc;
struct CausalConsumer {
consumer: StreamConsumer,
node_id: String,
entity_clocks: Arc<RwLock<HashMap<String, VectorClock>>>,
pending_buffer: Arc<RwLock<BTreeMap<String, Vec<CausalEnvelope<serde_json::Value>>>>>,
}
impl CausalConsumer {
pub fn new(brokers: &str, group_id: &str, node_id: String) -> Self {
let consumer: StreamConsumer = ClientConfig::new()
.set("group.id", group_id)
.set("bootstrap.servers", brokers)
.set("enable.auto.commit", "false")
.set("auto.offset.reset", "earliest")
.create()
.expect("Consumer creation failed");
CausalConsumer {
consumer,
node_id,
entity_clocks: Arc::new(RwLock::new(HashMap::new())),
pending_buffer: Arc::new(RwLock::new(BTreeMap::new())),
}
}
async fn process_message(&self, msg: BorrowedMessage<'_>) -> Result<(), Box<dyn std::error::Error>> {
let payload: CausalEnvelope<serde_json::Value> = serde_json::from_slice(msg.payload().ok_or("Empty payload")?)?;
let entity_key = self.extract_entity_key(&payload.payload);
let mut entity_clock = {
self.entity_clocks.read().await
.get(&entity_key)
.cloned()
.unwrap_or_else(VectorClock::new)
};
let incoming_clock = VectorClock { counters: payload.vector_clock.clone() };
match entity_clock.compare(&incoming_clock) {
Some(std::cmp::Ordering::Greater) | Some(std::cmp::Ordering::Equal) => {
self.handle_event(&payload).await?;
let mut committed = entity_clock;
committed.merge(&incoming_clock);
committed.tick(&self.node_id);
self.entity_clocks.write().await.insert(entity_key, committed);
self.unblock_pending(&entity_key).await;
},
None => {
self.pending_buffer.write().await
.entry(entity_key)
.or_default()
.push(payload);
},
Some(std::cmp::Ordering::Less) => {
tracing::info!(
event_id = %payload.event_id,
"Duplicate or stale event detected, skipping"
);
}
}
let offset = msg.offset();
let partition = msg.partition();
self.consumer.store_offset((msg.topic(), partition, offset))?;
Ok(())
}
async fn unblock_pending(&self, entity_key: &str) {
let mut buffer = self.pending_buffer.write().await;
if let Some(pending) = buffer.get_mut(entity_key) {
pending.retain(|envelope| {
let entity_clock = self.entity_clocks.read().await
.get(entity_key)
.cloned()
.unwrap_or_else(VectorClock::new);
let incoming = VectorClock { counters: envelope.vector_clock.clone() };
match entity_clock.compare(&incoming) {
Some(ord) if ord == std::cmp::Ordering::Less => false,
Some(_) => {
tokio::spawn(async move {
tracing::info!(event_id = %event.event_id, "Unblocked and processing");
});
false
}
None => true,
}
});
if pending.is_empty() {
buffer.remove(entity_key);
}
}
}
fn extract_entity_key(&self, payload: &serde_json::Value) -> String {
payload.get("order_id")
.and_then(|v| v.as_str())
.unwrap_or("unknown")
.to_string()
}
async fn handle_event(&self, envelope: &CausalEnvelope<serde_json::Value>) {
tracing::info!(
event_id = %envelope.event_id,
producer = %envelope.producer_node,
"Processing causally ordered event"
);
// ... business logic here ...
}
}
4. The Producer Side: Thread-Safe Clock Management
use rdkafka::producer::{FutureProducer, FutureRecord, OwnedDeliveryStatus};
struct CausalProducer {
producer: FutureProducer,
node_id: String,
clock: Arc<RwLock<VectorClock>>,
}
impl CausalProducer {
pub fn new(brokers: &str, node_id: String) -> Self {
let producer: FutureProducer = ClientConfig::new()
.set("bootstrap.servers", brokers)
.set("acks", "all")
.set("enable.idempotence", "true")
.create()
.expect("Producer creation failed");
CausalProducer {
producer,
node_id,
clock: Arc::new(RwLock::new(VectorClock::new())),
}
}
pub async fn produce<T: Serialize>(&self, topic: &str, key: &str, payload: &T) -> Result<(), Box<dyn std::error::Error>> {
let mut current_clock = self.clock.write().await.clone();
current_clock.tick(&self.node_id);
let new_clock = current_clock.clone();
let envelope = CausalEnvelope::new(
self.node_id.clone(),
serde_json::to_value(payload)?,
uuid::Uuid::new_v4().to_string(),
);
envelope.vector_clock = new_clock.counters.clone();
let key_bytes = key.as_bytes();
let payload_bytes = serde_json::to_vec(&envelope)?;
let record = FutureRecord::to(topic)
.key(key_bytes)
.payload(&payload_bytes);
match self.producer.send(record, std::time::Duration::from_secs(5)).await {
Ok(Ok(OwnedDeliveryStatus { .. })) => {
*self.clock.write().await = new_clock;
tracing::info!(
event_id = %envelope.event_id,
"Produced with causal clock"
);
Ok(())
},
Err((e, _)) => Err(Box::new(e)),
}
}
}
Handling the Edge Cases That Will Haunt You
Concurrent Events Are Not Errors — They're Information
When two events are concurrent (V1 ∥ V2), your system has a choice:
- Buffer indefinitely — dangerous under load. Memory grows, events starve.
- Apply a deterministic tiebreaker — use the wall-clock timestamp (with NTP-synchronized clocks) as a secondary sort key.
fn resolve_concurrency(a: &CausalEnvelope<serde_json::Value>, b: &CausalEnvelope<serde_json::Value>) -> &CausalEnvelope<serde_json::Value> {
a.timestamp_ns
.cmp(&b.timestamp_ns)
.then_with(|| a.event_id.cmp(&b.event_id))
}
Integration with Kubernetes: Sidecar Pattern for Clock Sync
apiVersion: apps/v1
kind: Deployment
metadata:
name: order-consumer
spec:
replicas: 3
template:
spec:
containers:
- name: consumer
image: my-registry/causal-kafka-consumer:v2.1
env:
- name: KAFKA_BROKERS
value: 'kafka-0.kafka-headless:9092,kafka-1.kafka-headless:9092'
- name: NODE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
securityContext:
privileges: true
volumes:
- name: chrony-config
configMap:
name: chrony-config
Final Thoughts
Causal consistency in distributed event streams is one of those topics that sits at the intersection of distributed systems theory and practical engineering. Most teams either ignore the problem or over-engineer a solution with distributed transactions. The vector clock approach — lightweight, embeddable in your consumer logic, and compatible with Kafka's partitioning model — gives you the right trade-off: causal correctness where it matters, with minimal overhead where it doesn't.
The full implementation, including integration tests with rdkafka test containers and property-based tests for the vector clock comparison logic, is available in the companion repository linked below.
What's your approach to causal ordering in event-driven systems? Have you hit the "concurrent event" edge case in production? I'd love to hear how you handled it.