Event-Driven Architecture at Scale: Production Patterns for Distributed Systems

// Traditional: Imperative, Coupled Approach
@Service
public class OrderService {
    @Autowired
    private PaymentService paymentService;
    @Autowired
    private InventoryService inventoryService;
    @Autowired
    private ShippingService shippingService;
    @Autowired
    private EmailService emailService;
    
    @Transactional
    public Order createOrder(OrderRequest request) {
        // Tightly coupled, synchronous calls
        Order order = new Order(request);
        
        PaymentResult payment = paymentService.processPayment(request.getPayment());
        if (!payment.isSuccessful()) {
            throw new PaymentFailedException();
        }
        
        inventoryService.reserveItems(order.getItems());
        ShippingLabel label = shippingService.createShipping(order);
        emailService.sendOrderConfirmation(order);
        
        return orderRepository.save(order);
    }
}

// Event-Driven: Reactive, Decoupled Approach
@Service
public class OrderService {
    @Autowired
    private EventPublisher eventPublisher;
    
    public Order createOrder(OrderRequest request) {
        // Simply create order and publish event
        Order order = new Order(request);
        order = orderRepository.save(order);
        
        // Other services react to this event independently
        eventPublisher.publish(new OrderCreatedEvent(
            order.getId(),
            order.getCustomerId(),
            order.getItems(),
            order.getTotalAmount()
        ));
        
        return order;
    }
}

// Represent something that happened in your business domain
public record OrderShippedEvent(
    String orderId,
    String trackingNumber,
    Instant shippedAt,
    Address destination
) implements DomainEvent {
    
    @Override
    public String getAggregateId() {
        return orderId;
    }
    
    @Override
    public String getEventType() {
        return "order.shipped";
    }
}

// For communication between bounded contexts
public record InventoryReservedEvent(
    String reservationId,
    String orderId,
    Map<String, Integer> items,
    Instant expiresAt
) implements IntegrationEvent {
    
    @Override
    public String getSourceContext() {
        return "inventory";
    }
    
    @Override
    public String getTargetContext() {
        return "order";
    }
}

// Infrastructure and operational events
public record ServiceHealthEvent(
    String serviceId,
    HealthStatus status,
    Map<String, Object> metrics,
    Instant timestamp
) implements SystemEvent {
    
    @Override
    public Priority getPriority() {
        return status == HealthStatus.CRITICAL ? 
            Priority.HIGH : Priority.NORMAL;
    }
}

@Configuration
@EnableKafka
public class KafkaConfiguration {
    
    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        
        // Production optimizations
        configs.put(ProducerConfig.ACKS_CONFIG, "all"); // Wait for all replicas
        configs.put(ProducerConfig.RETRIES_CONFIG, 3);
        configs.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        configs.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        configs.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        configs.put(ProducerConfig.IDEMPOTENCE_CONFIG, true); // Exactly-once semantics
        
        return new DefaultKafkaProducerFactory<>(configs);
    }
    
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        configs.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
        
        // Consumer optimizations for scale
        configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // Manual commit
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        configs.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        
        return new DefaultKafkaConsumerFactory<>(configs);
    }
}

@Component
@Slf4j
public class HighPerformanceEventProducer {
    
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final MeterRegistry meterRegistry;
    private final CircuitBreaker circuitBreaker;
    
    public HighPerformanceEventProducer(
            KafkaTemplate<String, Object> kafkaTemplate,
            MeterRegistry meterRegistry) {
        this.kafkaTemplate = kafkaTemplate;
        this.meterRegistry = meterRegistry;
        this.circuitBreaker = CircuitBreaker.ofDefaults("kafka-producer");
        
        // Configure circuit breaker
        circuitBreaker.getEventPublisher()
            .onStateTransition(event -> 
                log.warn("Circuit breaker state transition: {}", event));
    }
    
    public CompletableFuture<SendResult<String, Object>> publish(
            String topic, 
            String key, 
            Object event) {
        
        return circuitBreaker.executeSupplier(() -> {
            // Add headers for tracing and routing
            ProducerRecord<String, Object> record = new ProducerRecord<>(topic, key, event);
            record.headers()
                .add("X-Event-Type", event.getClass().getSimpleName().getBytes())
                .add("X-Correlation-Id", MDC.get("correlationId").getBytes())
                .add("X-Timestamp", String.valueOf(System.currentTimeMillis()).getBytes());
            
            // Track metrics
            Timer.Sample sample = Timer.start(meterRegistry);
            
            return kafkaTemplate.send(record)
                .whenComplete((result, ex) -> {
                    sample.stop(Timer.builder("events.published")
                        .tag("topic", topic)
                        .tag("type", event.getClass().getSimpleName())
                        .tag("status", ex == null ? "success" : "failure")
                        .register(meterRegistry));
                    
                    if (ex != null) {
                        log.error("Failed to publish event to topic {}: {}", 
                            topic, ex.getMessage());
                        meterRegistry.counter("events.publish.failures",
                            "topic", topic).increment();
                    }
                });
        });
    }
    
    // Batch publishing for high throughput
    public CompletableFuture<List<SendResult<String, Object>>> publishBatch(
            String topic, 
            List<Pair<String, Object>> events) {
        
        List<CompletableFuture<SendResult<String, Object>>> futures = 
            events.stream()
                .map(pair -> publish(topic, pair.getKey(), pair.getValue()))
                .collect(Collectors.toList());
                
        return CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        ).thenApply(v -> 
            futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
        );
    }
}

@Component
@Slf4j
public class ResilientEventConsumer {
    
    private final EventProcessor eventProcessor;
    private final DeadLetterPublisher deadLetterPublisher;
    private final MeterRegistry meterRegistry;
    
    @KafkaListener(
        topics = "${kafka.topics.orders}",
        containerFactory = "kafkaListenerContainerFactory",
        concurrency = "${kafka.consumer.concurrency:3}"
    )
    @Retryable(
        value = {RetryableException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    public void consume(
            @Payload String payload,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment acknowledgment) {
        
        Timer.Sample sample = Timer.start(meterRegistry);
        
        try {
            // Process event
            Event event = parseEvent(payload);
            validateEvent(event);
            
            ProcessingResult result = eventProcessor.process(event);
            
            if (result.requiresRetry()) {
                throw new RetryableException("Processing requires retry");
            }
            
            // Manual commit after successful processing
            acknowledgment.acknowledge();
            
            sample.stop(Timer.builder("events.processed")
                .tag("topic", topic)
                .tag("status", "success")
                .register(meterRegistry));
                
        } catch (RetryableException e) {
            // Will be retried by Spring Retry
            throw e;
            
        } catch (Exception e) {
            // Non-retryable error - send to DLQ
            log.error("Failed to process event from topic {} partition {} offset {}: {}",
                topic, partition, offset, e.getMessage());
                
            deadLetterPublisher.publish(
                new DeadLetterEvent(topic, partition, offset, payload, e.getMessage())
            );
            
            // Acknowledge to move forward
            acknowledgment.acknowledge();
            
            sample.stop(Timer.builder("events.processed")
                .tag("topic", topic)
                .tag("status", "dead-lettered")
                .register(meterRegistry));
        }
    }
    
    @Recover
    public void recover(RetryableException e, @Payload String payload) {
        log.error("Event processing failed after retries: {}", e.getMessage());
        deadLetterPublisher.publish(
            new DeadLetterEvent("retry-exhausted", payload, e.getMessage())
        );
    }
}

@Component
public class EventStore {
    
    private final EventRepository eventRepository;
    private final SnapshotRepository snapshotRepository;
    private final EventPublisher eventPublisher;
    
    @Transactional
    public void save(String aggregateId, List<DomainEvent> events, Long expectedVersion) {
        // Optimistic locking check
        Long currentVersion = eventRepository.getLatestVersion(aggregateId);
        if (!Objects.equals(currentVersion, expectedVersion)) {
            throw new ConcurrencyException(
                "Expected version %d but was %d".formatted(expectedVersion, currentVersion)
            );
        }
        
        // Store events
        Long version = currentVersion == null ? 0L : currentVersion;
        for (DomainEvent event : events) {
            version++;
            
            EventEntity entity = EventEntity.builder()
                .aggregateId(aggregateId)
                .eventType(event.getClass().getSimpleName())
                .eventData(serialize(event))
                .eventVersion(version)
                .timestamp(Instant.now())
                .build();
                
            eventRepository.save(entity);
            
            // Publish for projections
            eventPublisher.publish(event);
        }
        
        // Check if snapshot needed
        if (version % SNAPSHOT_FREQUENCY == 0) {
            createSnapshot(aggregateId, version);
        }
    }
    
    public <T extends AggregateRoot> T load(String aggregateId, Class<T> aggregateClass) {
        // Try to load from snapshot
        Optional<SnapshotEntity> snapshot = 
            snapshotRepository.findLatest(aggregateId);
            
        T aggregate;
        Long fromVersion;
        
        if (snapshot.isPresent()) {
            aggregate = deserialize(snapshot.get().getData(), aggregateClass);
            fromVersion = snapshot.get().getVersion();
        } else {
            try {
                aggregate = aggregateClass.getDeclaredConstructor().newInstance();
                fromVersion = 0L;
            } catch (Exception e) {
                throw new RuntimeException("Failed to create aggregate", e);
            }
        }
        
        // Replay events from snapshot
        List<EventEntity> events = eventRepository.findByAggregateIdAndVersionGreaterThan(
            aggregateId, fromVersion
        );
        
        for (EventEntity event : events) {
            DomainEvent domainEvent = deserialize(event.getEventData(), event.getEventType());
            aggregate.apply(domainEvent);
        }
        
        aggregate.markEventsAsCommitted();
        return aggregate;
    }
}

@Component
@EventListener
public class OrderProjectionHandler {
    
    private final OrderReadModelRepository readModelRepository;
    private final ProjectionCheckpointService checkpointService;
    
    @Async
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void on(OrderCreatedEvent event) {
        updateProjection(() -> {
            OrderReadModel readModel = new OrderReadModel();
            readModel.setOrderId(event.getOrderId());
            readModel.setCustomerId(event.getCustomerId());
            readModel.setStatus("CREATED");
            readModel.setTotalAmount(event.getTotalAmount());
            readModel.setCreatedAt(event.getTimestamp());
            
            readModelRepository.save(readModel);
        }, event);
    }
    
    @Async
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void on(OrderShippedEvent event) {
        updateProjection(() -> {
            OrderReadModel readModel = readModelRepository.findById(event.getOrderId())
                .orElseThrow(() -> new ProjectionException("Order not found"));
                
            readModel.setStatus("SHIPPED");
            readModel.setTrackingNumber(event.getTrackingNumber());
            readModel.setShippedAt(event.getShippedAt());
            
            readModelRepository.save(readModel);
        }, event);
    }
    
    private void updateProjection(Runnable projectionLogic, DomainEvent event) {
        try {
            projectionLogic.run();
            checkpointService.saveCheckpoint(
                this.getClass().getSimpleName(),
                event.getEventId(),
                event.getTimestamp()
            );
        } catch (Exception e) {
            log.error("Projection failed for event {}: {}", 
                event.getEventId(), e.getMessage());
            throw new ProjectionException("Projection failed", e);
        }
    }
}

@Component
@Slf4j
public class OrderSagaOrchestrator {
    
    private final Map<String, SagaStep> steps;
    private final SagaStateRepository stateRepository;
    private final EventPublisher eventPublisher;
    
    public OrderSagaOrchestrator() {
        this.steps = Map.of(
            "RESERVE_INVENTORY", new ReserveInventoryStep(),
            "PROCESS_PAYMENT", new ProcessPaymentStep(),
            "CREATE_SHIPPING", new CreateShippingStep(),
            "SEND_CONFIRMATION", new SendConfirmationStep()
        );
    }
    
    @Transactional
    public void startSaga(OrderSagaData sagaData) {
        // Create saga state
        SagaState state = SagaState.builder()
            .sagaId(UUID.randomUUID().toString())
            .sagaType("ORDER_FULFILLMENT")
            .currentStep("RESERVE_INVENTORY")
            .data(serialize(sagaData))
            .status(SagaStatus.RUNNING)
            .startTime(Instant.now())
            .build();
            
        stateRepository.save(state);
        
        // Start first step
        executeStep(state);
    }
    
    @EventListener
    public void handleStepCompleted(SagaStepCompletedEvent event) {
        SagaState state = stateRepository.findById(event.getSagaId())
            .orElseThrow(() -> new SagaException("Saga not found"));
            
        // Move to next step
        String nextStep = getNextStep(state.getCurrentStep());
        if (nextStep != null) {
            state.setCurrentStep(nextStep);
            state.addCompletedStep(event.getStepName());
            stateRepository.save(state);
            
            executeStep(state);
        } else {
            // Saga completed successfully
            completeSaga(state);
        }
    }
    
    @EventListener
    public void handleStepFailed(SagaStepFailedEvent event) {
        SagaState state = stateRepository.findById(event.getSagaId())
            .orElseThrow(() -> new SagaException("Saga not found"));
            
        log.error("Saga {} failed at step {}: {}", 
            event.getSagaId(), event.getStepName(), event.getReason());
            
        // Start compensation
        startCompensation(state);
    }
    
    private void executeStep(SagaState state) {
        SagaStep step = steps.get(state.getCurrentStep());
        OrderSagaData sagaData = deserialize(state.getData(), OrderSagaData.class);
        
        try {
            StepResult result = step.execute(sagaData);
            
            if (result.isSuccessful()) {
                eventPublisher.publish(new SagaStepCompletedEvent(
                    state.getSagaId(),
                    state.getCurrentStep(),
                    result.getOutput()
                ));
            } else {
                eventPublisher.publish(new SagaStepFailedEvent(
                    state.getSagaId(),
                    state.getCurrentStep(),
                    result.getError()
                ));
            }
        } catch (Exception e) {
            eventPublisher.publish(new SagaStepFailedEvent(
                state.getSagaId(),
                state.getCurrentStep(),
                e.getMessage()
            ));
        }
    }
    
    private void startCompensation(SagaState state) {
        state.setStatus(SagaStatus.COMPENSATING);
        stateRepository.save(state);
        
        // Execute compensation in reverse order
        List<String> completedSteps = new ArrayList<>(state.getCompletedSteps());
        Collections.reverse(completedSteps);
        
        for (String stepName : completedSteps) {
            SagaStep step = steps.get(stepName);
            OrderSagaData sagaData = deserialize(state.getData(), OrderSagaData.class);
            
            try {
                step.compensate(sagaData);
                log.info("Compensated step {} for saga {}", stepName, state.getSagaId());
            } catch (Exception e) {
                log.error("Failed to compensate step {} for saga {}: {}", 
                    stepName, state.getSagaId(), e.getMessage());
                // Continue with other compensations
            }
        }
        
        state.setStatus(SagaStatus.COMPENSATED);
        state.setEndTime(Instant.now());
        stateRepository.save(state);
    }
}

@Component
public class OrderSagaParticipant {
    
    @EventListener
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // Reserve inventory
            InventoryReservation reservation = inventoryService.reserve(
                event.getOrderId(),
                event.getItems()
            );
            
            eventPublisher.publish(new InventoryReservedEvent(
                event.getOrderId(),
                reservation.getReservationId(),
                event.getItems()
            ));
        } catch (InsufficientInventoryException e) {
            eventPublisher.publish(new OrderFailedEvent(
                event.getOrderId(),
                "INSUFFICIENT_INVENTORY",
                e.getMessage()
            ));
        }
    }
    
    @EventListener
    public void handleInventoryReserved(InventoryReservedEvent event) {
        try {
            // Process payment
            PaymentResult payment = paymentService.charge(
                event.getOrderId(),
                getOrderAmount(event.getOrderId())
            );
            
            eventPublisher.publish(new PaymentProcessedEvent(
                event.getOrderId(),
                payment.getTransactionId(),
                payment.getAmount()
            ));
        } catch (PaymentException e) {
            // Compensate by releasing inventory
            eventPublisher.publish(new ReleaseInventoryCommand(
                event.getReservationId(),
                event.getOrderId()
            ));
            
            eventPublisher.publish(new OrderFailedEvent(
                event.getOrderId(),
                "PAYMENT_FAILED",
                e.getMessage()
            ));
        }
    }
}

@Component
public class EventPartitioningStrategy {
    
    private final ConsistentHashing consistentHashing;
    
    public int determinePartition(String topic, Object event, int numPartitions) {
        String partitionKey = extractPartitionKey(event);
        
        // Use consistent hashing for even distribution
        return consistentHashing.hash(partitionKey) % numPartitions;
    }
    
    private String extractPartitionKey(Object event) {
        // Partition by aggregate ID for ordering guarantees
        if (event instanceof DomainEvent domainEvent) {
            return domainEvent.getAggregateId();
        }
        
        // Partition by customer ID for customer-centric events
        if (event instanceof CustomerEvent customerEvent) {
            return customerEvent.getCustomerId();
        }
        
        // Default: use event type for even distribution
        return event.getClass().getSimpleName();
    }
}

@Component
public class BatchEventProcessor {
    
    private final BlockingQueue<Event> eventQueue = 
        new LinkedBlockingQueue<>(10000);
    private final ScheduledExecutorService scheduler = 
        Executors.newScheduledThreadPool(2);
    
    @PostConstruct
    public void startBatchProcessor() {
        // Process batches every 100ms or when batch size reached
        scheduler.scheduleWithFixedDelay(
            this::processBatch, 0, 100, TimeUnit.MILLISECONDS
        );
    }
    
    public void submitEvent(Event event) {
        if (!eventQueue.offer(event)) {
            // Queue full - apply backpressure
            throw new BackpressureException("Event queue full");
        }
    }
    
    private void processBatch() {
        List<Event> batch = new ArrayList<>(BATCH_SIZE);
        eventQueue.drainTo(batch, BATCH_SIZE);
        
        if (batch.isEmpty()) {
            return;
        }
        
        try {
            // Group by type for efficient processing
            Map<Class<?>, List<Event>> eventsByType = batch.stream()
                .collect(Collectors.groupingBy(Event::getClass));
                
            // Process each type in parallel
            CompletableFuture<?>[] futures = eventsByType.entrySet().stream()
                .map(entry -> CompletableFuture.runAsync(() -> 
                    processEventType(entry.getKey(), entry.getValue())
                ))
                .toArray(CompletableFuture[]::new);
                
            CompletableFuture.allOf(futures).join();
            
        } catch (Exception e) {
            log.error("Batch processing failed", e);
            // Re-queue events for retry
            batch.forEach(this::submitEvent);
        }
    }
}

@Component
public class EventFlowMonitor {
    
    private final MeterRegistry meterRegistry;
    private final HealthIndicatorRegistry healthRegistry;
    
    @EventListener
    public void monitorEventFlow(ApplicationEvent event) {
        // Track event metrics
        String eventType = event.getClass().getSimpleName();
        
        meterRegistry.counter("events.published", 
            "type", eventType,
            "source", event.getSource().getClass().getSimpleName()
        ).increment();
        
        // Track event processing time
        if (event instanceof TimedEvent timedEvent) {
            meterRegistry.timer("events.processing.time",
                "type", eventType
            ).record(timedEvent.getProcessingTime(), TimeUnit.MILLISECONDS);
        }
    }
    
    @Scheduled(fixedRate = 60000)
    public void checkEventHealth() {
        // Monitor event lag
        kafkaConsumerMetrics.forEach((topic, metrics) -> {
            long lag = metrics.getTotalLag();
            
            if (lag > CRITICAL_LAG_THRESHOLD) {
                healthRegistry.register(
                    "events." + topic,
                    new HealthIndicator() {
                        @Override
                        public Health health() {
                            return Health.down()
                                .withDetail("lag", lag)
                                .withDetail("threshold", CRITICAL_LAG_THRESHOLD)
                                .build();
                        }
                    }
                );
            }
        });
    }
}

@Component
public class EventTracingInterceptor {
    
    @Around("@annotation(EventHandler)")
    public Object traceEventHandling(ProceedingJoinPoint joinPoint) throws Throwable {
        Event event = (Event) joinPoint.getArgs()[0];
        
        Span span = tracer.nextSpan()
            .name("event.handle." + event.getClass().getSimpleName())
            .tag("event.id", event.getEventId())
            .tag("event.type", event.getClass().getSimpleName())
            .tag("handler", joinPoint.getTarget().getClass().getSimpleName())
            .start();
            
        try (Tracer.SpanInScope scope = tracer.withSpanInScope(span)) {
            return joinPoint.proceed();
        } finally {
            span.end();
        }
    }
}