Async Processing Patterns Mastery: CQRS, Saga, Event Sourcing — Senior Backend Developer Essential Guide

Synchronous:  Client -> A(50ms) -> B(100ms) -> C(200ms) = 350ms total wait
Asynchronous: Client -> A(50ms) -> Message Broker -> (B, C process in parallel)
              Client gets response after 50ms, B and C process asynchronously

// Domain Event - within bounded context
public record OrderItemAdded(
    UUID orderId,
    UUID productId,
    int quantity,
    BigDecimal price,
    Instant occurredAt
) implements DomainEvent {}

// Integration Event - published to external services
public record OrderPlacedEvent(
    UUID orderId,
    UUID customerId,
    BigDecimal totalAmount,
    List<OrderLineItem> items,
    Instant occurredAt,
    int version  // schema version management
) implements IntegrationEvent {}

Event Bus (Kafka):
  Producer -> [Topic: orders] -> Consumer Group A (Order Service)
                               -> Consumer Group B (Notification Service)
                               -> Consumer Group C (Analytics Service)

Event Store:
  [Event #1: OrderCreated]
  [Event #2: ItemAdded]
  [Event #3: PaymentReceived]
  [Event #4: OrderShipped]
  -> Replay events #1-#4 to reconstruct current order state

Traditional CRUD:
  Client -> [Single Model] -> Database
            (both read/write)

CQRS:
  Client --Command--> [Write Model] -> Command DB (normalized)
                            | Events
  Client <--Query---- [Read Model]  <- Read DB (denormalized, optimized)

// === Command Side: Order Creation ===

@Service
@RequiredArgsConstructor
public class OrderCommandService {

    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Transactional
    public UUID createOrder(CreateOrderCommand command) {
        // 1. Business rule validation
        validateOrder(command);

        // 2. Process in domain model
        Order order = Order.create(
            command.customerId(),
            command.items(),
            command.shippingAddress()
        );

        // 3. Save to Write DB
        orderRepository.save(order);

        // 4. Publish event (for Read Model sync)
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(),
            order.getCustomerId(),
            order.getTotalAmount(),
            order.getItems().stream()
                .map(this::toEventItem)
                .toList(),
            Instant.now()
        );
        kafkaTemplate.send("order-events", order.getId().toString(), event);

        return order.getId();
    }

    private void validateOrder(CreateOrderCommand command) {
        if (command.items().isEmpty()) {
            throw new InvalidOrderException("Order items are empty");
        }
        // Check inventory, customer status, etc.
    }
}

// === Query Side: Order Query Projection ===

@Service
@RequiredArgsConstructor
public class OrderQueryService {

    private final OrderReadRepository readRepository;

    public OrderDetailView getOrderDetail(UUID orderId) {
        return readRepository.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));
    }

    public Page<OrderSummaryView> getCustomerOrders(
            UUID customerId, Pageable pageable) {
        return readRepository.findByCustomerId(customerId, pageable);
    }
}

// Read Model - denormalized view optimized for queries
@Document(collection = "order_views")
public record OrderDetailView(
    UUID orderId,
    UUID customerId,
    String customerName,      // Denormalized: includes customer name
    String customerEmail,
    BigDecimal totalAmount,
    String status,
    List<OrderItemView> items,
    AddressView shippingAddress,
    Instant createdAt,
    Instant lastUpdatedAt
) {}

// === Event Handler: Write -> Read Sync ===

@Component
@RequiredArgsConstructor
public class OrderProjectionHandler {

    private final OrderReadRepository readRepository;
    private final CustomerQueryClient customerClient;

    @KafkaListener(topics = "order-events", groupId = "order-projection")
    public void handleOrderEvent(OrderEvent event) {
        switch (event) {
            case OrderCreatedEvent e -> handleOrderCreated(e);
            case OrderStatusChangedEvent e -> handleStatusChanged(e);
            case OrderCancelledEvent e -> handleOrderCancelled(e);
            default -> log.warn("Unknown event type: {}", event.getClass());
        }
    }

    private void handleOrderCreated(OrderCreatedEvent event) {
        // Fetch customer info to create denormalized view
        CustomerInfo customer = customerClient.getCustomer(event.customerId());

        OrderDetailView view = new OrderDetailView(
            event.orderId(),
            event.customerId(),
            customer.name(),
            customer.email(),
            event.totalAmount(),
            "CREATED",
            event.items().stream().map(this::toItemView).toList(),
            null,
            event.occurredAt(),
            event.occurredAt()
        );
        readRepository.save(view);
    }
}

E-commerce Order Flow (Choreography):

Order Service --OrderPlaced--> Payment Service
                                    |
                            PaymentCompleted
                                    |
                                    v
                              Inventory Service
                                    |
                            InventoryReserved
                                    |
                                    v
                              Shipping Service
                                    |
                            ShipmentCreated
                                    |
                                    v
                              Order Service (status update)

On Failure (Compensating Transactions):
Inventory Service --InventoryReservationFailed--> Payment Service (refund)
                                                        |
                                                 PaymentRefunded
                                                        |
                                                        v
                                                 Order Service (cancel order)

// === Choreography Saga: Order Service ===

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Transactional
    public UUID placeOrder(PlaceOrderCommand command) {
        Order order = Order.create(command);
        order.setStatus(OrderStatus.PENDING);
        orderRepository.save(order);

        // Publish event - Payment Service subscribes
        kafkaTemplate.send("order-events", new OrderPlacedEvent(
            order.getId(),
            order.getCustomerId(),
            order.getTotalAmount(),
            order.getItems()
        ));

        return order.getId();
    }

    // Payment completed event listener
    @KafkaListener(topics = "payment-events", groupId = "order-service")
    public void onPaymentCompleted(PaymentCompletedEvent event) {
        Order order = orderRepository.findById(event.orderId()).orElseThrow();
        order.setStatus(OrderStatus.PAID);
        orderRepository.save(order);
    }

    // Payment failed event listener
    @KafkaListener(topics = "payment-events", groupId = "order-service")
    public void onPaymentFailed(PaymentFailedEvent event) {
        Order order = orderRepository.findById(event.orderId()).orElseThrow();
        order.setStatus(OrderStatus.CANCELLED);
        order.setCancellationReason(event.reason());
        orderRepository.save(order);
    }
}

// === Choreography Saga: Payment Service ===

@Service
@RequiredArgsConstructor
public class PaymentService {

    private final PaymentRepository paymentRepository;
    private final PaymentGateway paymentGateway;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @KafkaListener(topics = "order-events", groupId = "payment-service")
    @Transactional
    public void onOrderPlaced(OrderPlacedEvent event) {
        try {
            // Process payment
            PaymentResult result = paymentGateway.charge(
                event.customerId(),
                event.totalAmount()
            );

            Payment payment = Payment.create(
                event.orderId(),
                event.customerId(),
                event.totalAmount(),
                result.transactionId()
            );
            paymentRepository.save(payment);

            // Publish success event - Inventory Service subscribes
            kafkaTemplate.send("payment-events", new PaymentCompletedEvent(
                event.orderId(),
                payment.getId(),
                result.transactionId()
            ));

        } catch (PaymentException e) {
            // Publish failure event - Order Service subscribes and cancels
            kafkaTemplate.send("payment-events", new PaymentFailedEvent(
                event.orderId(),
                e.getMessage()
            ));
        }
    }
}

E-commerce Order Flow (Orchestration):

                    Order Saga Orchestrator
                   /          |          \
                  v           v           v
          Payment Service  Inventory  Shipping
          (charge)         (reserve)  (create)
                  v           v           v
          (result)        (result)    (result)
                   \          |          /
                    Order Saga Orchestrator
                         (final decision)

// === Orchestration Saga: Using Temporal Framework ===

@WorkflowInterface
public interface OrderSagaWorkflow {

    @WorkflowMethod
    OrderResult processOrder(PlaceOrderCommand command);
}

@WorkflowImpl(workflows = OrderSagaWorkflow.class)
public class OrderSagaWorkflowImpl implements OrderSagaWorkflow {

    // Activity stubs for each service
    private final PaymentActivities paymentActivities =
        Workflow.newActivityStub(PaymentActivities.class,
            ActivityOptions.newBuilder()
                .setStartToCloseTimeout(Duration.ofSeconds(30))
                .setRetryOptions(RetryOptions.newBuilder()
                    .setMaximumAttempts(3)
                    .build())
                .build());

    private final InventoryActivities inventoryActivities =
        Workflow.newActivityStub(InventoryActivities.class,
            ActivityOptions.newBuilder()
                .setStartToCloseTimeout(Duration.ofSeconds(30))
                .build());

    private final ShippingActivities shippingActivities =
        Workflow.newActivityStub(ShippingActivities.class,
            ActivityOptions.newBuilder()
                .setStartToCloseTimeout(Duration.ofSeconds(60))
                .build());

    @Override
    public OrderResult processOrder(PlaceOrderCommand command) {
        String paymentId = null;
        String reservationId = null;

        try {
            // Step 1: Payment
            paymentId = paymentActivities.processPayment(
                command.customerId(),
                command.totalAmount()
            );

            // Step 2: Inventory reservation
            reservationId = inventoryActivities.reserveInventory(
                command.items()
            );

            // Step 3: Create shipment
            String shipmentId = shippingActivities.createShipment(
                command.orderId(),
                command.shippingAddress()
            );

            return OrderResult.success(command.orderId(), shipmentId);

        } catch (Exception e) {
            // Execute compensating transactions
            compensate(paymentId, reservationId);
            return OrderResult.failed(command.orderId(), e.getMessage());
        }
    }

    private void compensate(String paymentId, String reservationId) {
        // Execute compensating transactions in reverse order
        if (reservationId != null) {
            inventoryActivities.cancelReservation(reservationId);
        }
        if (paymentId != null) {
            paymentActivities.refundPayment(paymentId);
        }
    }
}

// === Activity Implementation ===

@ActivityInterface
public interface PaymentActivities {
    String processPayment(UUID customerId, BigDecimal amount);
    void refundPayment(String paymentId);
}

@Component
@RequiredArgsConstructor
public class PaymentActivitiesImpl implements PaymentActivities {

    private final PaymentGateway paymentGateway;
    private final PaymentRepository paymentRepository;

    @Override
    public String processPayment(UUID customerId, BigDecimal amount) {
        PaymentResult result = paymentGateway.charge(customerId, amount);
        Payment payment = Payment.create(customerId, amount, result.transactionId());
        paymentRepository.save(payment);
        return payment.getId().toString();
    }

    @Override
    public void refundPayment(String paymentId) {
        Payment payment = paymentRepository.findById(UUID.fromString(paymentId))
            .orElseThrow();
        paymentGateway.refund(payment.getTransactionId());
        payment.setStatus(PaymentStatus.REFUNDED);
        paymentRepository.save(payment);
    }
}

// Compensating transaction mapping
public enum CompensationAction {
    PAYMENT_CHARGE    -> PAYMENT_REFUND,
    INVENTORY_RESERVE -> INVENTORY_RELEASE,
    SHIPPING_CREATE   -> SHIPPING_CANCEL,
    COUPON_APPLY      -> COUPON_RESTORE,
    POINTS_DEDUCT     -> POINTS_RESTORE
}

Traditional CRUD:
  Account table: id=1, balance=750

Event Sourcing:
  Event #1: AccountCreated(id=1, initialBalance=1000)
  Event #2: MoneyWithdrawn(id=1, amount=200)
  Event #3: MoneyDeposited(id=1, amount=150)
  Event #4: MoneyWithdrawn(id=1, amount=200)
  -> Replay: 1000 - 200 + 150 - 200 = 750 (current balance)

// === Event Store Interface ===

public interface EventStore {
    void saveEvents(UUID aggregateId, List<DomainEvent> events, int expectedVersion);
    List<DomainEvent> getEvents(UUID aggregateId);
    List<DomainEvent> getEvents(UUID aggregateId, int fromVersion);
}

// === PostgreSQL-based Event Store Implementation ===

@Repository
@RequiredArgsConstructor
public class PostgresEventStore implements EventStore {

    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;

    @Override
    @Transactional
    public void saveEvents(UUID aggregateId, List<DomainEvent> events,
                          int expectedVersion) {
        // Optimistic concurrency control
        Integer currentVersion = jdbcTemplate.queryForObject(
            "SELECT COALESCE(MAX(version), 0) FROM event_store WHERE aggregate_id = ?",
            Integer.class, aggregateId
        );

        if (currentVersion != expectedVersion) {
            throw new ConcurrencyException(
                "Concurrency conflict: expected version " + expectedVersion +
                ", actual version " + currentVersion
            );
        }

        int version = expectedVersion;
        for (DomainEvent event : events) {
            version++;
            jdbcTemplate.update(
                """
                INSERT INTO event_store
                (event_id, aggregate_id, aggregate_type, event_type,
                 event_data, version, created_at)
                VALUES (?, ?, ?, ?, ?::jsonb, ?, ?)
                """,
                UUID.randomUUID(),
                aggregateId,
                event.getAggregateType(),
                event.getClass().getSimpleName(),
                objectMapper.writeValueAsString(event),
                version,
                Instant.now()
            );
        }
    }

    @Override
    public List<DomainEvent> getEvents(UUID aggregateId) {
        return jdbcTemplate.query(
            "SELECT * FROM event_store WHERE aggregate_id = ? ORDER BY version",
            this::mapRowToEvent,
            aggregateId
        );
    }
}

-- Event Store table schema
CREATE TABLE event_store (
    event_id       UUID PRIMARY KEY,
    aggregate_id   UUID NOT NULL,
    aggregate_type VARCHAR(255) NOT NULL,
    event_type     VARCHAR(255) NOT NULL,
    event_data     JSONB NOT NULL,
    version        INTEGER NOT NULL,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    UNIQUE (aggregate_id, version)  -- optimistic concurrency control
);

CREATE INDEX idx_event_store_aggregate ON event_store (aggregate_id, version);
CREATE INDEX idx_event_store_type ON event_store (event_type, created_at);

// === Event-Sourced Aggregate ===

public class OrderAggregate {

    private UUID id;
    private UUID customerId;
    private OrderStatus status;
    private BigDecimal totalAmount;
    private List<OrderItem> items = new ArrayList<>();
    private int version = 0;

    // Uncommitted events list
    private final List<DomainEvent> uncommittedEvents = new ArrayList<>();

    // Restore state from event history
    public static OrderAggregate fromHistory(List<DomainEvent> history) {
        OrderAggregate aggregate = new OrderAggregate();
        for (DomainEvent event : history) {
            aggregate.apply(event, false);
        }
        return aggregate;
    }

    // Handle new commands
    public void placeOrder(UUID customerId, List<OrderItem> items) {
        if (this.status != null) {
            throw new IllegalStateException("Order already created");
        }

        BigDecimal total = items.stream()
            .map(item -> item.price().multiply(BigDecimal.valueOf(item.quantity())))
            .reduce(BigDecimal.ZERO, BigDecimal::add);

        apply(new OrderPlacedEvent(this.id, customerId, total, items, Instant.now()), true);
    }

    public void cancelOrder(String reason) {
        if (this.status == OrderStatus.SHIPPED) {
            throw new IllegalStateException("Cannot cancel shipped orders");
        }
        apply(new OrderCancelledEvent(this.id, reason, Instant.now()), true);
    }

    // Apply event (state change)
    private void apply(DomainEvent event, boolean isNew) {
        when(event);  // State mutation
        this.version++;
        if (isNew) {
            uncommittedEvents.add(event);
        }
    }

    // State change logic per event type
    private void when(DomainEvent event) {
        switch (event) {
            case OrderPlacedEvent e -> {
                this.id = e.orderId();
                this.customerId = e.customerId();
                this.totalAmount = e.totalAmount();
                this.items = new ArrayList<>(e.items());
                this.status = OrderStatus.PLACED;
            }
            case OrderCancelledEvent e -> {
                this.status = OrderStatus.CANCELLED;
            }
            case OrderShippedEvent e -> {
                this.status = OrderStatus.SHIPPED;
            }
            default -> throw new IllegalArgumentException("Unknown event: " + event);
        }
    }

    public List<DomainEvent> getUncommittedEvents() {
        return Collections.unmodifiableList(uncommittedEvents);
    }

    public void clearUncommittedEvents() {
        uncommittedEvents.clear();
    }
}

// Snapshot save and restore
@Service
public class SnapshotService {

    private static final int SNAPSHOT_INTERVAL = 100; // every 100 events

    public OrderAggregate loadAggregate(UUID aggregateId) {
        // 1. Get latest snapshot
        Optional<Snapshot> snapshot = snapshotRepository
            .findLatest(aggregateId);

        if (snapshot.isPresent()) {
            // 2. Restore state from snapshot
            OrderAggregate aggregate = deserialize(snapshot.get().getData());
            int fromVersion = snapshot.get().getVersion();

            // 3. Replay only events after snapshot
            List<DomainEvent> newEvents = eventStore
                .getEvents(aggregateId, fromVersion + 1);
            for (DomainEvent event : newEvents) {
                aggregate.apply(event, false);
            }
            return aggregate;

        } else {
            // No snapshot, replay all
            List<DomainEvent> allEvents = eventStore.getEvents(aggregateId);
            return OrderAggregate.fromHistory(allEvents);
        }
    }

    public void saveIfNeeded(OrderAggregate aggregate) {
        if (aggregate.getVersion() % SNAPSHOT_INTERVAL == 0) {
            snapshotRepository.save(new Snapshot(
                aggregate.getId(),
                aggregate.getVersion(),
                serialize(aggregate),
                Instant.now()
            ));
        }
    }
}

// Dangerous code: Dual write problem
@Transactional
public void createOrder(Order order) {
    orderRepository.save(order);           // Step 1: DB save
    kafkaTemplate.send("orders", event);   // Step 2: Message publish
    // Step 1 success + Step 2 failure = In DB but event not published
    // Step 1 success + Step 2 success + DB commit failure = Event published but not in DB
}

1. Business logic + outbox INSERT = single DB transaction (atomicity guaranteed)
2. Separate process (CDC/Poller) reads outbox table and publishes to Kafka
3. After successful publish, mark outbox record as completed

-- Outbox table schema
CREATE TABLE outbox (
    id              UUID PRIMARY KEY,
    aggregate_type  VARCHAR(255) NOT NULL,
    aggregate_id    UUID NOT NULL,
    event_type      VARCHAR(255) NOT NULL,
    payload         JSONB NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at    TIMESTAMPTZ,
    status          VARCHAR(20) DEFAULT 'PENDING'
);

CREATE INDEX idx_outbox_pending ON outbox (status, created_at)
    WHERE status = 'PENDING';

// === Outbox Pattern Implementation ===

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxRepository outboxRepository;

    @Transactional  // Guaranteed by single transaction
    public UUID createOrder(CreateOrderCommand command) {
        // 1. Execute business logic
        Order order = Order.create(command);
        orderRepository.save(order);

        // 2. Save event to outbox in same transaction
        OutboxMessage message = OutboxMessage.builder()
            .aggregateType("Order")
            .aggregateId(order.getId())
            .eventType("OrderCreated")
            .payload(objectMapper.writeValueAsString(
                new OrderCreatedEvent(order.getId(), order.getTotalAmount())
            ))
            .build();
        outboxRepository.save(message);

        return order.getId();
        // When transaction commits, both order and outbox are saved together
    }
}

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "dbz_password",
    "database.dbname": "orderdb",
    "database.server.name": "order-service",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.type": "event_type",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.route.topic.replacement": "events.order"
  }
}

# Kafka Producer configuration
spring.kafka.producer.properties.enable.idempotence=true
spring.kafka.producer.properties.acks=all
spring.kafka.producer.properties.max.in.flight.requests.per.connection=5
spring.kafka.producer.properties.retries=2147483647

// === Idempotent Consumer Implementation ===

@Component
@RequiredArgsConstructor
public class IdempotentOrderEventHandler {

    private final ProcessedEventRepository processedEventRepository;
    private final OrderProjectionRepository orderProjectionRepository;

    @KafkaListener(topics = "order-events", groupId = "order-projection")
    @Transactional
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            @Header("eventId") String eventId) {

        // 1. Check if event already processed
        if (processedEventRepository.existsById(eventId)) {
            log.info("Ignoring duplicate event: eventId={}", eventId);
            return;
        }

        // 2. Process business logic
        processEvent(event);

        // 3. Record as processed (same transaction)
        processedEventRepository.save(new ProcessedEvent(
            eventId,
            event.getClass().getSimpleName(),
            Instant.now()
        ));
    }

    private void processEvent(OrderEvent event) {
        switch (event) {
            case OrderCreatedEvent e -> createOrderProjection(e);
            case OrderStatusChangedEvent e -> updateOrderStatus(e);
            default -> log.warn("Unknown event type");
        }
    }
}

-- Processed events table
CREATE TABLE processed_events (
    event_id    VARCHAR(255) PRIMARY KEY,
    event_type  VARCHAR(255) NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

// Redis SET NX for fast duplicate check
@Component
public class RedisIdempotencyGuard {

    private final StringRedisTemplate redisTemplate;
    private static final Duration TTL = Duration.ofHours(24);

    public boolean tryAcquire(String idempotencyKey) {
        Boolean result = redisTemplate.opsForValue()
            .setIfAbsent(
                "idempotency:" + idempotencyKey,
                "processed",
                TTL
            );
        return Boolean.TRUE.equals(result);
    }
}

// === Spring Kafka Retry + DLQ Configuration ===

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public DefaultErrorHandler errorHandler(
            KafkaTemplate<String, Object> kafkaTemplate) {

        // Recoverer that sends to DLQ
        DeadLetterPublishingRecoverer recoverer =
            new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, ex) -> new TopicPartition(
                    record.topic() + ".DLQ", record.partition()));

        // Exponential Backoff retry configuration
        ExponentialBackOff backOff = new ExponentialBackOff();
        backOff.setInitialInterval(1000L);    // 1 second
        backOff.setMultiplier(2.0);            // 2x increase
        backOff.setMaxInterval(60000L);        // max 60 seconds
        backOff.setMaxElapsedTime(300000L);    // max 5 minutes

        DefaultErrorHandler errorHandler =
            new DefaultErrorHandler(recoverer, backOff);

        // Non-retryable exceptions (business errors)
        errorHandler.addNotRetryableExceptions(
            InvalidOrderException.class,
            DuplicateEventException.class
        );

        return errorHandler;
    }
}

// === DLQ Monitoring Service ===

@Component
@RequiredArgsConstructor
public class DlqMonitoringService {

    private final MeterRegistry meterRegistry;
    private final AlertService alertService;

    @KafkaListener(topics = "order-events.DLQ", groupId = "dlq-monitor")
    public void monitorDlq(
            ConsumerRecord<String, Object> record,
            @Header(KafkaHeaders.DLT_EXCEPTION_MESSAGE) String errorMessage,
            @Header(KafkaHeaders.DLT_ORIGINAL_TOPIC) String originalTopic) {

        // Record metrics
        meterRegistry.counter("dlq.messages.received",
            "originalTopic", originalTopic,
            "errorType", classifyError(errorMessage)
        ).increment();

        // Send alert
        alertService.sendAlert(DlqAlert.builder()
            .originalTopic(originalTopic)
            .messageKey(record.key())
            .errorMessage(errorMessage)
            .timestamp(Instant.ofEpochMilli(record.timestamp()))
            .severity(classifySeverity(errorMessage))
            .build()
        );

        log.error("DLQ message received: topic={}, key={}, error={}",
            originalTopic, record.key(), errorMessage);
    }
}

// === DLQ Message Reprocessing ===

@RestController
@RequestMapping("/api/dlq")
@RequiredArgsConstructor
public class DlqReprocessController {

    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final DlqMessageRepository dlqRepository;

    // Single message reprocessing
    @PostMapping("/reprocess/single")
    public ResponseEntity<String> reprocessSingle(
            @RequestParam String messageId) {
        DlqMessage message = dlqRepository.findById(messageId).orElseThrow();

        // Republish to original topic
        kafkaTemplate.send(
            message.getOriginalTopic(),
            message.getKey(),
            message.getValue()
        );

        message.setStatus("REPROCESSED");
        message.setReprocessedAt(Instant.now());
        dlqRepository.save(message);

        return ResponseEntity.ok("Reprocessing requested: " + messageId);
    }

    // Batch reprocessing
    @PostMapping("/reprocess/batch")
    public ResponseEntity<String> reprocessBatch(
            @RequestParam String originalTopic,
            @RequestParam(defaultValue = "100") int batchSize) {
        List<DlqMessage> messages = dlqRepository
            .findByOriginalTopicAndStatus(originalTopic, "PENDING",
                PageRequest.of(0, batchSize));

        int count = 0;
        for (DlqMessage message : messages) {
            kafkaTemplate.send(message.getOriginalTopic(),
                message.getKey(), message.getValue());
            message.setStatus("REPROCESSED");
            count++;
        }
        dlqRepository.saveAll(messages);

        return ResponseEntity.ok(count + " messages reprocessing requested");
    }
}

// === Kafka Exactly-Once Configuration ===

@Configuration
public class KafkaExactlyOnceConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-");
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTransactionManager<String, Object> kafkaTransactionManager(
            ProducerFactory<String, Object> producerFactory) {
        return new KafkaTransactionManager<>(producerFactory);
    }

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        return new DefaultKafkaConsumerFactory<>(config);
    }
}

End-to-End Exactly-Once Chain:

Producer (idempotent + transactions)
    -> Kafka Broker (transaction log)
        -> Consumer (read_committed + manual offset)
            -> Database (idempotency key + business logic)

All links in the chain must work together for Exactly-Once guarantee

// === End-to-End Exactly-Once Processing ===

@Component
@RequiredArgsConstructor
public class ExactlyOnceOrderProcessor {

    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final OrderRepository orderRepository;
    private final ProcessedOffsetRepository offsetRepository;

    @KafkaListener(topics = "payment-completed")
    @Transactional("chainedKafkaTransactionManager")
    public void processPaymentCompleted(
            ConsumerRecord<String, PaymentCompletedEvent> record) {

        String offsetKey = record.topic() + "-" +
            record.partition() + "-" + record.offset();

        // Check if offset already processed
        if (offsetRepository.existsById(offsetKey)) {
            return;
        }

        // Business logic
        PaymentCompletedEvent event = record.value();
        Order order = orderRepository.findById(event.orderId()).orElseThrow();
        order.markAsPaid(event.paymentId());
        orderRepository.save(order);

        // Publish next step event (same Kafka transaction)
        kafkaTemplate.send("order-paid",
            new OrderPaidEvent(order.getId(), event.paymentId()));

        // Record processed offset (same DB transaction)
        offsetRepository.save(new ProcessedOffset(offsetKey, Instant.now()));
    }
}

// === Reactor-based Async Processing ===

@Service
public class ReactiveOrderService {

    public Flux<OrderEvent> processOrderStream() {
        return kafkaReceiver.receive()
            .flatMap(record -> {
                OrderEvent event = record.value();
                return processEvent(event)
                    .then(record.receiverOffset().commit())
                    .thenReturn(event);
            }, 16)  // Concurrency limit (backpressure)
            .onErrorResume(e -> {
                log.error("Processing failed", e);
                return Mono.empty();
            })
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)));
    }

    private Mono<Void> processEvent(OrderEvent event) {
        return Mono.fromCallable(() -> {
            // Business logic
            return null;
        }).subscribeOn(Schedulers.boundedElastic()).then();
    }
}

// === Virtual Threads-based Async Processing ===

@Configuration
public class VirtualThreadConfig {

    @Bean
    public TomcatProtocolHandlerCustomizer<?> protocolHandlerCustomizer() {
        return protocolHandler -> {
            protocolHandler.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
        };
    }
}

@Service
public class VirtualThreadOrderService {

    // Synchronous code but running on Virtual Threads
    // -> blocking I/O only blocks Virtual Thread, not OS thread
    public OrderResult processOrder(UUID orderId) {
        // Parallel processing: using StructuredTaskScope
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {

            Subtask<PaymentResult> paymentTask = scope.fork(() ->
                paymentService.processPayment(orderId)
            );
            Subtask<InventoryResult> inventoryTask = scope.fork(() ->
                inventoryService.checkInventory(orderId)
            );

            scope.join();
            scope.throwIfFailed();

            return new OrderResult(
                paymentTask.get(),
                inventoryTask.get()
            );
        }
    }
}

                     E-commerce Payment System Architecture
                     ======================================

[Client] -> [API Gateway]
                |
                v
        [Order Service] --- Write DB (PostgreSQL)
           |                     |
           | (Outbox Pattern)    | (CDC - Debezium)
           |                     v
           |              [Kafka: order-events]
           |                /        |        \
           v               v        v         v
    [Saga Orchestrator]  [Payment] [Inventory] [Notification]
    (Temporal)           Service   Service     Service
           |                |        |
           | commands       |        |
           v                v        v
    [payment-commands]  [payment-  [inventory-
                        events]    events]
                           |
                    [DLQ + Monitoring]
                           |
                    [Alerting (PagerDuty)]

Read Side:
    [Kafka: order-events] -> [Order Projection] -> Read DB (MongoDB)
                           -> [Analytics Service] -> Data Warehouse
                           -> [Search Service] -> Elasticsearch