Event-driven architecture is a paradigm where components communicate through events rather than direct calls. Services emit events when state changes, enabling loose coupling, scalability, and asynchronous processing across distributed systems.

// Event class
public class OrderCreatedEvent extends ApplicationEvent {
    private String orderId;
    private double amount;
    
    public OrderCreatedEvent(Object source, String orderId, double amount) {
        super(source);
        this.orderId = orderId;
        this.amount = amount;
    }
}

// Publisher
@Service
public class OrderService {
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    public void createOrder(String orderId, double amount) {
        // Business logic
        eventPublisher.publishEvent(
            new OrderCreatedEvent(this, orderId, amount));
    }
}

// Listener
@Component
public class OrderListener {
    @EventListener
    public void onOrderCreated(OrderCreatedEvent event) {
        System.out.println('Processing order: ' + event.getOrderId());
        // Send email, update inventory, etc.
    }
}

// Benefits:
// 1. Loose coupling - no direct dependencies
// 2. Asynchronous - non-blocking execution
// 3. Scalability - easy to add new listeners
// 4. Event sourcing - audit trail of all events
// 5. Time travel - replay events to reconstruct state

Apache Kafka is a distributed event streaming platform for building real-time data pipelines. Spring Boot integrates via Spring Cloud Stream or spring-kafka, enabling producer/consumer patterns for high-throughput, fault-tolerant message processing.

// Maven dependency
// spring-kafka or spring-cloud-stream-kafka

// Kafka Producer
@Service
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message)
            .addCallback(
                result -> System.out.println('Sent: ' + message),
                ex -> System.err.println('Error: ' + ex.getMessage())
            );
    }
}

// Kafka Consumer
@Service
public class KafkaConsumer {
    @KafkaListener(topics = 'orders', groupId = 'order-service')
    public void consumeMessage(String message) {
        System.out.println('Received: ' + message);
        // Process message
    }
}

// With Spring Cloud Stream
@Configuration
public class KafkaStreamConfig {
    @Bean
    public Function<String, String> process() {
        return input -> {
            System.out.println('Processing: ' + input);
            return input.toUpperCase();
        };
    }
}

// Kafka concepts:
// Topic - category of events
// Partition - parallel processing unit
// Consumer Group - multiple consumers for same topic
// Offset - position in partition
// Replication - fault tolerance

// Key features:
// 1. High throughput - millions of messages/sec
// 2. Durability - persisted to disk
// 3. Scalability - add brokers dynamically
// 4. Stream processing - Kafka Streams API
// 5. Exactly-once semantics - no duplicate processing

RabbitMQ is a traditional message broker using AMQP protocol for queuing. Unlike Kafka (event streaming), RabbitMQ is optimized for point-to-point messaging, request-reply patterns, and guaranteed delivery with less throughput but simpler semantics.

// RabbitMQ Configuration
@Configuration
public class RabbitMQConfig {
    public static final String QUEUE = 'orders.queue';
    public static final String EXCHANGE = 'orders.exchange';
    public static final String ROUTING_KEY = 'order.#';
    
    @Bean
    public Queue queue() {
        return new Queue(QUEUE);
    }
    
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(EXCHANGE);
    }
    
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue())
            .to(exchange())
            .with(ROUTING_KEY);
    }
}

// RabbitMQ Producer
@Service
public class RabbitProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend(
            'orders.exchange', 'order.created', message);
    }
}

// RabbitMQ Consumer
@Service
public class RabbitConsumer {
    @RabbitListener(queues = 'orders.queue')
    public void consumeMessage(String message) {
        System.out.println('Received: ' + message);
    }
}

// RabbitMQ patterns:
// 1. Work queues - load balancing
// 2. Pub/Sub - multiple consumers
// 3. Routing - topic-based
// 4. RPC - request-reply

// When to use RabbitMQ:
// - Task queues (email, notifications)
// - Request-reply patterns
// - Complex routing
// - Guaranteed delivery
// - Lower throughput acceptable

Spring Cloud Stream provides a unified programming model for building message-driven microservices, abstracting underlying broker details (Kafka, RabbitMQ, etc.). Use suppliers, functions, and consumers to decouple from specific implementations.

// Spring Cloud Stream with Kafka/RabbitMQ abstraction

// application.yml
spring:
  cloud:
    stream:
      bindings:
        input:
          destination: orders
          group: order-processor
        output:
          destination: notifications
      kafka:  # or rabbit
        binder:
          brokers: localhost:9092

// Supplier (Producer)
@Configuration
public class OrderProducer {
    @Bean
    public Supplier<Order> orderSupplier() {
        return () -> new Order('ORD-001', 100.0);
    }
}

// Function (Processor)
@Configuration
public class OrderProcessor {
    @Bean
    public Function<Order, Notification> processOrder() {
        return order -> {
            System.out.println('Processing: ' + order.getId());
            return new Notification(
                'Order ' + order.getId() + ' confirmed');
        };
    }
}

// Consumer
@Configuration
public class NotificationSender {
    @Bean
    public Consumer<Notification> sendNotification() {
        return notification -> {
            System.out.println('Sending: ' + notification.getMessage());
        };
    }
}

// Switch between Kafka and RabbitMQ with config only:
// Change spring.cloud.stream.default-binder: kafka|rabbit

// Benefits of Spring Cloud Stream:
// 1. Broker agnostic - switch easily
// 2. Functional programming model
// 3. Testable - easy mocking
// 4. Built-in error handling
// 5. Dead letter queue support

// Partitioning for parallel processing
spring:
  cloud:
    stream:
      bindings:
        input:
          consumer:
            partitioned: true
          instanceCount: 3
          instanceIndex: 0

Saga pattern manages long-running distributed transactions by breaking them into local transactions coordinated through events. Two approaches: choreography (events trigger actions) and orchestration (central coordinator). Handles failures with compensating transactions.

// Saga Pattern - Orchestration approach

// Events
public class OrderCreatedEvent {
    private String orderId;
    private double amount;
}

public class PaymentProcessedEvent {
    private String orderId;
    private String paymentId;
}

public class OrderFailedEvent {
    private String orderId;
    private String reason;
}

// Order Service
@Service
public class OrderService {
    @Autowired
    private KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;
    
    public void createOrder(Order order) {
        // Create order
        kafkaTemplate.send('orders', new OrderCreatedEvent(
            order.getId(), order.getAmount()));
    }
}

// Payment Service (listens to OrderCreatedEvent)
@Service
public class PaymentService {
    @KafkaListener(topics = 'orders', groupId = 'payment-service')
    public void processPayment(OrderCreatedEvent event) {
        try {
            // Process payment
            kafkaTemplate.send('payments', new PaymentProcessedEvent(
                event.getOrderId(), 'PAY-123'));
        } catch (Exception e) {
            // Publish compensation event
            kafkaTemplate.send('order-failures', new OrderFailedEvent(
                event.getOrderId(), e.getMessage()));
        }
    }
}

// Saga Orchestrator (centralized coordinator)
@Service
public class OrderSagaOrchestrator {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    public void startSaga(Order order) {
        // Step 1: Reserve inventory
        kafkaTemplate.send('inventory', order);
    }
    
    @KafkaListener(topics = 'inventory-reserved')
    public void onInventoryReserved(InventoryReservedEvent event) {
        // Step 2: Process payment
        kafkaTemplate.send('payments', event.getOrder());
    }
    
    @KafkaListener(topics = 'payment-failed')
    public void onPaymentFailed(PaymentFailedEvent event) {
        // Step 3: Compensate - release inventory
        kafkaTemplate.send('inventory', new ReleaseInventoryCommand(
            event.getOrderId()));
    }
}

// Saga Pattern Benefits:
// 1. Handles distributed transactions
// 2. Ensures data consistency
// 3. Handles failures gracefully
// 4. No 2PC (two-phase commit) needed
// 5. Event-driven architecture

// Two approaches:
// Choreography - services listen and react
// Orchestration - central orchestrator coordinates

@EventListener is Spring's built-in in-process event handling mechanism using ApplicationEventPublisher. Lightweight for local communication but synchronous by default. Use with @Async for async handling, or choose Kafka/RabbitMQ for distributed systems.

// Local In-Process Events

// Event definition
public class UserRegisteredEvent extends ApplicationEvent {
    private String email;
    
    public UserRegisteredEvent(Object source, String email) {
        super(source);
        this.email = email;
    }
    
    public String getEmail() {
        return email;
    }
}

// Event publisher
@Service
public class UserService {
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    public void registerUser(String email) {
        // Register user logic
        eventPublisher.publishEvent(new UserRegisteredEvent(this, email));
    }
}

// Event listeners
@Component
public class EmailNotificationListener {
    @EventListener
    public void onUserRegistered(UserRegisteredEvent event) {
        System.out.println('Sending email to: ' + event.getEmail());
        // Send email
    }
}

@Component
public class AnalyticsListener {
    @EventListener
    public void onUserRegistered(UserRegisteredEvent event) {
        System.out.println('Logging user signup');
        // Log to analytics
    }
}

// Async event handling
@Component
public class SlowTaskListener {
    @EventListener
    @Async
    public void onUserRegistered(UserRegisteredEvent event) {
        // Runs in separate thread
        System.out.println('Processing long task for: ' + event.getEmail());
        // Heavy operation
    }
}

// Conditional listeners
@Component
public class ConditionalListener {
    @EventListener(condition = "#event.email.endsWith('@company.com')")
    public void onInternalUserRegistered(UserRegisteredEvent event) {
        System.out.println('Internal user registered');
    }
}

// Comparison:
// @EventListener: In-process, synchronous, JVM-local
// Kafka/RabbitMQ: Distributed, async, network-based

// Use @EventListener for:
// 1. Simple internal communication
// 2. Loosely coupled components
// 3. When all services in same JVM
// 4. Local event broadcasting

// Use Kafka/RabbitMQ for:
// 1. Microservices communication
// 2. Event sourcing
// 3. Cross-service notifications
// 4. Message persistence needed

Pub-Sub is a messaging pattern where publishers send messages to topics without knowing subscribers, and subscribers receive all messages from subscribed topics. Enables loose coupling and one-to-many communication in distributed systems.

// Spring Cloud Stream Pub-Sub Example

// Publisher
@RestController
@RequestMapping('/api/orders')
public class OrderPublisher {
    @Autowired
    private StreamBridge streamBridge;
    
    @PostMapping
    public void createOrder(@RequestBody Order order) {
        // Publish to 'order-events' destination
        streamBridge.send('order-events', order);
    }
}

// Subscriber 1: Inventory Service
@Service
public class InventoryService {
    @Bean
    public Consumer<Order> inventoryProcessor() {
        return order -> {
            System.out.println('Updating inventory for order: ' + 
                order.getId());
            // Update inventory
        };
    }
}

// Subscriber 2: Notification Service
@Service
public class NotificationService {
    @Bean
    public Consumer<Order> notificationProcessor() {
        return order -> {
            System.out.println('Sending notification for order: ' + 
                order.getId());
            // Send email/SMS
        };
    }
}

// Subscriber 3: Analytics Service
@Service
public class AnalyticsService {
    @Bean
    public Consumer<Order> analyticsProcessor() {
        return order -> {
            System.out.println('Recording order metric: ' + 
                order.getId());
            // Log metrics
        };
    }
}

// Configuration
spring:
  cloud:
    function:
      definition: inventoryProcessor;notificationProcessor;analyticsProcessor
    stream:
      bindings:
        inventoryProcessor-in-0:
          destination: order-events
          group: inventory-group
        notificationProcessor-in-0:
          destination: order-events
          group: notification-group
        analyticsProcessor-in-0:
          destination: order-events
          group: analytics-group

// Pub-Sub guarantees:
// 1. Fan-out - single message to multiple subscribers
// 2. Loose coupling - publishers don't know subscribers
// 3. Scalability - add subscribers without changing publisher
// 4. Independent processing - each subscriber own pace
// 5. Message durability - queue persistence

Dead Letter Queues hold messages that failed processing after retries. Spring Cloud Stream provides @ServiceActivator for DLQ handling, error recovery, and retry policies. Essential for production message processing reliability.

// Dead Letter Queue Configuration

@Configuration
public class DLQConfiguration {
    
    // Main queue
    @Bean
    public Queue mainQueue() {
        return new Queue('order.queue');
    }
    
    // Dead letter queue
    @Bean
    public Queue deadLetterQueue() {
        return new Queue('order.dlq');
    }
    
    @Bean
    public TopicExchange dlxExchange() {
        return new TopicExchange('order.dlx');
    }
    
    @Bean
    public Binding dlBinding() {
        return BindingBuilder.bind(deadLetterQueue())
            .to(dlxExchange())
            .with('*');
    }
}

// Consumer with DLQ handling
@Service
public class OrderProcessor {
    
    @RabbitListener(queues = 'order.queue')
    public void processOrder(Order order) throws Exception {
        try {
            if (order.getAmount() <= 0) {
                throw new InvalidOrderException('Invalid amount');
            }
            System.out.println('Processing order: ' + order.getId());
        } catch (Exception e) {
            // Will be sent to DLQ after max retries
            throw e;
        }
    }
    
    @RabbitListener(queues = 'order.dlq')
    public void handleFailedOrder(Order order) {
        System.err.println('Order failed: ' + order.getId());
        // Log, alert, manual intervention
    }
}

// Spring Cloud Stream DLQ
spring:
  cloud:
    stream:
      kafka:
        bindings:
          input:
            consumer:
              enable-dlq: true
              dlq-name: order.dlq
              max-attempts: 3
              backoff-initial-delay: 1000
              backoff-max-delay: 10000
              backoff-multiplier: 2.0

// Error handler
@Component
public class OrderErrorHandler {
    @ServiceActivator(inputChannel = 'order.errors')
    public void handleError(Message<?> message) {
        System.err.println('Error: ' + message.getPayload());
        // Send to DLQ, alert administrators
    }
}

// Retry configuration
@Configuration
public class RetryConfiguration {
    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();
        
        ExponentialBackOffPolicy backOffPolicy = 
            new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(1000);
        backOffPolicy.setMaxInterval(10000);
        backOffPolicy.setMultiplier(2.0);
        retryTemplate.setBackOffPolicy(backOffPolicy);
        
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);
        retryTemplate.setRetryPolicy(retryPolicy);
        
        return retryTemplate;
    }
}

// Benefits of DLQ:
// 1. Message preservation - don't lose failed messages
// 2. Debugging - analyze failure causes
// 3. Monitoring - alert on DLQ messages
// 4. Recovery - reprocess after fixing issues
// 5. Separation - keep main queue clean

Message ordering in Kafka depends on partitions - messages in same partition maintain order. Use message key for ordering guarantee. Partitioning strategy determines throughput vs order: single partition (strict order but slow) vs multiple partitions (parallel but order within partition only).

// Kafka Partitioning and Ordering

// Strategy 1: Use message key for ordering
@Service
public class OrderEventProducer {
    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;
    
    public void publishOrderEvent(OrderEvent event) {
        // Key is customerId - ensures all orders for customer in same partition
        kafkaTemplate.send(
            'order-events',
            event.getCustomerId(),  // Key - orders from same customer ordered
            event
        );
    }
}

// Strategy 2: Configure partition count
spring:
  kafka:
    topic:
      order-events:
        partitions: 3  # 3 partitions for parallel processing
        replication-factor: 1

// Consumer with partition assignment
@Service
public class OrderEventConsumer {
    @KafkaListener(
        topics = 'order-events',
        groupId = 'order-service',
        concurrency = '3'  # One thread per partition
    )
    public void consumeOrderEvent(OrderEvent event) {
        System.out.println('Consumed: ' + event.getId() + 
            ' from partition');
    }
}

// Ordering guarantees in Kafka:
// 1. Single partition (0) -> strict ordering (bottleneck)
// 2. Multiple partitions -> ordering within partition
// 3. Key-based -> messages with same key go to same partition

public class OrderingExample {
    public static void main(String[] args) {
        // Scenario: Customer orders 3 items
        // Order 1: Order -> Payment -> Shipping (must be in order)
        
        // Use customerId as key
        // customerId='C001' -> always same partition (e.g., partition 0)
        // Order -> Payment -> Shipping guaranteed in order
        // 
        // Meanwhile customerId='C002' -> partition 1
        // Parallel processing, but C002's orders also ordered
    }
}

// Partitioning best practices:
// 1. Choose key wisely (customer, tenant, account)
// 2. Balance partitions with consumers
// 3. Monitor partition lag
// 4. Consider rebalancing impact
// 5. Test throughput vs ordering requirements

// Tools for monitoring
@Component
public class KafkaMetricsMonitor {
    @Autowired
    private MeterRegistry meterRegistry;
    
    public void trackConsumerLag(String topic, String group) {
        // Monitor lag per partition
        // Alert if lag exceeds threshold
    }
}

Event Sourcing stores application state as immutable sequence of events. Instead of storing current state, store all state changes. Enables time-travel, audit trails, and event replay. Combine with Kafka and event store database for implementation.

// Event Sourcing Implementation

// Event definitions
public abstract class DomainEvent {
    private String aggregateId;
    private long timestamp;
    private int version;
    
    // getters
}

public class AccountCreatedEvent extends DomainEvent {
    private String accountNumber;
    private String accountHolder;
    
    public AccountCreatedEvent(String aggregateId, String accountNumber) {
        this.aggregateId = aggregateId;
        this.accountNumber = accountNumber;
    }
}

public class MoneyDepositedEvent extends DomainEvent {
    private double amount;
    
    public MoneyDepositedEvent(String aggregateId, double amount) {
        this.aggregateId = aggregateId;
        this.amount = amount;
    }
}

// Event Store
@Repository
public class EventStore {
    @Autowired
    private EventRepository eventRepository;
    
    public void saveEvent(DomainEvent event) {
        eventRepository.save(event);
    }
    
    public List<DomainEvent> getEvents(String aggregateId) {
        return eventRepository.findByAggregateId(aggregateId);
    }
}

// Aggregate (Account) built from events
@Service
public class AccountService {
    @Autowired
    private EventStore eventStore;
    
    @Autowired
    private KafkaTemplate<String, DomainEvent> kafkaTemplate;
    
    public void createAccount(String accountNumber) {
        AccountCreatedEvent event = new AccountCreatedEvent(
            UUID.randomUUID().toString(), accountNumber);
        
        // Save to event store
        eventStore.saveEvent(event);
        
        // Publish event
        kafkaTemplate.send('account-events', event);
    }
    
    public void deposit(String accountId, double amount) {
        // Verify account exists by replaying events
        List<DomainEvent> events = eventStore.getEvents(accountId);
        Account account = rebuildAccount(accountId, events);
        
        if (account == null) {
            throw new AccountNotFoundException();
        }
        
        // Create deposit event
        MoneyDepositedEvent event = new MoneyDepositedEvent(accountId, amount);
        eventStore.saveEvent(event);
        kafkaTemplate.send('account-events', event);
    }
    
    // Rebuild current state from events
    private Account rebuildAccount(String accountId, 
            List<DomainEvent> events) {
        Account account = null;
        
        for (DomainEvent event : events) {
            if (event instanceof AccountCreatedEvent) {
                account = new Account();
                account.setId(accountId);
            } else if (event instanceof MoneyDepositedEvent) {
                account.deposit(((MoneyDepositedEvent) event).getAmount());
            }
        }
        
        return account;
    }
}

// Read Model (Materialized View)
@Component
public class AccountProjection {
    @KafkaListener(topics = 'account-events', groupId = 'account-projection')
    public void handleEvent(DomainEvent event) {
        if (event instanceof AccountCreatedEvent) {
            // Update read model
        } else if (event instanceof MoneyDepositedEvent) {
            // Update read model
        }
    }
}

// Event Sourcing benefits:
// 1. Complete audit trail - all changes tracked
// 2. Time travel - reconstruct past states
// 3. Debugging - replay events to understand issue
// 4. Scalability - separate read/write models
// 5. Event-driven - react to domain events

// Challenges:
// 1. Event schema evolution
// 2. Event store maintenance
// 3. Eventual consistency
// 4. Debugging complexity