10.1.4 – End-to-End EOS Pipelines

Module: M10 – Advanced Topics & Architecture
Section: 10.1 – Transactions & Exactly-Once Semantics
Goal: Design and implement end-to-end exactly-once semantic pipelines

Introduction

This module covers designing end-to-end EOS pipelines that guarantee exactly-once processing across multiple stages, topics, and systems.

What You'll Learn:

• Multi-stage pipeline architecture
• Chaining transactional producers and consumers
• Integrating external systems (databases, APIs)
• State management in EOS pipelines
• Error recovery and failover strategies

EOS Pipeline Architecture

Single-Stage Pipeline

Architecture:

Kafka Topic A
  ↓ (transactional consumer + producer)
Transform
  ↓
Kafka Topic B

Configuration:

// Consumer: read_committed
consumer.props.put("isolation.level", "read_committed");
consumer.props.put("enable.auto.commit", "false");

// Producer: transactional
producer.props.put("transactional.id", "transform-stage-1");
producer.props.put("enable.idempotence", "true");

Code:

producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    if (!records.isEmpty()) {
        try {
            producer.beginTransaction();
            
            for (ConsumerRecord<String, String> record : records) {
                String transformed = transform(record.value());
                producer.send(new ProducerRecord<>("topic-b", record.key(), transformed));
            }
            
            Map<TopicPartition, OffsetAndMetadata> offsets = getOffsets(records);
            producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
            producer.commitTransaction();
            
        } catch (ProducerFencedException | OutOfOrderSequenceException e) {
            producer.close();
            consumer.close();
            throw e;
        } catch (KafkaException e) {
            producer.abortTransaction();
        }
    }
}

Guarantee: Messages from Topic A are transformed and written to Topic B exactly once.

Multi-Stage Pipeline

Architecture:

Kafka Topic A
  ↓ (Stage 1: filter)
Kafka Topic B
  ↓ (Stage 2: enrich)
Kafka Topic C
  ↓ (Stage 3: aggregate)
Kafka Topic D

Each Stage:

• Transactional consumer (reads from upstream)
• Transactional producer (writes to downstream)
• Independent transactional.id per stage

Scaling:

• Each stage runs independently
• Can scale horizontally (multiple instances per stage)
• Failures isolated per stage

Fan-Out Pipeline

Architecture:

Kafka Topic (orders)
  ├─→ Topic A (payments)
  ├─→ Topic B (inventory)
  └─→ Topic C (notifications)

Code:

producer.beginTransaction();

for (ConsumerRecord<String, String> record : records) {
    Order order = parse(record.value());
    
    producer.send(new ProducerRecord<>("payments", order.getPaymentData()));
    producer.send(new ProducerRecord<>("inventory", order.getInventoryData()));
    producer.send(new ProducerRecord<>("notifications", order.getNotificationData()));
}

producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();

Guarantee: All three downstream messages written atomically.

Fan-In Pipeline

Architecture:

Topic A (orders)    ─┐
Topic B (payments)  ─┼─→ Join/Aggregate → Topic C
Topic C (inventory) ─┘

Challenge: Coordinating offsets from multiple input topics.

Solution: Use Kafka Streams (handles fan-in EOS automatically) or manually track offsets per input partition.

State Management in EOS Pipelines

Stateless Transformations

Example: Filter, map, enrich (without state).

// No state - each message processed independently
String transformed = record.value().toUpperCase();
producer.send(new ProducerRecord<>("output", transformed));

Advantage: Simple, no state recovery needed.

Stateful Transformations

Example: Aggregations, joins, windowing (require state).

Challenge: State must be consistent with committed offsets.

Solution 1: Kafka Streams (recommended for stateful EOS)

Configuration:

Properties config = new Properties();
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "eos-streams-app");
config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");  // Enable EOS
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

Example 1: Stateless Transformation with EOS:

StreamsBuilder builder = new StreamsBuilder();

builder.stream("orders")
       .mapValues(value -> "Processed: " + value)
       .to("processed-orders");

KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();

Example 2: Stateful Aggregation with EOS:

// Kafka Streams handles state + EOS automatically
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> input = builder.stream("input-topic");

KTable<String, Long> counts = input
    .groupByKey()
    .count(Materialized.as("counts-store"));  // State store

counts.toStream().to("output-topic");

KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();

// Graceful shutdown
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

Example 3: Join with EOS:

StreamsBuilder builder = new StreamsBuilder();

KStream<String, Order> orders = builder.stream("orders");
KTable<String, Customer> customers = builder.table("customers");

KStream<String, EnrichedOrder> enriched = orders.join(
    customers,
    (order, customer) -> new EnrichedOrder(order, customer)
);

enriched.to("enriched-orders");

// With exactly_once_v2, all joins and writes are atomic
KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();

What Kafka Streams Does for EOS:

1. Transactional Producer: Internally uses transactional producer for writes
2. Atomic State Updates: State store updates and output writes committed together
3. Offset Management: Consumer offsets committed via sendOffsetsToTransaction()
4. Changelog Topics: State store changelogs replicated transactionally
5. Automatic Recovery: On failure, rewinds to last committed offset and replays

Guarantee: State, changelogs, output messages, and offsets committed atomically.

Key Configuration:

# exactly_once_v2 (recommended, requires Kafka 2.5+)
processing.guarantee=exactly_once_v2

# Legacy (Kafka 0.11 - 2.4)
processing.guarantee=exactly_once

# No EOS (at-least-once)
processing.guarantee=at_least_once

Performance Notes:

• exactly_once_v2: ~5-10% overhead (optimized two-phase commit)
• exactly_once (legacy): ~15-20% overhead
• Works best with batching (commit.interval.ms=100 default)

Solution 2: External State Store (e.g., database)

Architecture:

Kafka Topic
  ↓
Consumer (read_committed)
  ↓
Update Database + Send to Kafka (transaction)

Challenge: Database and Kafka are separate systems—no distributed transaction.

Workaround 1: Idempotent Database Writes

producer.beginTransaction();

for (ConsumerRecord<String, String> record : records) {
    // Idempotent UPDATE (uses unique key, won't duplicate)
    db.execute("UPDATE users SET balance = ? WHERE id = ?", balance, userId);
    
    producer.send(new ProducerRecord<>("output", record.key(), record.value()));
}

producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();

Guarantee: If Kafka commits but DB fails, retry is safe (idempotent).

Workaround 2: Outbox Pattern

// 1. Write to DB outbox table (single DB transaction)
db.beginTransaction();
db.insert("INSERT INTO events (id, payload) VALUES (?, ?)", eventId, payload);
db.commit();

// 2. Separate process reads outbox and sends to Kafka transactionally
while (true) {
    List<Event> events = db.query("SELECT * FROM events WHERE sent = false LIMIT 100");
    
    producer.beginTransaction();
    for (Event event : events) {
        producer.send(new ProducerRecord<>("topic", event.getPayload()));
        db.execute("UPDATE events SET sent = true WHERE id = ?", event.getId());
    }
    producer.commitTransaction();
}

Guarantee: Database is source of truth, Kafka is eventually consistent.

External System Integration Patterns (Deep Dive)

The Challenge

Kafka's EOS guarantees only apply within Kafka. Once you write to external systems (databases, APIs), you need additional patterns to prevent:

• Duplicates: If a consumer crashes after writing externally but before committing offsets
• Inconsistency: If only half of a multi-system update succeeds

Pattern 1: Idempotent Database Writes

A. Database UPSERTs with Message IDs

Pattern:

1. Store a unique message ID (e.g., kafka_offset + partition) in the database
2. Use INSERT ... ON CONFLICT (message_id) DO NOTHING (PostgreSQL) or REPLACE (MySQL)

Example (PostgreSQL):

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    message_id VARCHAR(255) UNIQUE,  -- Kafka offset + partition
    order_data JSONB
);

-- Idempotent write in consumer:
INSERT INTO orders (message_id, order_data)
VALUES ('orders-0-123', '{"amount": 100}')  -- 'orders-0-123' = topic-partition-offset
ON CONFLICT (message_id) DO NOTHING;

Java Implementation:

producer.beginTransaction();

for (ConsumerRecord<String, String> record : records) {
    // Create unique message ID
    String messageId = String.format("%s-%d-%d", 
                                      record.topic(), 
                                      record.partition(), 
                                      record.offset());
    
    // Idempotent database write
    jdbcTemplate.update(
        "INSERT INTO orders (message_id, order_data) VALUES (?, ?) ON CONFLICT (message_id) DO NOTHING",
        messageId, record.value()
    );
    
    // Produce to Kafka
    producer.send(new ProducerRecord<>("processed-orders", record.key(), "processed"));
}

producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();

Pros: Simple, no distributed transactions
Cons: Requires deterministic message IDs

B. HTTP APIs with Idempotency Keys

Pattern:

1. Send a unique Idempotency-Key header (e.g., Kafka message key + offset)
2. Server caches responses for duplicate keys

Example (Stripe-style API):

// Consumer code:
for (ConsumerRecord<String, String> record : records) {
    String idempotencyKey = record.key() + "-" + record.offset();
    
    HttpRequest request = HttpRequest.newBuilder()
        .uri(URI.create("https://api.example.com/payments"))
        .header("Idempotency-Key", idempotencyKey)
        .POST(HttpRequest.BodyPublishers.ofString(record.value()))
        .build();
    
    HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
}

// Server-side logic (pseudo-code):
if (cache.contains(idempotencyKey)) {
    return cache.get(idempotencyKey);  // Return cached response
} else {
    processPayment();
    cache.put(idempotencyKey, response);
}

Pros: Works with any REST API
Cons: Server must implement deduplication

Pattern 2: Two-Phase Commit (2PC)

For atomic updates across Kafka and databases.

Pattern:

1. Phase 1 (Prepare):
   • Write to DB in a "pending" state (e.g., status = 'PROCESSING')
   • Send a Kafka message (e.g., payment_pending)
2. Phase 2 (Commit):
   • After Kafka transaction commits, update DB to status = 'COMPLETED'

Example Flow:

// Pseudocode for a payment service
public void processPayment(Payment payment) {
    // Phase 1: Prepare
    db.execute("UPDATE payments SET status = 'PROCESSING' WHERE id = ?", payment.getId());
    producer.send(new ProducerRecord<>("payment_events", payment.getId(), "pending"));

    // Phase 2: Commit (only if Kafka succeeds)
    producer.commitTransaction();
    db.execute("UPDATE payments SET status = 'COMPLETED' WHERE id = ?", payment.getId());
}

Complete Implementation:

public class TwoPhaseCommitProcessor {
    private final JdbcTemplate db;
    private final KafkaProducer<String, String> producer;
    
    public void processOrder(ConsumerRecord<String, Order> record) {
        Order order = record.value();
        
        try {
            // Phase 1: Prepare DB
            db.update("UPDATE orders SET status = 'PROCESSING' WHERE id = ?", order.getId());
            
            // Phase 1: Prepare Kafka
            producer.beginTransaction();
            producer.send(new ProducerRecord<>("order-events", order.getId(), "processing"));
            
            // Phase 2: Commit Kafka
            producer.sendOffsetsToTransaction(getOffsets(record), consumer.groupMetadata());
            producer.commitTransaction();
            
            // Phase 2: Complete DB
            db.update("UPDATE orders SET status = 'COMPLETED' WHERE id = ?", order.getId());
            
        } catch (Exception e) {
            // Rollback
            producer.abortTransaction();
            db.update("UPDATE orders SET status = 'FAILED' WHERE id = ?", order.getId());
            throw e;
        }
    }
}

Pros: Strong consistency
Cons: Complex; requires careful error handling

Pattern 3: Outbox Pattern (Database → Kafka)

Best Practice: Outbox pattern for guaranteed delivery from database to Kafka.

Architecture:

Application
  ↓ (single DB transaction)
Database
  ├─ Business tables
  └─ Outbox table
  
Outbox Processor (separate process)
  ↓ (reads outbox, sends to Kafka transactionally)
Kafka

Implementation:

Step 1: Application writes to outbox:

@Transactional
public void createOrder(Order order) {
    // Business logic
    orderRepository.save(order);
    
    // Write to outbox
    OutboxEvent event = new OutboxEvent(
        UUID.randomUUID(),
        "order-created",
        order.toJson(),
        false  // not yet sent
    );
    outboxRepository.save(event);
    
    // Single DB transaction commits both
}

Step 2: Outbox processor sends to Kafka:

public class OutboxProcessor {
    @Scheduled(fixedDelay = 100)
    public void processOutbox() {
        List<OutboxEvent> events = outboxRepository.findBysentFalseLimit(100);
        
        if (!events.isEmpty()) {
            producer.beginTransaction();
            
            for (OutboxEvent event : events) {
                producer.send(new ProducerRecord<>(
                    event.getTopic(),
                    event.getKey(),
                    event.getPayload()
                ));
            }
            
            producer.commitTransaction();
            
            // Mark as sent (separate DB transaction)
            events.forEach(e -> e.setSent(true));
            outboxRepository.saveAll(events);
        }
    }
}

Pros: Database is source of truth, Kafka eventually consistent
Cons: Additional latency (polling delay)

Pattern 4: Saga Pattern (Distributed Transactions)

For coordinating multiple services with compensating transactions.

Example: Order placement saga

Order Service → Payment Service → Inventory Service → Notification Service
                     ↓ (failure)
                  Compensate (refund)

Implementation with Kafka:

// Orchestrator
producer.beginTransaction();
producer.send(new ProducerRecord<>("order-saga", orderId, "start"));
producer.send(new ProducerRecord<>("payment-commands", orderId, "charge"));
producer.commitTransaction();

// Payment Service
if (paymentSuccess) {
    producer.send(new ProducerRecord<>("inventory-commands", orderId, "reserve"));
} else {
    producer.send(new ProducerRecord<>("order-saga", orderId, "compensate"));
}

// Compensating transaction
producer.send(new ProducerRecord<>("payment-commands", orderId, "refund"));

Best Practices Summary

Database Write + Kafka (EOS)

Anti-Pattern: Non-atomic writes

// ❌ Database and Kafka committed separately (can diverge)
db.update(...);
producer.send(...);
producer.commitTransaction();

Best Practice: Use one of the patterns above (Outbox, 2PC, or Idempotent Writes).

API Calls in EOS Pipeline

Challenge: HTTP APIs are not transactional.

Solution 1: Idempotent API Calls

producer.beginTransaction();

for (ConsumerRecord<String, String> record : records) {
    // Idempotent HTTP PUT (idempotency key)
    httpClient.put("/resource/" + record.key(), record.value());
    
    producer.send(new ProducerRecord<>("output", record.key(), "processed"));
}

producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();

Guarantee: If transaction aborts, API call retried with same idempotency key (no duplication).

Solution 2: Compensating Transactions

try {
    producer.beginTransaction();
    
    // Call API
    String result = apiClient.processOrder(order);
    
    producer.send(new ProducerRecord<>("output", result));
    producer.commitTransaction();
    
} catch (Exception e) {
    // Compensate (undo API call)
    apiClient.cancelOrder(order.getId());
    producer.abortTransaction();
}

Error Recovery and Failover

Producer Fencing (Zombie Prevention)

Scenario: Old producer instance hasn't shut down cleanly.

Producer A (transactional.id=pipeline-1, epoch=1) running
Producer A crashes (doesn't clean up)

Producer B (same transactional.id=pipeline-1) starts
  → Coordinator increments epoch to 2
  → Returns PID with epoch=2

Producer A comes back online (epoch=1)
  → Sends message with epoch=1
  → Broker rejects: "Fenced by higher epoch"

Producer B continues (epoch=2) ✅

Key Point: Only one producer with a given transactional.id can write at a time.

Consumer Rebalancing During Transaction

Scenario: Consumer rebalances mid-transaction.

Consumer starts transaction
Consumer reads 100 messages
Consumer loses partition (rebalanced to another consumer)
Consumer tries to commit transaction
  → Fails: "No longer owns partition"

Solution: Abort transaction on rebalance, let new consumer retry.

consumer.subscribe(Arrays.asList("input"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        try {
            producer.commitTransaction();  // Try to commit before losing partition
        } catch (Exception e) {
            producer.abortTransaction();  // Abort if can't commit
        }
    }
    
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Start fresh transaction
    }
});

Coordinator Failover

Scenario: Transaction coordinator broker fails.

Producer mid-transaction
Coordinator fails
New coordinator elected
Producer retries commitTransaction()
  → New coordinator doesn't know about transaction
  → Aborts transaction automatically

Recovery: Producer detects failure, aborts, retries full transaction.

Performance Optimization

Batch Size Tuning

Anti-Pattern: One transaction per message

for (Message msg : messages) {
    producer.beginTransaction();
    producer.send(msg);
    producer.commitTransaction();  // ❌ Overhead: 5-10ms per transaction
}

Best Practice: Batch 100-1000 messages per transaction

producer.beginTransaction();
for (Message msg : batch) {
    producer.send(msg);  // Batch 100-1000 messages
}
producer.commitTransaction();  // ✅ Overhead amortized

Throughput Impact: 10-100x improvement with batching.

Parallelism

Architecture:

Kafka Topic (10 partitions)
  ↓
10 Consumer Instances (each with unique transactional.id)
  ↓
Process independently

Key: Each instance has unique transactional.id (e.g., pipeline-1-instance-0, pipeline-1-instance-1, etc.).

Monitoring EOS Pipelines

Key Metrics

1. Transaction Duration

kafka.producer:type=producer-metrics,client-id=*,attribute=txn-commit-time-avg

Alert If: > 10 seconds (slow transactions blocking consumers).

2. Aborted Transactions

kafka.producer:type=producer-metrics,client-id=*,attribute=txn-abort-total

Alert If: Increasing steadily (indicates errors).

3. Consumer Lag (with LSO)

LEO - LSO = "Blocked Lag" (messages blocked by uncommitted transactions)

Alert If: Blocked lag > 1000 messages (stuck transaction).

Key Takeaways

1. End-to-End EOS: Chain transactional consumers and producers with read_committed isolation.

2. Multi-Stage: Independent stages with unique transactional.id per stage.

3. State Management: Use Kafka Streams for stateful EOS, or external stores with idempotency.

4. External Systems: Use outbox pattern or idempotent APIs for database/HTTP integration.

5. Error Recovery: Fencing prevents zombie producers. Abort on rebalance and retry.

6. Performance: Batch 100-1000 messages per transaction to amortize overhead.

7. Monitoring: Track transaction duration, aborted transactions, and blocked lag.

What's Next?

Next Module: 10.1.5 – EOS Anti-Patterns and Debugging

What We Covered:

• ✅ End-to-end EOS pipeline architecture
• ✅ Multi-stage, fan-out, fan-in patterns
• ✅ State management strategies
• ✅ External system integration
• ✅ Error recovery and performance optimization

What's Coming:

• ❌ Common EOS anti-patterns
• ❌ Debugging transactional issues
• ❌ Performance tradeoffs

References:

• Transaction protocol → 10.1.1
• Producer transactions → 10.1.2
• Consumer isolation → 10.1.3
• Kafka Streams EOS → M5: Kafka Streams