@PostMapping("/orders")
@Transactional
public ResponseEntity<OrderResponse> placeOrder(@RequestBody OrderRequest request) {
 
    // Step 1: Fraud check - SYNCHRONOUS (must block order if fraudulent)
    FraudAssessment fraud = fraudService.assess(request);  // 50ms
    if (fraud.isHighRisk()) {
        return ResponseEntity.status(402).body(OrderResponse.rejected("Fraud risk detected"));
    }
 
    // Step 2: Payment authorization - SYNCHRONOUS (must succeed before committing)
    PaymentResult payment = paymentGateway.authorize(request.getPaymentDetails()); // 200ms
    if (!payment.isSuccessful()) {
        return ResponseEntity.status(402).body(OrderResponse.rejected(payment.getDeclineReason()));
    }
 
    // Step 3: Save order and outbox event in ONE transaction (Outbox Pattern)
    Order order = Order.create(request, payment.getAuthorizationId());
    orderRepository.save(order);
    outboxRepository.save(OutboxEntry.orderPlaced(order));  // Same DB transaction
 
    // All other processing (inventory, email, warehouse, analytics) is ASYNC
    // These happen after the transaction commits via the Outbox relay
 
    return ResponseEntity.accepted().body(
        OrderResponse.accepted(order.getId(), "/orders/" + order.getId())
    );
    // Total response time: ~300ms (fraud + payment only)
    // User gets "Order confirmed" - everything else runs asynchronously
}

// Tier 1: Must complete within 1 minute (business-critical)
@KafkaListener(
    topics = "orders.events.order-lifecycle",
    groupId = "inventory-reservation",  // SLA: 30 seconds
    containerFactory = "highPriorityConsumerFactory"
)
public void reserveInventory(OrderPlacedEvent event) { ... }
 
// Tier 2: Must complete within 10 minutes (important but not critical)
@KafkaListener(
    topics = "orders.events.order-lifecycle",
    groupId = "warehouse-fulfillment",  // SLA: 5 minutes
    containerFactory = "standardConsumerFactory"
)
public void notifyWarehouse(OrderPlacedEvent event) { ... }
 
@KafkaListener(
    topics = "orders.events.order-lifecycle",
    groupId = "customer-notifications",  // SLA: 2 minutes
    containerFactory = "standardConsumerFactory"
)
public void sendConfirmationEmail(OrderPlacedEvent event) { ... }
 
// Tier 3: Can lag by hours (non-critical)
@KafkaListener(
    topics = "orders.events.order-lifecycle",
    groupId = "analytics-pipeline",  // SLA: 24 hours
    containerFactory = "lowPriorityConsumerFactory"
)
public void trackOrderAnalytics(OrderPlacedEvent event) { ... }
 
// Tier 3 consumers use different consumer configuration:
// - Larger batch sizes (efficiency over latency)
// - Higher linger.ms
// - Alerts trigger only when lag > 1 hour (not immediately)

// Inventory reservation can fail (item out of stock)
// Must trigger order cancellation and refund
 
@Service
public class InventoryReservationConsumer {
 
    @KafkaListener(topics = "orders.events.order-lifecycle",
                   groupId = "inventory-reservation")
    @Transactional
    public void handleOrderPlaced(OrderPlacedEvent event, Acknowledgment ack) {
        for (OrderItem item : event.getItems()) {
            boolean reserved = inventoryService.reserve(
                item.getSku(), item.getQuantity(), event.getOrderId()
            );
 
            if (!reserved) {
                // Saga compensating action: cancel the order
                kafkaTemplate.send("orders.commands.cancel-order",
                    event.getOrderId(),
                    CancelOrderCommand.builder()
                        .orderId(event.getOrderId())
                        .reason("INVENTORY_UNAVAILABLE")
                        .failedSku(item.getSku())
                        .triggeredBy("inventory-reservation-service")
                        .build()
                );
                ack.acknowledge();
                return;
            }
        }
        // All items reserved successfully - publish confirmation
        kafkaTemplate.send("orders.events.inventory-reserved",
            event.getOrderId(), InventoryReservedEvent.from(event));
        ack.acknowledge();
    }
}

@Service
public class LedgerUpdateConsumer {
 
    @KafkaListener(
        topics = "payments.events.transaction-lifecycle",
        groupId = "ledger-service",
        // CRITICAL: concurrency=1 per partition to maintain ordering
        // partition key = accountId ensures all transactions for same account
        // are processed in sequence
        containerFactory = "orderedFinancialConsumerFactory"
    )
    @Transactional(isolation = Isolation.SERIALIZABLE)
    public void updateLedger(
            TransactionEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {
 
        // Check if already applied (idempotency)
        if (ledgerRepository.isTransactionApplied(event.getTransactionId())) {
            log.info("Duplicate transaction ignored: {}", event.getTransactionId());
            ack.acknowledge();
            return;
        }
 
        Account account = accountRepository.findByIdWithLock(event.getAccountId())
            .orElseThrow(() -> new AccountNotFoundException(event.getAccountId()));
 
        // Validate business rules
        if (event.getType() == DEBIT && account.getBalance().compareTo(event.getAmount()) < 0) {
            // Insufficient funds - publish compensation event
            publishInsufficientFundsEvent(event);
            ack.acknowledge();
            return;
        }
 
        // Apply the balance change
        BigDecimal newBalance = event.getType() == DEBIT
            ? account.getBalance().subtract(event.getAmount())
            : account.getBalance().add(event.getAmount());
 
        // Update with version check (optimistic locking)
        int rowsUpdated = accountRepository.updateBalance(
            event.getAccountId(),
            newBalance,
            account.getVersion(),
            account.getVersion() + 1
        );
 
        if (rowsUpdated == 0) {
            // Concurrent modification - retry (will be re-delivered by Kafka)
            throw new OptimisticLockException("Concurrent balance update: " + event.getAccountId());
        }
 
        // Record ledger entry (immutable)
        ledgerRepository.insertEntry(LedgerEntry.builder()
            .transactionId(event.getTransactionId())
            .accountId(event.getAccountId())
            .type(event.getType())
            .amount(event.getAmount())
            .balanceBefore(account.getBalance())
            .balanceAfter(newBalance)
            .kafkaPartition(partition)
            .kafkaOffset(offset)
            .processedAt(Instant.now())
            .build()
        );
 
        ack.acknowledge();
    }
}

// Separate audit topic with 7-year retention
// Consumers write to it but can never modify or delete
 
@Component
public class AuditEventPublisher {
 
    @KafkaListener(topics = "payments.events.transaction-lifecycle",
                   groupId = "audit-writer")
    public void writeAuditRecord(TransactionEvent event, Acknowledgment ack) {
        // Transform to immutable audit record (no PII - only hashed identifiers)
        AuditRecord auditRecord = AuditRecord.builder()
            .eventId(event.getTransactionId())
            .eventType(event.getType().name())
            .accountIdHash(hashPII(event.getAccountId()))  // Hashed for GDPR
            .amount(event.getAmount())
            .currency(event.getCurrency())
            .timestamp(event.getTimestamp())
            .sourceRegion(event.getSourceRegion())
            .processorId(event.getProcessorId())
            .build();
 
        // Write to immutable audit topic (producer-only for most services)
        kafkaTemplate.send("payments.audit.transactions", event.getTransactionId(), auditRecord);
 
        // Also write to WORM storage (S3 Glacier) for regulatory compliance
        s3Client.putObject("financial-audit-archive",
            "transactions/" + event.getTimestamp().toLocalDate() + "/" + event.getTransactionId(),
            auditRecord);
 
        ack.acknowledge();
    }
}

@Service
public class NotificationFanOutConsumer {
 
    private static final int CELEBRITY_THRESHOLD = 100_000;
 
    @KafkaListener(
        topics = "social.events.user-activity",
        groupId = "notification-fanout",
        containerFactory = "highThroughputConsumerFactory"
    )
    public void handleUserActivity(
            UserActivityEvent event,
            Acknowledgment ack) {
 
        int followerCount = followerCountCache.get(event.getActorId());
 
        if (followerCount > CELEBRITY_THRESHOLD) {
            handleCelebrityFanOut(event);
        } else {
            handleNormalFanOut(event);
        }
        ack.acknowledge();
    }
 
    private void handleCelebrityFanOut(UserActivityEvent event) {
        // Store activity in Redis - followers PULL when they open the app
        // Key: celebrity:{userId}:feed, Value: sorted set by timestamp
        redisTemplate.opsForZSet().add(
            "celebrity:" + event.getActorId() + ":feed",
            event,
            (double) event.getTimestamp().toEpochMilli()
        );
        // Set TTL: activities older than 7 days auto-expire
        redisTemplate.expire("celebrity:" + event.getActorId() + ":feed", 7, TimeUnit.DAYS);
 
        // Optionally: send push notification to a SAMPLE of active followers
        // (too expensive to notify all 10M followers; sample 1M most-active)
        List<String> activeFollowers = followerService.getMostActive(event.getActorId(), 1_000_000);
        publishNotificationsBatch(activeFollowers, event);
    }
 
    private void handleNormalFanOut(UserActivityEvent event) {
        List<String> followers = followerService.getFollowers(event.getActorId());
 
        // Batch publish to notification topic (don't flood with individual messages)
        List<ProducerRecord<String, NotificationEvent>> batch = followers.stream()
            .map(followerId -> new ProducerRecord<>(
                "social.notifications.delivery",
                followerId,  // Partition by recipient for ordering
                NotificationEvent.of(followerId, event)
            ))
            .collect(toList());
 
        // Publish batch (Kafka handles batching internally)
        batch.forEach(record -> kafkaTemplate.send(record));
    }
}

// Final delivery leg: Kafka notification → WebSocket to live users
 
@Service
public class WebSocketDeliveryConsumer {
 
    private final Map<String, WebSocketSession> activeSessions = new ConcurrentHashMap<>();
 
    @KafkaListener(
        topics = "social.notifications.delivery",
        groupId = "websocket-delivery"
    )
    public void deliverToConnectedUser(
            NotificationEvent notification,
            Acknowledgment ack) {
 
        WebSocketSession session = activeSessions.get(notification.getRecipientId());
 
        if (session != null && session.isOpen()) {
            // User is online - deliver via WebSocket for instant delivery
            session.sendMessage(new TextMessage(serialize(notification)));
            log.debug("Delivered via WebSocket: userId={}", notification.getRecipientId());
        } else {
            // User is offline - send push notification instead
            pushNotificationService.send(notification);
            log.debug("Delivered via push: userId={}", notification.getRecipientId());
        }
 
        ack.acknowledge();
    }
}

// At 2M messages/second: every byte saved = significant network/storage savings
 
// BAD: JSON schema (bloated, slow)
{
  "deviceId": "device-4f8a3b2c",
  "timestamp": "2026-06-05T12:34:56.789Z",
  "temperature": 23.5,
  "humidity": 65.2,
  "pressure": 1013.25,
  "batteryLevel": 87
}
// Size: ~120 bytes per message
// 2M/sec × 120 bytes = 240 MB/second network load
 
// GOOD: Avro binary schema (compact, fast)
// Schema defined once in Schema Registry
{
  "type": "record",
  "name": "SensorReading",
  "namespace": "com.company.iot",
  "fields": [
    {"name": "deviceId",     "type": {"type": "fixed", "size": 16}},  // UUID as bytes
    {"name": "timestamp",    "type": "long"},                          // epoch millis
    {"name": "temperature",  "type": "float"},
    {"name": "humidity",     "type": "float"},
    {"name": "pressure",     "type": "float"},
    {"name": "batteryLevel", "type": "int"}
  ]
}
// Size: ~30 bytes per message (4× smaller)
// 2M/sec × 30 bytes = 60 MB/second (4× less network load)
// 5-year storage at 1-day retention: 60MB/sec × 86,400 = 5TB/day saved
 
// BEST for IoT: Protocol Buffers (even more compact for repeated messages)
message SensorReading {
  bytes device_id = 1;       // 16 bytes
  int64 timestamp = 2;       // 8 bytes
  float temperature = 3;     // 4 bytes
  float humidity = 4;        // 4 bytes
  float pressure = 5;        // 4 bytes
  int32 battery_level = 6;   // 4 bytes
}
// Size: ~42 bytes with field tags (similar to Avro, better for nested messages)

// Kafka Streams for real-time alerting with sliding windows
 
@Configuration
public class IoTAlertingTopology {
 
    @Bean
    public KStream<String, SensorReading> buildAlertingTopology(StreamsBuilder builder) {
 
        KStream<String, SensorReading> readings = builder.stream(
            "iot.events.sensor-readings",
            Consumed.with(Serdes.String(), sensorReadingAvroSerde())
        );
 
        // Alert: Temperature exceeding threshold for 5 consecutive minutes
        // (single spike OK; sustained high temp = equipment problem)
        readings
            .filter((deviceId, reading) -> reading.getTemperature() > 80.0f) // Threshold
            .groupByKey()
            .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)))
            .count()
            .toStream()
            .filter((windowedDeviceId, count) -> count >= 5) // 5 readings in 5 min = alert
            .mapValues((windowedDeviceId, count) -> DeviceAlert.builder()
                .deviceId(windowedDeviceId.key())
                .alertType("SUSTAINED_HIGH_TEMPERATURE")
                .windowStart(windowedDeviceId.window().startTime())
                .windowEnd(windowedDeviceId.window().endTime())
                .readingCount(count)
                .build()
            )
            .to("alert.events.device-alerts");
 
        // Alert: Battery below 10% AND below 5% in last reading
        readings
            .filter((deviceId, reading) -> reading.getBatteryLevel() < 10)
            .to("alert.events.low-battery");
 
        return readings;
    }
}

// Write IoT readings to ClickHouse in micro-batches for efficiency
// Individual inserts at 2M/sec would kill any database
// Batch 10,000 readings every 100ms instead
 
@Component
public class IoTBatchStorageConsumer {
 
    private final List<SensorReading> batch = new ArrayList<>(10000);
    private final ScheduledExecutorService flusher = Executors.newScheduledThreadPool(1);
 
    @PostConstruct
    public void scheduleFlushing() {
        flusher.scheduleAtFixedRate(this::flush, 100, 100, TimeUnit.MILLISECONDS);
    }
 
    @KafkaListener(
        topics = "iot.events.sensor-readings",
        groupId = "iot-clickhouse-writer",
        containerFactory = "batchIoTConsumerFactory"
    )
    public void bufferReadings(List<SensorReading> readings) {
        synchronized (batch) {
            batch.addAll(readings);
            if (batch.size() >= 10000) {
                flush(); // Force flush if batch is large enough
            }
        }
    }
 
    private void flush() {
        List<SensorReading> toFlush;
        synchronized (batch) {
            if (batch.isEmpty()) return;
            toFlush = new ArrayList<>(batch);
            batch.clear();
        }
 
        try {
            clickhouseClient.insertBatch("iot_readings", toFlush);
            // ClickHouse handles 100M+ rows/second in batch mode
        } catch (Exception e) {
            log.error("ClickHouse batch write failed: {} readings", toFlush.size(), e);
            // Write failed readings to error topic for retry
            toFlush.forEach(r -> kafkaTemplate.send("iot.events.storage-failures", r));
        }
    }
}

// Command handler: validates, then publishes events to Kafka (the event store)
 
@Service
public class AccountCommandHandler {
 
    @PostMapping("/accounts/{accountId}/deposits")
    @Transactional
    public DepositResult handleDeposit(
            @PathVariable String accountId,
            @RequestBody DepositCommand command) {
 
        // Load current state by replaying events (or from snapshot + recent events)
        Account account = accountProjector.getCurrentState(accountId);
 
        // Validate the command against current state
        account.validateDeposit(command); // Throws if invalid
 
        // Create the event (the fact that will be stored)
        MoneyDepositedEvent event = MoneyDepositedEvent.builder()
            .eventId(UUID.randomUUID().toString())
            .accountId(accountId)
            .amount(command.getAmount())
            .currency(command.getCurrency())
            .reference(command.getReference())
            .depositedAt(Instant.now())
            .sequenceNumber(account.getEventSequenceNumber() + 1)
            .build();
 
        // Publish to Kafka (the event store)
        // This is the ONLY write - Kafka IS the source of truth
        kafkaTemplate.send("accounts.events.account-lifecycle", accountId, event)
            .get(5, TimeUnit.SECONDS); // Synchronous for command handlers
 
        return DepositResult.success(event.getEventId(), event.getSequenceNumber());
    }
}
 
// Account state is built by replaying events
@Service
public class AccountProjector {
 
    public Account getCurrentState(String accountId) {
        // Try snapshot first (for performance with long event histories)
        Optional<AccountSnapshot> snapshot = snapshotRepository.findLatest(accountId);
 
        Account account;
        long fromSequenceNumber;
 
        if (snapshot.isPresent()) {
            account = Account.fromSnapshot(snapshot.get());
            fromSequenceNumber = snapshot.get().getSequenceNumber() + 1;
        } else {
            account = Account.empty(accountId);
            fromSequenceNumber = 0;
        }
 
        // Replay events after snapshot
        List<AccountEvent> events = accountEventRepository.findByAccountIdAndSequenceGte(
            accountId, fromSequenceNumber
        );
 
        for (AccountEvent event : events) {
            account.apply(event);
        }
 
        return account;
    }
}

// The killer feature of event sourcing + Kafka: rebuild any projection from scratch
 
@Service
public class AccountBalanceProjectionRebuilder {
 
    /**
     * Rebuild the balance projection from scratch.
     * Called when: schema change, bug fix, new analytics requirement.
     * Reads ALL historical events from Kafka.
     */
    public void rebuildFromScratch() {
        log.info("Starting balance projection rebuild from Kafka offset 0");
 
        // Create a temporary consumer with a unique group ID for this rebuild
        String rebuildGroupId = "rebuild-balance-projection-" + Instant.now().toEpochMilli();
 
        try (KafkaConsumer<String, AccountEvent> consumer = createConsumer(rebuildGroupId)) {
            consumer.subscribe(Collections.singletonList("accounts.events.account-lifecycle"));
            consumer.seekToBeginning(consumer.assignment());
 
            // Clear existing projection (rebuild = fresh start)
            balanceProjectionRepository.truncate();
 
            Map<String, BigDecimal> inMemoryBalances = new HashMap<>();
            int count = 0;
 
            while (true) {
                ConsumerRecords<String, AccountEvent> records = consumer.poll(Duration.ofSeconds(1));
                if (records.isEmpty()) break; // Reached end of topic
 
                for (ConsumerRecord<String, AccountEvent> record : records) {
                    applyEventToProjection(record.value(), inMemoryBalances);
                    count++;
                }
 
                consumer.commitSync();
                if (count % 100_000 == 0) {
                    // Flush in-memory state to DB every 100K events
                    flushToDatabase(inMemoryBalances);
                    log.info("Rebuild progress: {} events processed", count);
                }
            }
 
            flushToDatabase(inMemoryBalances); // Final flush
            log.info("Rebuild complete: {} total events processed", count);
        }
    }
}

# MirrorMaker 2 configuration (running in EU-West, replicating FROM US-East)
clusters = us-east, eu-west
 
us-east.bootstrap.servers = us-east-kafka-1:9093,us-east-kafka-2:9093,us-east-kafka-3:9093
eu-west.bootstrap.servers = eu-west-kafka-1:9093,eu-west-kafka-2:9093,eu-west-kafka-3:9093
 
# Replicate these critical topics
us-east->eu-west.topics = orders.events.order-lifecycle, \
                           payments.events.transaction-lifecycle, \
                           users.events.user-lifecycle
 
# Preserve source offsets for seamless consumer group failover
# This is the key: EU consumers can resume from EXACTLY where US consumers left off
us-east->eu-west.emit.checkpoints.enabled = true
us-east->eu-west.sync.group.offsets.enabled = true
us-east->eu-west.sync.group.offsets.interval.seconds = 10
 
# Topic name in EU: us-east.orders.events.order-lifecycle
# Consumers use the prefix to distinguish replicated topics from local topics

// Health monitor that triggers failover
 
@Component
public class RegionFailoverController {
 
    private static final Duration US_EAST_TIMEOUT = Duration.ofSeconds(30);
    private volatile boolean failoverActive = false;
 
    @Scheduled(fixedDelay = 5000) // Check every 5 seconds
    public void checkPrimaryRegionHealth() {
        if (failoverActive) return;
 
        boolean usEastHealthy = checkUSEastHealth();
        if (!usEastHealthy) {
            log.error("US-East region health check failed - initiating failover to EU-West");
            initiateFailover();
        }
    }
 
    private void initiateFailover() {
        failoverActive = true;
 
        // Step 1: Switch DNS to EU-West load balancer (< 60 seconds TTL)
        dnsService.updateRecord("orders.api.company.com", euWestLoadBalancer);
 
        // Step 2: Start EU-West consumers (they were idle/suspended)
        // EU consumers use the MIRRORED topic with translated consumer group offsets
        euWestConsumerManager.activateConsumerGroups(
            "order-processor",
            "payment-processor",
            "notification-service"
        );
        // These consumers start from the same position US consumers left off
        // because MirrorMaker syncs consumer group offsets every 10 seconds
 
        // Step 3: Alert on-call team
        alertingService.sendCritical(
            "FAILOVER ACTIVE: Orders processing shifted to EU-West",
            Map.of("region", "EU-West", "trigger", "US-East health check failure")
        );
 
        log.info("Failover complete: EU-West is now processing orders");
    }
 
    private boolean checkUSEastHealth() {
        try {
            // Check: Can we connect to US-East Kafka broker?
            // Check: Is consumer lag within acceptable bounds?
            // Check: Can we reach US-East API health endpoint?
            return usEastKafkaAdmin.listTopics().names().get(US_EAST_TIMEOUT) != null
                && consumerLagMonitor.getMaxLag("order-processor") < 10_000
                && httpClient.get("https://us-east.internal/health").getStatusCode() == 200;
        } catch (Exception e) {
            log.warn("US-East health check failed: {}", e.getMessage());
            return false;
        }
    }
}

@Service
public class OrderService {
 
    @Value("${feature.async.email:false}")
    private boolean asyncEmail;
 
    @Value("${feature.async.warehouse:false}")
    private boolean asyncWarehouse;
 
    @Value("${feature.async.inventory:false}")
    private boolean asyncInventory;
 
    @Value("${feature.dual-publish:true}")
    private boolean dualPublish;
 
    public OrderResult placeOrder(OrderRequest request) {
        Order order = createOrder(request);
 
        // Always publish to Kafka (dual-publish during migration)
        if (dualPublish) {
            kafkaTemplate.send("order-events", order.getId(), OrderPlacedEvent.from(order));
        }
 
        // Service 1: Inventory (safest to migrate first)
        if (!asyncInventory) {
            inventoryService.reserve(order);  // Sync (old way)
        }
        // If asyncInventory=true: Kafka consumer handles this
 
        // Service 2: Email (already async)
        if (!asyncEmail) {
            emailService.sendConfirmation(order);  // Sync fallback
        }
        // If asyncEmail=true: Kafka consumer handles this
 
        // Service 3: Warehouse (higher risk - migrate last)
        if (!asyncWarehouse) {
            warehouseService.notifyFulfillment(order);  // Sync fallback
        }
        // If asyncWarehouse=true: Kafka consumer handles this
 
        return OrderResult.success(order.getId());
    }
}

// During migration: publish to BOTH brokers
 
@Service
public class MigrationAwareEventPublisher {
 
    @Value("${migration.dual-publish.kafka.enabled:false}")
    private boolean kafkaPublishEnabled;
 
    @Value("${migration.dual-publish.rabbitmq.enabled:true}")
    private boolean rabbitPublishEnabled;
 
    public void publish(DomainEvent event) {
        Throwable rabbitError = null;
        Throwable kafkaError = null;
 
        if (rabbitPublishEnabled) {
            try {
                rabbitTemplate.convertAndSend(
                    event.getExchange(), event.getRoutingKey(), event);
            } catch (Exception e) {
                rabbitError = e;
            }
        }
 
        if (kafkaPublishEnabled) {
            try {
                kafkaTemplate.send(event.getKafkaTopic(), event.getEntityId(), event).get(3, SECONDS);
            } catch (Exception e) {
                kafkaError = e;
            }
        }
 
        // Fail if the currently-primary broker fails
        // During early phases: RabbitMQ is primary
        // During later phases: Kafka is primary
        boolean kafkaIsPrimary = kafkaPublishEnabled && !rabbitPublishEnabled;
        if (kafkaIsPrimary && kafkaError != null) {
            throw new EventPublishException("Kafka (primary) publish failed", kafkaError);
        }
        if (!kafkaIsPrimary && rabbitError != null) {
            throw new EventPublishException("RabbitMQ (primary) publish failed", rabbitError);
        }
 
        // Log if secondary broker failed (alert but don't fail the request)
        if (rabbitError != null) log.warn("RabbitMQ (secondary) publish failed: {}", rabbitError.getMessage());
        if (kafkaError != null) log.warn("Kafka (secondary) publish failed: {}", kafkaError.getMessage());
    }
}

// Financial settlement systems batch transactions for clearing house submission
// Window: 09:00 to 16:30 (market hours)
// Settlement: submitted at end of day for next-day clearing
 
@Service
public class SettlementWindowConsumer {
 
    @KafkaListener(topics = "payments.events.transaction-lifecycle",
                   groupId = "settlement-aggregator")
    public void accumulateForSettlement(TransactionEvent event) {
        // Calculate which settlement window this belongs to
        SettlementWindow window = windowCalculator.calculateWindow(event.getTimestamp());
 
        if (window.isClosed()) {
            // Transaction arrived after settlement window closed
            // Route to next-day settlement
            kafkaTemplate.send("payments.commands.next-day-settlement", event.getAccountId(), event);
            return;
        }
 
        // Accumulate into settlement batch for this window
        settlementBatchRepository.addTransaction(window.getWindowId(), event);
    }
 
    // Scheduled: end of each settlement window
    @Scheduled(cron = "0 30 16 * * MON-FRI")  // 4:30pm weekdays
    public void submitSettlementBatch() {
        SettlementWindow closedWindow = windowCalculator.closeCurrentWindow();
        List<TransactionEvent> batch = settlementBatchRepository.getAll(closedWindow.getWindowId());
 
        clearingHouseClient.submitBatch(SettlementBatch.of(closedWindow, batch));
        log.info("Settlement batch submitted: {} transactions, window: {}",
            batch.size(), closedWindow.getWindowId());
    }
}

// Healthcare events contain PHI (Protected Health Information)
// Must meet HIPAA audit, encryption, and access control requirements
 
@Configuration
public class HIPAACompliantKafkaConfig {
 
    @Bean
    public ProducerFactory<String, PatientEvent> hipaaProducerFactory() {
        Map<String, Object> config = new HashMap<>();
 
        // Encryption in transit (required)
        config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
 
        // Encryption at rest: Kafka log dirs on LUKS-encrypted volumes (broker config)
 
        // Field-level encryption for PHI fields (defense in depth)
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, HIIPAAEncryptingSerializer.class);
        config.put("hipaa.field.encryption.key.id", "${HIPAA_KEY_ID}");
        config.put("hipaa.field.encryption.kms.url", "${KMS_URL}");
 
        return new DefaultKafkaProducerFactory<>(config);
    }
}
 
@Service
public class PatientEventPublisher {
 
    public void publishAppointmentBooked(AppointmentBookedEvent event) {
        // Encrypt PHI fields before publishing
        AppointmentBookedEvent encrypted = AppointmentBookedEvent.builder()
            .appointmentId(event.getAppointmentId())  // Not PHI - not encrypted
            .patientIdHash(hashPHI(event.getPatientId()))  // Hash for correlation
            .encryptedPatientName(kmsEncrypt(event.getPatientName()))
            .encryptedDateOfBirth(kmsEncrypt(event.getDateOfBirth()))
            .encryptedDiagnosis(kmsEncrypt(event.getDiagnosis()))
            .appointmentType(event.getAppointmentType())  // Not PHI
            .timestamp(event.getTimestamp())
            .hipaaRequestId(hipaaAuditLogger.logAccess(event)) // Audit every access
            .build();
 
        kafkaTemplate.send("healthcare.events.appointments", event.getAppointmentId(), encrypted);
    }
}

// Live class session: instructor actions must be synchronized to all students in < 200ms
 
@Service
public class LiveSessionEventBus {
 
    // Instructor actions published here
    @KafkaListener(topics = "elearning.events.session-actions",
                   groupId = "session-broadcast")
    public void broadcastToStudents(SessionActionEvent event) {
        // Fetch all active session participants
        List<String> participantIds = sessionParticipantCache.get(event.getSessionId());
 
        // Publish individual student notification events
        for (String participantId : participantIds) {
            kafkaTemplate.send("elearning.events.student-actions", participantId,
                StudentActionEvent.of(participantId, event));
        }
    }
 
    // Student WebSocket consumers receive their individual events
    @KafkaListener(topics = "elearning.events.student-actions",
                   groupId = "websocket-delivery")
    public void deliverToStudent(StudentActionEvent event) {
        webSocketSessionManager.send(event.getStudentId(), event);
        // Sub-100ms end-to-end for online students
    }
}

-- Outbox table schema
CREATE TABLE order_outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(50) NOT NULL,  -- 'ORDER'
    aggregate_id VARCHAR(36) NOT NULL,     -- orderId
    event_type VARCHAR(100) NOT NULL,      -- 'OrderPlaced'
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at TIMESTAMPTZ,              -- Null = not yet published
    INDEX idx_outbox_unpublished (published_at) WHERE published_at IS NULL
);

kafkaTemplate.send("orders.events.order-lifecycle", order.getId(), event);
// hash(orderId) % partitionCount = partition assignment
// All events for the same orderId always go to the same partition

// Each service defines its own consumer group
@KafkaListener(topics = "orders.events.order-lifecycle",
               groupId = "inventory-service")   // Dedicated group
public void handleForInventory(OrderPlacedEvent event) { ... }
 
@KafkaListener(topics = "orders.events.order-lifecycle",
               groupId = "email-notification-service") // Dedicated group
public void handleForEmail(OrderPlacedEvent event) { ... }