Exactly-Once Delivery (Challenges)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

/**
 * Why exactly-once delivery is impossible:
 * A code demonstration of the Two Generals Problem
 */
 
interface MessageResult {
    delivered: boolean;
    ackReceived: boolean;
}
 
class UnreliableNetwork {
    // Messages have a chance of being lost
    private lossRate: number = 0.1; // 10% loss
 
    send(message: any): boolean {
        return Math.random() > this.lossRate;
    }
}
 
class Producer {
    private network: UnreliableNetwork;
 
    async sendExactlyOnce(message: Message): Promise<boolean> {
        // Attempt 1: Send the message
        const sent = this.network.send(message);
        
        if (!sent) {
            // We don't know if it was delivered
            // Option A: Retry → might cause duplicate
            // Option B: Don't retry → might be lost
            return false;
        }
 
        // Assume broker received and persisted
        // Now wait for acknowledgment
        const ackReceived = this.network.send({ type: 'ack' }); // ACK also uses network!
        
        if (!ackReceived) {
            // Broker received and persisted (we know this)
            // But ACK was lost
            // 
            // From producer's perspective: Did it work?
            // We don't know!
            //
            // If we retry: DUPLICATE
            // If we don't retry: MAYBE LOST (we don't know)
            //
            // THIS is why exactly-once is impossible.
            // No amount of ACKs can resolve this uncertainty.
        }
 
        return ackReceived;
    }
}
 
/**
 * The fundamental insight:
 * 
 * At the last step of any protocol, there is always a final message.
 * The sender of that message can NEVER know if it was received.
 * 
 * Therefore:
 * - If we assume success → might be wrong (message lost)
 * - If we assume failure and retry → might cause duplicate
 * 
 * There is no third option that guarantees exactly-once.
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

/**
 * Achieving exactly-once PROCESSING semantics
 * despite at-least-once DELIVERY semantics
 */
 
class OrderProcessor {
    private db: Database;
    private processedMessageIds: Set<string> = new Set();
 
    /**
     * This handler may be called MULTIPLE times for the same message
     * (at-least-once delivery) but the ORDER will only be created ONCE
     * (exactly-once processing)
     */
    async handleOrderMessage(message: OrderMessage): Promise<void> {
        // STEP 1: Check if we've processed this message before
        // This is the DEDUPLICATION step
        if (this.processedMessageIds.has(message.messageId)) {
            console.log(`Message ${message.messageId} already processed, skipping`);
            return; // Idempotent: second processing is no-op
        }
 
        // Check database for more durable deduplication
        const existingOrder = await this.db.orders.findUnique({
            where: { orderId: message.orderId }
        });
 
        if (existingOrder) {
            console.log(`Order ${message.orderId} already exists, skipping`);
            // Mark as processed so we skip faster next time
            this.processedMessageIds.add(message.messageId);
            return;
        }
 
        // STEP 2: Create the order (first time only)
        await this.db.orders.create({
            data: {
                orderId: message.orderId,
                customerId: message.customerId,
                items: message.items,
                processedFromMessageId: message.messageId, // Track for debugging
            }
        });
 
        // STEP 3: Mark as processed
        this.processedMessageIds.add(message.messageId);
 
        console.log(`Order ${message.orderId} created successfully`);
    }
}
 
/**
 * From the MESSAGE perspective (delivery):
 *   - Message might be delivered 1, 2, or 3 times
 *   - This is AT-LEAST-ONCE delivery
 * 
 * From the ORDER perspective (processing):
 *   - Order is created exactly once
 *   - This is EXACTLY-ONCE processing
 * 
 * The combination gives us the behavior we want:
 *   - No order is lost (at-least-once)
 *   - No duplicate orders (deduplication)
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

/**
 * Complete pattern for effective exactly-once semantics
 */
 
interface ProcessedMessage {
    messageId: string;
    processedAt: Date;
    result: 'success' | 'failure';
}
 
class EffectivelyExactlyOnceProcessor {
    private db: Database;
    
    /**
     * Process a message with effectively exactly-once semantics.
     * 
     * This is achieved by:
     * 1. Deduplication check (filter already-processed messages)
     * 2. Idempotent state change (atomic with deduplication record)
     * 3. Record processing (prevent future reprocessing)
     */
    async processExactlyOnce<T>(
        messageId: string,
        processFunc: () => Promise<T>
    ): Promise<{ processed: boolean; result?: T }> {
        
        // STEP 1: Check if already processed (deduplication)
        const existing = await this.db.processedMessages.findUnique({
            where: { messageId }
        });
        
        if (existing) {
            // Duplicate detected - skip processing
            // This is what makes it "exactly-once" from state perspective
            console.log(`Duplicate message ${messageId} detected`);
            return { processed: false };
        }
 
        // STEP 2: Process and record atomically
        // Use a transaction to ensure atomicity
        return await this.db.$transaction(async (tx) => {
            // Double-check within transaction (handle race conditions)
            const doubleCheck = await tx.processedMessages.findUnique({
                where: { messageId }
            });
            
            if (doubleCheck) {
                return { processed: false }; // Race condition, skip
            }
 
            // Execute the actual business logic
            const result = await processFunc();
 
            // Record that we processed this message
            await tx.processedMessages.create({
                data: {
                    messageId,
                    processedAt: new Date(),
                    result: 'success',
                }
            });
 
            return { processed: true, result };
        });
    }
}
 
// Usage in consumer
const processor = new EffectivelyExactlyOnceProcessor(db);
 
consumer.on('message', async (message) => {
    const { processed, result } = await processor.processExactlyOnce(
        message.id,
        async () => {
            // This function runs exactly once per unique message.id
            // Even if the message is delivered multiple times
            
            return await orderService.createOrder({
                orderId: message.payload.orderId,
                customerId: message.payload.customerId,
                items: message.payload.items,
            });
        }
    );
 
    if (processed) {
        console.log('Order created:', result);
    } else {
        console.log('Duplicate message, order already created');
    }
    
    // Acknowledge message (even if duplicate - it's handled)
    await message.ack();
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

import { Kafka } from 'kafkajs';
 
const kafka = new Kafka({
    clientId: 'order-service',
    brokers: ['kafka:9092'],
});
 
/**
 * Idempotent Producer
 * 
 * The broker assigns a Producer ID (PID) and tracks sequence numbers.
 * If the producer retries and sends the same (PID, sequence) pair,
 * the broker detects the duplicate and ignores it.
 * 
 * This solves: Producer retry causing duplicate messages
 */
const idempotentProducer = kafka.producer({
    // Enable idempotency
    idempotent: true,
    
    // Required for idempotency
    maxInFlightRequests: 5, // Max 5 (Kafka requirement)
    
    // Retry settings become safer with idempotency
    retry: {
        retries: Number.MAX_SAFE_INTEGER, // Safe to retry forever
    },
});
 
await idempotentProducer.connect();
 
// If this send is retried due to network issues,
// the broker will detect and deduplicate
await idempotentProducer.send({
    topic: 'orders',
    messages: [{ value: JSON.stringify(order) }],
});
 
/**
 * How it works:
 * 
 * Producer gets PID=5 on startup
 * 
 * Send message with sequence=1 → Broker stores, ACK
 * ACK lost, producer retries
 * Send message with sequence=1 again → Broker detects duplicate
 *                                       Returns success without re-storing
 * 
 * Result: Message appears exactly once in the log
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

/**
 * Transactional Producer
 * 
 * Allows producing to multiple partitions/topics atomically.
 * Either ALL messages are committed or NONE are.
 * 
 * This solves: Partial writes (some messages succeed, some fail)
 */
const transactionalProducer = kafka.producer({
    transactionalId: 'order-processor-tx', // Unique ID for this producer
    idempotent: true, // Required for transactions
});
 
await transactionalProducer.connect();
 
async function processOrderBatch(orders: Order[]): Promise<void> {
    // Begin transaction
    const transaction = await transactionalProducer.transaction();
 
    try {
        // All these messages are part of the same transaction
        for (const order of orders) {
            await transaction.send({
                topic: 'processed-orders',
                messages: [{ value: JSON.stringify(order) }],
            });
 
            await transaction.send({
                topic: 'order-events',
                messages: [{ 
                    value: JSON.stringify({ 
                        type: 'ORDER_PROCESSED', 
                        orderId: order.id 
                    }) 
                }],
            });
        }
 
        // Commit: ALL messages become visible atomically
        await transaction.commit();
        
    } catch (error) {
        // Abort: NONE of the messages become visible
        await transaction.abort();
        throw error;
    }
}
 
/**
 * Transactional guarantees:
 * 
 * - All messages in a transaction are committed atomically
 * - Consumers with isolation.level=read_committed see only committed messages
 * - If producer crashes mid-transaction, Kafka auto-aborts
 * - Recovery uses the transactionalId to resume or abort pending transactions
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

/**
 * The Complete Pattern: Consume-Transform-Produce
 * 
 * This is where Kafka's exactly-once really shines.
 * We can consume, process, and produce in a single atomic transaction.
 */
 
const consumer = kafka.consumer({
    groupId: 'order-processor',
    readUncommitted: false, // Only read committed messages
});
 
const producer = kafka.producer({
    transactionalId: 'order-processor-tx',
    idempotent: true,
});
 
await consumer.connect();
await producer.connect();
 
await consumer.subscribe({ topic: 'raw-orders' });
 
await consumer.run({
    autoCommit: false, // CRITICAL: We handle commits in transactions
    
    eachMessage: async ({ topic, partition, message }) => {
        const transaction = await producer.transaction();
        
        try {
            // 1. Process the message
            const processedOrder = await processOrder(
                JSON.parse(message.value!.toString())
            );
            
            // 2. Produce result to output topic (part of transaction)
            await transaction.send({
                topic: 'processed-orders',
                messages: [{ value: JSON.stringify(processedOrder) }],
            });
            
            // 3. Commit consumer offset AS PART OF the transaction
            await transaction.sendOffsets({
                consumerGroupId: 'order-processor',
                topics: [{
                    topic,
                    partitions: [{
                        partition,
                        offset: (BigInt(message.offset) + 1n).toString(),
                    }],
                }],
            });
            
            // 4. Commit everything atomically
            await transaction.commit();
            // At this point:
            // - Output message is visible
            // - Consumer offset is updated
            // Both or neither happen
            
        } catch (error) {
            await transaction.abort();
            // Nothing happened:
            // - No output message
            // - Consumer offset unchanged
            // - Message will be reprocessed
        }
    },
});
 
/**
 * EXACTLY-ONCE in Kafka-to-Kafka workflows:
 * 
 * 1. Input message consumed
 * 2. Output message produced 
 * 3. Input offset committed
 * 
 * All three happen atomically.
 * 
 * If crash after step 1: Offset not committed, message reprocessed
 * If crash after step 2: Transaction not committed, output not visible,
 *                        message reprocessed
 * If crash after step 3: Transaction committed, no reprocessing
 * 
 * Result: Each input message produces exactly one output message
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

class PaymentProcessor {
    /**
     * This is the classic external side effect problem.
     * 
     * We need to:
     * 1. Process a payment (external API call)
     * 2. Acknowledge the message (message broker)
     * 
     * But we can't make these atomic!
     */
    async processPaymentMessage(message: PaymentMessage): Promise<void> {
        // STEP 1: Call external payment API
        const paymentResult = await paymentGateway.charge({
            amount: message.amount,
            customerId: message.customerId,
            orderId: message.orderId,
        });
 
        // GAP: If we crash here, the payment was processed
        //      but the message wasn't acknowledged.
        //      Message will be redelivered.
        //      Customer will be charged TWICE.
 
        // STEP 2: Acknowledge the message
        await message.ack();
    }
}
 
/**
 * The failure modes:
 * 
 * Scenario A: Crash after payment, before ack
 *   - Payment processed
 *   - Message not acknowledged
 *   - Message redelivered
 *   - Payment processed AGAIN → DUPLICATE!
 * 
 * Scenario B: Ack before payment (at-most-once approach)
 *   - Message acknowledged
 *   - Crash before payment
 *   - Payment never processed → LOST!
 * 
 * There's no ordering that makes this atomic.
 * The payment API doesn't know about our message acknowledgment.
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

class IdempotentPaymentProcessor {
    /**
     * Use idempotency keys to make external API calls safe.
     * 
     * The payment gateway tracks idempotency keys and ensures
     * the same key results in the same response, without duplicate charges.
     */
    async processPaymentMessage(message: PaymentMessage): Promise<void> {
        // Use message ID as idempotency key
        // Same message = same idempotency key = no duplicate charge
        const idempotencyKey = `payment-${message.messageId}`;
 
        // Call payment API with idempotency key
        const paymentResult = await paymentGateway.charge({
            amount: message.amount,
            customerId: message.customerId,
            orderId: message.orderId,
            idempotencyKey,  // ← The crucial field
        });
 
        // Even if this message is redelivered and we call the API again,
        // the payment gateway will recognize the idempotency key and
        // return the original result without charging again.
 
        await message.ack();
    }
}
 
/**
 * How idempotency keys work at the payment gateway:
 * 
 * First call with key "payment-msg-123":
 *   - Process payment
 *   - Store result keyed by "payment-msg-123"
 *   - Return success
 * 
 * Second call with key "payment-msg-123":
 *   - Lookup "payment-msg-123"
 *   - Found! Return cached result
 *   - No duplicate charge
 * 
 * Popular APIs that support idempotency keys:
 *   - Stripe (Idempotency-Key header)
 *   - PayPal
 *   - Square
 *   - Many modern payment processors
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

class SafeOrderProcessor {
    private db: Database;
 
    /**
     * Check external state before acting.
     * 
     * This pattern queries the external system to determine
     * if the action was already performed.
     */
    async processOrderMessage(message: OrderMessage): Promise<void> {
        // Check if order was already processed
        // This could be a local database or the external system
        const existingShipment = await this.shippingService.findByOrderId(
            message.orderId
        );
 
        if (existingShipment) {
            // Already processed - this is a duplicate message
            console.log(`Order ${message.orderId} already shipped`);
            await message.ack();
            return;
        }
 
        // Not yet processed - safe to proceed
        await this.shippingService.createShipment({
            orderId: message.orderId,
            items: message.items,
            address: message.shippingAddress,
        });
 
        await message.ack();
    }
}
 
/**
 * Trade-offs of check-before-act:
 * 
 * Pros:
 *   - Simple to understand
 *   - Works with any external system
 * 
 * Cons:
 *   - Race condition window between check and act
 *   - Extra latency for the check call
 *   - Requires external system to support querying for duplicates
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

/**
 * Kafka Exactly-Once Semantics Performance Impact
 * 
 * Based on real-world measurements:
 */
 
const performanceMetrics = {
    // Idempotent Producer (vs non-idempotent)
    idempotentProducer: {
        latencyIncrease: '5-10%',   // Near negligible
        throughputImpact: '~0%',    // No significant impact
        reason: 'Sequence number tracking is lightweight'
    },
 
    // Transactional Producer (vs non-transactional)
    transactionalProducer: {
        latencyIncrease: '20-50%',  // Transaction overhead
        throughputImpact: '20-40%', // Commit/abort protocol
        reason: 'Two-phase commit with transaction coordinator'
    },
 
    // Consume-Transform-Produce with EOS
    fullEOS: {
        latencyIncrease: '30-80%',  // Full transaction path
        throughputImpact: '30-50%', // Depends on message size
        additionalConsiderations: [
            'Transaction coordinator becomes bottleneck at scale',
            'Longer processing times = longer transaction windows',
            'Broker resource usage increases',
            'Consumer group rebalancing becomes more complex'
        ]
    }
};
 
/**
 * When EOS overhead is acceptable:
 *   - Correctness is more valuable than raw throughput
 *   - Message value justifies the overhead
 *   - System isn't at extreme scale
 * 
 * When to reconsider EOS:
 *   - Processing millions of messages/second
 *   - Low-latency requirements (< 10ms)
 *   - Messages are low-value (metrics, logs)
 */