Event Schema Versioning, Correlation IDs, Idempotency & Deduplication in Salesforce
In event-driven Salesforce architectures, messages often outlive the code that processes them. That’s why it’s critical to evolve schemas safely, trace events end-to-end, and handle duplicates without causing side effects. This guide walks you through practical approaches for versioned Platform Events, correlation/causation IDs, and idempotent consumers with deduplication, including compact Apex snippets you can use right away.
1) Event Schema Versioning
Goal:
You want to add new fields and evolve your schema without breaking existing consumers — while still letting new consumers process older events.
Guiding Principles:
-
Additive changes are safe: You can freely add optional fields.
-
Avoid renaming or removing fields: Instead, add new ones, mark old ones as deprecated, and use an upcaster on the consumer side.
-
Include a version field: For example,
Version__cmakes each payload self-describing. -
Use an upcaster: Converts older versions to the current internal model, filling in defaults as needed.
-
Maintain a registry: Store
MinSupportedVersionandCurrentVersionin Custom Metadata. Reject events that are too old but still handle them idempotently.
Versioning Approaches:
-
Single event type with evolving payload: Common for additive changes.
-
Separate event type per major version: For unavoidable breaking changes.
-
Optional schema hash: Protects against accidental schema drift.
2) Correlation & Causation IDs
Why they matter:
In distributed systems spanning multiple triggers, flows, Queueables, and external integrations, correlation and causation IDs let you trace the full journey of a transaction.
Fields to Include:
-
MessageId: A unique identifier for each event (UUID).
-
CorrelationId: Remains constant across the entire business process (e.g., saga, order, user journey).
-
CausationId: The MessageId of the event that triggered this one — useful for building a complete event lineage.
Best Practices:
-
Pass CorrelationId downstream unchanged.
-
Set CausationId to the MessageId of the triggering event.
-
Log all three IDs on writes, updates, or errors — invaluable for debugging distributed flows.
3) Idempotency & Deduplication
The Challenge:
Platform Events are delivered at least once and replay mechanisms can sometimes resend messages or deliver them out of order.
The Solution:
-
Claim-then-act pattern:
-
Build a deterministic DedupKey (e.g.,
${BusinessId}:${LogicalVersion}or a UUID). -
Try inserting it into a dedicated Idempotency__c table with a unique constraint.
-
If insertion succeeds, process the event. If it fails with a duplicate, skip safely — the event is already handled.
-
Keep deduplication records at least as long as your Platform Event retention window.
-
Design Tips:
-
Prefer predictable, business-based keys over random UUIDs.
-
Commit both the claim and business logic in the same transaction.
-
Separate retryable errors from permanent ones, and avoid looping indefinitely on the same key.
Real-World Example with Code
Scenario:
A fulfillment system publishes Order_Event__e events for Created, Updated, and Shipped actions. You’ll add schema versioning, correlation/causation tracking, and an idempotent consumer that handles version upgrades automatically.
A) Platform Event Definition & Publisher (Versioned + Traced)
// Fields on Order_Event__e (example):
// - Version__c (Number, 2,0)
// - EventType__c (Text)
// - OrderId__c (Text)
// - MessageId__c (Text) // UUID
// - CorrelationId__c (Text) // trace across services
// - CausationId__c (Text) // parent event's MessageId
// - DedupKey__c (Text) // unique logical key
// - Payload__c (Long Text Area) // JSON
public with sharing class OrderEventPublisher {
public static final Integer CURRENT_VERSION = 2;
public class PayloadV2 {
public Decimal amount;
public String currency;
public String customerTier; // new in v2
}
public static void publishCreated(String orderId, String correlationId, PayloadV2 p) {
String msgId = Crypto.getRandomUUID();
String dedup = orderId + ':created:1'; // deterministic logical version for "created"
Order_Event__e e = new Order_Event__e(
Version__c = CURRENT_VERSION,
EventType__c = 'Created',
OrderId__c = orderId,
MessageId__c = msgId,
CorrelationId__c= String.isBlank(correlationId) ? msgId : correlationId,
CausationId__c = null,
DedupKey__c = dedup,
Payload__c = JSON.serialize(p)
);
Database.SaveResult sr = EventBus.publish(e);
System.assert(sr.isSuccess(), 'PE publish failed: ' + sr);
}
}
B) Idempotency Store (Unique Claim Table)
// Custom object: Idempotency__c
// Fields: Key__c (Text, External ID, Unique), ConsumedAt__c (Datetime), Notes__c (Long Text)
public with sharing class Idempotency {
public static Boolean claim(String key, String notes) {
try {
insert new Idempotency__c(Key__c = key, ConsumedAt__c = System.now(), Notes__c = notes);
return true;
} catch (DmlException ex) {
// DUPLICATE_VALUE means already processed (safe to skip)
for (Database.Error err : ex.getDmlFields(0).getErrors())
if (err.getStatusCode() == StatusCode.DUPLICATE_VALUE) return false;
throw ex;
}
}
}
C) Version-Aware Consumer with Upcaster (Normalize to v2)
// Upcaster: converts older payloads to the current internal model
public with sharing class OrderUpcaster {
public class OrderModel {
public String orderId;
public String type;
public Decimal amount;
public String currency;
public String customerTier; // default if missing
}
public static OrderModel toCurrent(Integer version, String eventType, String orderId, String payloadJson) {
Map<String,Object> m = (Map<String,Object>) JSON.deserializeUntyped(payloadJson);
OrderModel out = new OrderModel();
out.orderId = orderId;
out.type = eventType;
if (version == null || version <= 1) {
// v1 had fields: amount, currency (no customerTier)
out.amount = (Decimal) m.get('amount');
out.currency = (String) m.get('currency');
out.customerTier = 'Standard'; // default for old messages
} else {
// v2 adds customerTier
out.amount = (Decimal) m.get('amount');
out.currency = (String) m.get('currency');
out.customerTier = (String) m.get('customerTier');
if (String.isBlank(out.customerTier)) out.customerTier = 'Standard';
}
return out;
}
}
// OrderEventTrigger.trigger — idempotent, version-aware consumer
trigger OrderEventTrigger on Order_Event__e (after insert) {
// Optional guard: ensure we don't accept events older than our min support
Integer MIN_SUPPORTED = 1;
List<Order__c> upserts = new List<Order__c>();
for (Order_Event__e e : Trigger.New) {
if (e.Version__c != null && e.Version__c < MIN_SUPPORTED) {
// Too old; record and skip (still idempotent if it replays)
Idempotency.claim('skip:'+e.DedupKey__c, 'Too old version '+e.Version__c);
continue;
}
// Claim idempotency; if duplicate, skip processing
if (!Idempotency.claim(e.DedupKey__c, 'MsgId='+e.MessageId__c)) continue;
OrderUpcaster.OrderModel model =
OrderUpcaster.toCurrent((Integer)e.Version__c, e.EventType__c, e.OrderId__c, e.Payload__c);
// Apply side effects (upsert by external id keeps it idempotent too)
if (model.type == 'Created') {
upserts.add(new Order__c(External_Id__c = model.orderId,
Status__c='Open',
Amount__c=model.amount,
CurrencyIsoCode=model.currency,
Customer_Tier__c=model.customerTier,
Last_Correlation_Id__c=e.CorrelationId__c,
Last_Causation_Id__c=e.CausationId__c));
} else if (model.type == 'Updated') {
upserts.add(new Order__c(External_Id__c = model.orderId,
Amount__c=model.amount,
CurrencyIsoCode=model.currency,
Customer_Tier__c=model.customerTier,
Last_Correlation_Id__c=e.CorrelationId__c,
Last_Causation_Id__c=e.CausationId__c));
} else if (model.type == 'Shipped') {
upserts.add(new Order__c(External_Id__c=model.orderId,
Status__c='Shipped',
Shipped_At__c=System.now(),
Last_Correlation_Id__c=e.CorrelationId__c,
Last_Causation_Id__c=e.CausationId__c));
}
}
if (!upserts.isEmpty()) {
Database.upsert(upserts, Order__c.Fields.External_Id__c, /*allOrNone*/ false);
}
}
D) OrderEventTrigger — Idempotent, Version-Aware Consumer
// Custom Metadata: Event_Versioning__mdt (DeveloperName='OrderEvent')
// Fields: MinSupported__c (Number), Current__c (Number)
public with sharing class EventVersioning {
public static Integer minSupported(String devName) {
Event_Versioning__mdt cfg = [SELECT MinSupported__c FROM Event_Versioning__mdt WHERE DeveloperName = :devName LIMIT 1];
return (Integer) cfg.MinSupported__c;
}
}
Final Thoughts
Treat Platform Events like public APIs — once published, they should never break. Evolve them additively, upcast old payloads when necessary, and make every consumer idempotent with a claim-then-act deduplication approach.
With these patterns in place, you can:
-
Replay events safely
-
Scale subscribers independently
-
Change schemas without causing production issues
