Spring Boot Event Driven Patterns

Spring Boot Event-Driven Patterns

Overview

Implement Event-Driven Architecture (EDA) patterns in Spring Boot 3.x using domain events, ApplicationEventPublisher, @TransactionalEventListener, and distributed messaging with Kafka and Spring Cloud Stream.

When to Use

• Implementing event-driven microservices with Kafka messaging
• Publishing domain events from aggregate roots in DDD architectures
• Setting up transactional event listeners that fire after database commits
• Adding async messaging with producers and consumers via Spring Kafka
• Ensuring reliable event delivery using the transactional outbox pattern
• Replacing synchronous calls with event-based communication between services

Quick Reference

Examples

Monolithic to Event-Driven Refactoring

Before (Anti-Pattern):

@Transactional
public Order processOrder(OrderRequest request) {
    Order order = orderRepository.save(request);
    inventoryService.reserve(order.getItems()); // Blocking
    paymentService.charge(order.getPayment()); // Blocking
    emailService.sendConfirmation(order); // Blocking
    return order;
}

After (Event-Driven):

@Transactional
public Order processOrder(OrderRequest request) {
    Order order = Order.create(request);
    orderRepository.save(order);

    // Publish event after transaction commits
    eventPublisher.publishEvent(new OrderCreatedEvent(order.getId(), order.getItems()));

    return order;
}

@Component
public class OrderEventHandler {
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Execute asynchronously after the order is saved
        inventoryService.reserve(event.getItems());
        paymentService.charge(event.getPayment());
    }
}

See examples.md for complete working examples.

Instructions

1. Design Domain Events

Create immutable event classes extending a base DomainEvent class:

public abstract class DomainEvent {
    private final UUID eventId;
    private final LocalDateTime occurredAt;
    private final UUID correlationId;
}

public class ProductCreatedEvent extends DomainEvent {
    private final ProductId productId;
    private final String name;
    private final BigDecimal price;
}

See domain-events-design.md for patterns.

2. Publish Events from Aggregates

Add domain events to aggregate roots, publish via ApplicationEventPublisher:

@Service
@Transactional
public class ProductService {
    public Product createProduct(CreateProductRequest request) {
        Product product = Product.create(request.getName(), request.getPrice(), request.getStock());
        repository.save(product);

        product.getDomainEvents().forEach(eventPublisher::publishEvent);
        product.clearDomainEvents();

        return product;
    }
}

See aggregate-root-patterns.md for DDD patterns.

3. Handle Events Transactionally

Use @TransactionalEventListener for reliable event handling:

@Component
public class ProductEventHandler {
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onProductCreated(ProductCreatedEvent event) {
        notificationService.sendProductCreatedNotification(event.getName());
    }
}

Validate: Confirm the event handler fires only after the transaction commits by checking that the database state is committed before the handler executes.

See event-handling.md for handling patterns.

4. Configure Kafka Infrastructure

Configure KafkaTemplate for publishing, @KafkaListener for consuming:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

Validate: Send a test event via KafkaTemplate and confirm it appears in the consumer logs before proceeding to production patterns.

See dependency-setup.md and configuration.md.

5. Implement Outbox Pattern

Create OutboxEvent entity for atomic event storage:

@Entity
public class OutboxEvent {
    private UUID id;
    private String aggregateId;
    private String eventType;
    private String payload;
    private LocalDateTime publishedAt;
}

Validate: Confirm the scheduled processor picks up pending events by checking the publishedAt timestamp is set after the scheduled run.

Scheduled processor publishes pending events. See outbox-pattern.md.

6. Handle Failure Scenarios

Implement retry logic, dead-letter queues, idempotent handlers:

@RetryableTopic(attempts = "3")
@KafkaListener(topics = "product-events")
public void handleProductEvent(ProductCreatedEventDto event) {
    orderService.onProductCreated(event);
}

Validate: Confirm messages reach the dead-letter topic after exhausting retries before moving to observability.

7. Add Observability

Enable Spring Cloud Sleuth for distributed tracing, monitor metrics.

Best Practices

• Use past tense naming: ProductCreated (not CreateProduct)
• Keep events immutable: All fields should be final
• Include correlation IDs: For tracing events across services
• Use AFTER_COMMIT phase: Ensures events are published after successful database transaction
• Implement idempotent handlers: Handle duplicate events gracefully
• Add retry mechanisms: For failed event processing with exponential backoff
• Implement dead-letter queues: For events that fail processing after retries
• Log all failures: Include sufficient context for debugging
• Make handlers order-independent: Event ordering is not guaranteed in distributed systems
• Batch event processing: When handling high volumes
• Monitor event latencies: Set up alerts for slow processing

References

• [dependency-setup.md](references/dependency-setup.md) — Maven/Gradle dependencies
• [configuration.md](references/configuration.md) — Kafka and Spring Cloud Stream configuration
• [domain-events-design.md](references/domain-events-design.md) — Domain event design patterns
• [aggregate-root-patterns.md](references/aggregate-root-patterns.md) — Aggregate root with event publishing
• [event-publishing.md](references/event-publishing.md) — Local and distributed event publishing
• [event-handling.md](references/event-handling.md) — Event handling and consumption patterns
• [outbox-pattern.md](references/outbox-pattern.md) — Transactional outbox pattern for reliability
• [testing-strategies.md](references/testing-strategies.md) — Unit and integration testing approaches
• [examples.md](references/examples.md) — Complete working examples
• [event-driven-patterns-reference.md](references/event-driven-patterns-reference.md) — Detailed reference documentation

Constraints and Warnings

• Events published with @TransactionalEventListener only fire after transaction commit
• Avoid publishing large objects in events (memory pressure, serialization issues)
• Be cautious with async event handlers (separate threads, concurrency issues)
• Kafka consumers must handle duplicate messages (implement idempotent processing)
• Event ordering is not guaranteed in distributed systems (design handlers to be order-independent)
• Never perform blocking operations in event listeners on the main transaction thread
• Monitor for event processing backlogs (indicate system capacity issues)

Related Skills

• spring-boot-security-jwt — JWT authentication for secure event publishing
• spring-boot-test-patterns — Testing event-driven applications
• aws-sdk-java-v2-lambda — Event-driven processing with AWS Lambda
• langchain4j-tool-function-calling-patterns — AI-driven event processing

Aggregate Root with Event Publishing

Aggregate Root Design

Base Aggregate Root

import jakarta.persistence.*;
import lombok.*;
import java.util.ArrayList;
import java.util.List;

@MappedSuperclass
@Getter
public abstract class AggregateRoot<ID> {
    @Transient
    protected List<DomainEvent> domainEvents = new ArrayList<>();

    public List<DomainEvent> getDomainEvents() {
        return new ArrayList<>(domainEvents);
    }

    public void clearDomainEvents() {
        domainEvents.clear();
    }

    protected void addDomainEvent(DomainEvent event) {
        domainEvents.add(event);
    }
}

Product Aggregate

import jakarta.persistence.*;
import lombok.*;
import java.math.BigDecimal;

@Entity
@Table(name = "products")
@Getter
@Setter(AccessLevel.PROTECTED)
@EqualsAndHashCode(of = "id")
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class Product extends AggregateRoot<ProductId> {

    @Id
    @Embedded
    private ProductId id;

    @Column(nullable = false)
    private String name;

    @Column(nullable = false, precision = 10, scale = 2)
    private BigDecimal price;

    @Column(nullable = false)
    private Integer stock;

    // Factory method
    public static Product create(String name, BigDecimal price, Integer stock) {
        Product product = new Product();
        product.id = ProductId.generate();
        product.name = name;
        product.price = price;
        product.stock = stock;
        product.addDomainEvent(new ProductCreatedEvent(product.id, name, price, stock));
        return product;
    }

    // Domain behavior
    public void decreaseStock(Integer quantity) {
        if (this.stock < quantity) {
            throw new InsufficientStockException(
                String.format("Insufficient stock: requested=%d, available=%d",
                    quantity, this.stock)
            );
        }

        this.stock -= quantity;
        addDomainEvent(new ProductStockDecreasedEvent(this.id, quantity, this.stock));
    }

    public void increaseStock(Integer quantity) {
        this.stock += quantity;
        addDomainEvent(new ProductStockIncreasedEvent(this.id, quantity, this.stock));
    }

    public void updatePrice(BigDecimal newPrice) {
        if (newPrice.compareTo(BigDecimal.ZERO) <= 0) {
            throw new InvalidPriceException("Price must be positive");
        }

        BigDecimal oldPrice = this.price;
        this.price = newPrice;
        addDomainEvent(new ProductPriceUpdatedEvent(this.id, oldPrice, newPrice));
    }

    public void discontinue() {
        addDomainEvent(new ProductDiscontinuedEvent(this.id, this.name, this.stock));
    }

    @Embeddable
    @EqualsAndHashCode
    @NoArgsConstructor(access = AccessLevel.PROTECTED)
    @AllArgsConstructor(access = AccessLevel.PRIVATE)
    public static class ProductId {
        private String value;

        public static ProductId of(String value) {
            return new ProductId(value);
        }

        public static ProductId generate() {
            return new ProductId(UUID.randomUUID().toString());
        }

        @Override
        public String toString() {
            return value;
        }
    }
}

Order Aggregate

@Entity
@Table(name = "orders")
@Getter
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class Order extends AggregateRoot<OrderId> {

    @Id
    @Embedded
    private OrderId id;

    @Embedded
    private CustomerId customerId;

    @ElementCollection
    @CollectionTable(name = "order_items", joinColumns = @JoinColumn(name = "order_id"))
    private List<OrderItem> items = new ArrayList<>();

    @Enumerated(Enum.STRING)
    private OrderStatus status;

    @Column(nullable = false, precision = 10, scale = 2)
    private BigDecimal totalAmount;

    public static Order create(CustomerId customerId, List<OrderItem> items) {
        Order order = new Order();
        order.id = OrderId.generate();
        order.customerId = customerId;
        order.items = List.copyOf(items);
        order.status = OrderStatus.PENDING;
        order.totalAmount = calculateTotal(items);
        order.addDomainEvent(new OrderCreatedEvent(
            order.id,
            order.customerId,
            order.items,
            order.totalAmount
        ));
        return order;
    }

    public void pay(PaymentMethod paymentMethod) {
        if (this.status != OrderStatus.PENDING) {
            throw new InvalidOrderStatusException(
                "Cannot pay order in status: " + this.status
            );
        }

        this.status = OrderStatus.PAID;
        addDomainEvent(new OrderPaidEvent(
            this.id,
            this.customerId,
            this.totalAmount,
            paymentMethod
        ));
    }

    public void ship(ShippingAddress shippingAddress) {
        if (this.status != OrderStatus.PAID) {
            throw new InvalidOrderStatusException(
                "Cannot ship order in status: " + this.status
            );
        }

        this.status = OrderStatus.SHIPPED;
        addDomainEvent(new OrderShippedEvent(
            this.id,
            this.customerId,
            shippingAddress
        ));
    }

    public void cancel(String reason) {
        if (this.status == OrderStatus.SHIPPED || this.status == OrderStatus.DELIVERED) {
            throw new InvalidOrderStatusException(
                "Cannot cancel order in status: " + this.status
            );
        }

        this.status = OrderStatus.CANCELLED;
        addDomainEvent(new OrderCancelledEvent(
            this.id,
            this.customerId,
            reason
        ));
    }

    private static BigDecimal calculateTotal(List<OrderItem> items) {
        return items.stream()
            .map(item -> item.getUnitPrice().multiply(BigDecimal.valueOf(item.getQuantity())))
            .reduce(BigDecimal.ZERO, BigDecimal::add);
    }

    @Embeddable
    @EqualsAndHashCode
    @NoArgsConstructor(access = AccessLevel.PROTECTED)
    @AllArgsConstructor(access = AccessLevel.PRIVATE)
    public static class OrderId {
        private String value;

        public static OrderId of(String value) {
            return new OrderId(value);
        }

        public static OrderId generate() {
            return new OrderId(UUID.randomUUID().toString());
        }
    }

    @Embeddable
    @AllArgsConstructor
    @NoArgsConstructor
    public static class OrderItem {
        private ProductId productId;
        private String productName;
        private Integer quantity;
        private BigDecimal unitPrice;
    }

    public enum OrderStatus {
        PENDING, PAID, SHIPPED, DELIVERED, CANCELLED
    }
}

Repository Pattern

import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;

import java.util.Optional;

@Repository
public interface ProductRepository extends JpaRepository<Product, Product.ProductId> {
    Optional<Product> findByProductName(String name);
}

Application Service Pattern

import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import lombok.RequiredArgsConstructor;

@Service
@RequiredArgsConstructor
public class ProductApplicationService {
    private final ProductRepository productRepository;
    private final ApplicationEventPublisher eventPublisher;

    @Transactional
    public ProductResponse createProduct(CreateProductRequest request) {
        Product product = Product.create(
            request.getName(),
            request.getPrice(),
            request.getStock()
        );

        productRepository.save(product);

        // Publish domain events
        product.getDomainEvents().forEach(eventPublisher::publishEvent);
        product.clearDomainEvents();

        return mapToResponse(product);
    }

    @Transactional
    public void decreaseStock(DecreaseStockRequest request) {
        Product product = productRepository.findById(request.getProductId())
            .orElseThrow(() -> new ProductNotFoundException(request.getProductId()));

        product.decreaseStock(request.getQuantity());

        productRepository.save(product);

        // Publish domain events
        product.getDomainEvents().forEach(eventPublisher::publishEvent);
        product.clearDomainEvents();
    }

    private ProductResponse mapToResponse(Product product) {
        return new ProductResponse(
            product.getId().getValue(),
            product.getName(),
            product.getPrice(),
            product.getStock()
        );
    }
}

Event-Driven Architecture Configuration

Basic Configuration

application.properties

# Server Configuration
server.port=8080

# Kafka Configuration
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

# Spring Cloud Stream Configuration
spring.cloud.stream.kafka.binder.brokers=localhost:9092

application.yml

server:
  port: 8080

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"

  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
      bindings:
        productCreated-in-0:
          destination: product-events
          group: order-service

Advanced Configuration

Kafka Producer Configuration

spring:
  kafka:
    producer:
      acks: all
      retries: 3
      properties:
        retry.backoff.ms: 1000
      enable.idempotence: true
      compression-type: snappy

Kafka Consumer Configuration

spring:
  kafka:
    consumer:
      auto-offset-reset: earliest
      enable-auto-commit: false
      max-poll-records: 100
    listener:
      ack-mode: manual_immediate

Error Handling Configuration

spring:
  kafka:
    producer:
      properties:
        retries: 3
        retry.backoff.ms: 1000
    consumer:
      properties:
        max.poll.interval.ms: 300000

Spring Cloud Stream Bindings

Producer Binding

spring:
  cloud:
    stream:
      bindings:
        productCreated-out-0:
          destination: product-events
          producer:
            partition-count: 3

Consumer Binding

spring:
  cloud:
    stream:
      bindings:
        productCreated-in-0:
          destination: product-events
          group: order-service
          consumer:
            max-attempts: 3
            back-off-initial-interval: 1000

Environment-Specific Configuration

Development

spring:
  kafka:
    bootstrap-servers: localhost:9092

Production

spring:
  kafka:
    bootstrap-servers: kafka1.prod.example.com:9092,kafka2.prod.example.com:9092,kafka3.prod.example.com:9092
    properties:
      security.protocol: SSL
      ssl.truststore.location: /etc/kafka/truststore.jks
      ssl.keystore.location: /etc/kafka/keystore.jks

Event-Driven Architecture Dependencies

Maven Dependencies

<dependencies>
    <!-- Spring Boot Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <!-- Spring Data JPA -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>

    <!-- Kafka for distributed messaging -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>

    <!-- Spring Cloud Stream -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream</artifactId>
        <version>4.0.4</version>
    </dependency>

    <!-- Testing -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>

    <!-- Testcontainers for integration testing -->
    <dependency>
        <groupId>org.testcontainers</groupId>
        <artifactId>testcontainers</artifactId>
        <version>1.19.0</version>
        <scope>test</scope>
    </dependency>
</dependencies>

Gradle Dependencies

dependencies {
    // Spring Boot Web
    implementation 'org.springframework.boot:spring-boot-starter-web'

    // Spring Data JPA
    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'

    // Kafka
    implementation 'org.springframework.kafka:spring-kafka'

    // Spring Cloud Stream
    implementation 'org.springframework.cloud:spring-cloud-stream:4.0.4'

    // Testing
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.testcontainers:testcontainers:1.19.0'
}

Version Selection

• Spring Boot 3.x: Use Spring Kafka 3.x, Spring Cloud Stream 4.x
• Spring Boot 2.x: Use Spring Kafka 2.x, Spring Cloud Stream 3.x
• Always check for compatible versions at Spring Cloud Release Train

Domain Events Design

Domain Event Base Class

Create an immutable base class for all domain events:

import java.time.LocalDateTime;
import java.util.UUID;

public abstract class DomainEvent {
    private final UUID eventId;
    private final LocalDateTime occurredAt;
    private final UUID correlationId;

    protected DomainEvent() {
        this.eventId = UUID.randomUUID();
        this.occurredAt = LocalDateTime.now();
        this.correlationId = UUID.randomUUID();
    }

    protected DomainEvent(UUID correlationId) {
        this.eventId = UUID.randomUUID();
        this.occurredAt = LocalDateTime.now();
        this.correlationId = correlationId;
    }

    public UUID getEventId() {
        return eventId;
    }

    public LocalDateTime getOccurredAt() {
        return occurredAt;
    }

    public UUID getCorrelationId() {
        return correlationId;
    }
}

Specific Domain Events

Product Created Event

import java.math.BigDecimal;
import java.util.UUID;

public class ProductCreatedEvent extends DomainEvent {
    private final ProductId productId;
    private final String name;
    private final BigDecimal price;
    private final Integer stock;

    public ProductCreatedEvent(ProductId productId, String name, BigDecimal price, Integer stock) {
        super();
        this.productId = productId;
        this.name = name;
        this.price = price;
        this.stock = stock;
    }

    public ProductId getProductId() {
        return productId;
    }

    public String getName() {
        return name;
    }

    public BigDecimal getPrice() {
        return price;
    }

    public Integer getStock() {
        return stock;
    }
}

Product Stock Decreased Event

public class ProductStockDecreasedEvent extends DomainEvent {
    private final ProductId productId;
    private final Integer quantity;
    private final Integer remainingStock;

    public ProductStockDecreasedEvent(ProductId productId, Integer quantity, Integer remainingStock) {
        super();
        this.productId = productId;
        this.quantity = quantity;
        this.remainingStock = remainingStock;
    }

    public ProductId getProductId() {
        return productId;
    }

    public Integer getQuantity() {
        return quantity;
    }

    public Integer getRemainingStock() {
        return remainingStock;
    }
}

Order Created Event

import java.util.List;

public class OrderCreatedEvent extends DomainEvent {
    private final OrderId orderId;
    private final CustomerId customerId;
    private final List<OrderItem> items;
    private final BigDecimal total;

    public OrderCreatedEvent(OrderId orderId, CustomerId customerId, List<OrderItem> items, BigDecimal total) {
        super();
        this.orderId = orderId;
        this.customerId = customerId;
        this.items = List.copyOf(items);
        this.total = total;
    }

    public OrderId getOrderId() {
        return orderId;
    }

    public CustomerId getCustomerId() {
        return customerId;
    }

    public List<OrderItem> getItems() {
        return items;
    }

    public BigDecimal getTotal() {
        return total;
    }
}

Event Design Guidelines

Naming Conventions

• Use past tense: ProductCreated (not CreateProduct)
• Reflect business domain: OrderPaid, InventoryReserved
• Be explicit: ProductStockDecreased (not ProductStockChanged)

Event Content

• Include all relevant data: Events should be self-contained
• Use value objects: ProductId, OrderId instead of primitive Long
• Make events immutable: All fields should be final

Event Metadata

• eventId: Unique identifier for the event
• occurredAt: Timestamp when the event occurred
• correlationId: Links related events across aggregates

Example: Rich Event Design

public class OrderPlacedEvent extends DomainEvent {
    private final OrderId orderId;
    private final CustomerId customerId;
    private final List<OrderItem> items;
    private final BigDecimal totalAmount;
    private final String shippingAddress;
    private final PaymentMethod paymentMethod;
    private final Instant estimatedDeliveryDate;

    public OrderPlacedEvent(
        OrderId orderId,
        CustomerId customerId,
        List<OrderItem> items,
        BigDecimal totalAmount,
        String shippingAddress,
        PaymentMethod paymentMethod,
        Instant estimatedDeliveryDate,
        UUID correlationId
    ) {
        super(correlationId);
        this.orderId = orderId;
        this.customerId = customerId;
        this.items = List.copyOf(items);
        this.totalAmount = totalAmount;
        this.shippingAddress = shippingAddress;
        this.paymentMethod = paymentMethod;
        this.estimatedDeliveryDate = estimatedDeliveryDate;
    }

    // Getters...

    public record OrderItem(
        ProductId productId,
        String productName,
        Integer quantity,
        BigDecimal unitPrice
    ) {}
}

Event Serialization

JSON Serialization

import com.fasterxml.jackson.annotation.JsonFormat;

public class ProductCreatedEvent extends DomainEvent {
    private final String productId;
    private final String name;

    @JsonFormat(shape = JsonFormat.Shape.STRING)
    private final BigDecimal price;

    private final Integer stock;

    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss")
    private final LocalDateTime occurredAt;

    // Constructor and getters...
}

Event DTO Pattern

// Domain event (internal)
public class ProductCreatedEvent extends DomainEvent {
    private final ProductId productId;
    private final String name;
    private final BigDecimal price;
}

// Event DTO (external communication)
public class ProductCreatedEventDto {
    private final String eventId;
    private final String productId;
    private final String name;
    private final BigDecimal price;
    private final LocalDateTime occurredAt;
    private final String correlationId;

    public static ProductCreatedEventDto from(ProductCreatedEvent event) {
        return new ProductCreatedEventDto(
            event.getEventId().toString(),
            event.getProductId().getValue(),
            event.getName(),
            event.getPrice(),
            event.getOccurredAt(),
            event.getCorrelationId().toString()
        );
    }
}

Event Versioning

Versioned Events

public class ProductCreatedEventV2 extends DomainEvent {
    private final ProductId productId;
    private final String name;
    private final BigDecimal price;
    private final Integer stock;
    private final String category; // New field in V2

    // Include version information
    private final String eventVersion = "2.0";

    // Constructor and getters...
}

Upcaster Pattern

@Component
public class EventUpcaster {
    public ProductCreatedEventV2 upcast(ProductCreatedEventV1 v1Event) {
        return new ProductCreatedEventV2(
            v1Event.getProductId(),
            v1Event.getName(),
            v1Event.getPrice(),
            v1Event.getStock(),
            "uncategorized" // Default value for new field
        );
    }
}

Spring Boot Event-Driven Patterns - References

Complete API reference for event-driven architecture in Spring Boot applications.

Domain Event Annotations and Interfaces

ApplicationEvent

Base class for Spring events (deprecated in newer versions in favor of plain objects).

public abstract class ApplicationEvent extends EventObject {
    private final long timestamp;
    
    public ApplicationEvent(Object source) {
        super(source);
        this.timestamp = System.currentTimeMillis();
    }
    
    public long getTimestamp() {
        return timestamp;
    }
}

// Modern approach: Use plain POJOs
public record ProductCreatedEvent(String productId, String name, BigDecimal price) {}

Custom Domain Event Base Class

public abstract class DomainEvent {
    private final UUID eventId;
    private final LocalDateTime occurredAt;
    private final UUID correlationId;

    protected DomainEvent() {
        this.eventId = UUID.randomUUID();
        this.occurredAt = LocalDateTime.now();
        this.correlationId = UUID.randomUUID();
    }
}

Event Publishing Annotations

@EventListener

Register event listener methods.

@EventListener
public void onProductCreated(ProductCreatedEvent event) { }

@EventListener(condition = "#event.productId == '123'")  // SpEL condition
public void onSpecificProduct(ProductCreatedEvent event) { }

@EventListener(classes = { ProductCreatedEvent.class, ProductUpdatedEvent.class })
public void onProductEvent(DomainEvent event) { }

@TransactionalEventListener

Listen to events within transaction lifecycle.

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void onProductCreated(ProductCreatedEvent event) { }

@TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
public void beforeCommit(ProductCreatedEvent event) { }

@TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
public void afterRollback(ProductCreatedEvent event) { }

TransactionPhase Values:

• BEFORE_COMMIT - Before transaction commits
• AFTER_COMMIT - After successful commit (recommended)
• AFTER_ROLLBACK - After transaction rollback
• AFTER_COMPLETION - After transaction completion (success or rollback)

Event Publishing Reference

ApplicationEventPublisher Interface

public interface ApplicationEventPublisher {
    void publishEvent(ApplicationEvent event);
    void publishEvent(Object event);  // Modern approach
}

Usage Pattern

@Service
@RequiredArgsConstructor
public class ProductService {
    private final ApplicationEventPublisher eventPublisher;

    public Product create(CreateProductRequest request) {
        Product product = Product.create(request);
        Product saved = repository.save(product);
        
        // Publish events
        saved.getDomainEvents().forEach(eventPublisher::publishEvent);
        saved.clearDomainEvents();
        
        return saved;
    }
}

Kafka Spring Cloud Stream Reference

Stream Binders Configuration

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          default-binder: kafka
          configuration:
            linger.ms: 10
            batch.size: 1024
      
      bindings:
        # Consumer binding
        productCreatedConsumer-in-0:
          destination: product-events
          group: product-service
          consumer:
            max-attempts: 3
            back-off-initial-interval: 1000
            back-off-max-interval: 10000
        
        # Producer binding
        eventPublisher-out-0:
          destination: product-events
          producer:
            partition-key-expression: headers['partitionKey']

Consumer Function Binding

@Configuration
public class EventConsumers {
    
    @Bean
    public java.util.function.Consumer<ProductCreatedEvent> productCreatedConsumer(
            InventoryService inventoryService) {
        return event -> {
            log.info("Consumed: {}", event);
            inventoryService.process(event);
        };
    }
    
    // Multiple consumers
    @Bean
    public java.util.function.Consumer<ProductUpdatedEvent> productUpdatedConsumer() {
        return event -> { };
    }
}

// application.yml
spring.cloud.stream.bindings.productCreatedConsumer-in-0.destination=product-events
spring.cloud.stream.bindings.productUpdatedConsumer-in-0.destination=product-events

Producer Function Binding

@Configuration
public class EventProducers {
    
    @Bean
    public java.util.function.Supplier<ProductCreatedEvent> eventPublisher() {
        return () -> new ProductCreatedEvent("prod-123", "Laptop", BigDecimal.TEN);
    }
}

// application.yml
spring.cloud.stream.bindings.eventPublisher-out-0.destination=product-events

Transactional Outbox Pattern Reference

Outbox Entity Schema

CREATE TABLE outbox_events (
    id UUID PRIMARY KEY,
    aggregate_id VARCHAR(255) NOT NULL,
    aggregate_type VARCHAR(255),
    event_type VARCHAR(255) NOT NULL,
    payload TEXT NOT NULL,
    correlation_id UUID,
    created_at TIMESTAMP NOT NULL,
    published_at TIMESTAMP,
    retry_count INTEGER DEFAULT 0,
    KEY idx_published (published_at),
    KEY idx_created (created_at)
);

Implementation Pattern

// In single transaction:
// 1. Update aggregate
product = repository.save(product);

// 2. Store events in outbox
product.getDomainEvents().forEach(event -> {
    outboxRepository.save(new OutboxEvent(
        aggregateId, eventType, payload, correlationId
    ));
});

// Then separately, scheduled task publishes from outbox
@Scheduled(fixedDelay = 5000)
public void publishPendingEvents() {
    List<OutboxEvent> pending = outboxRepository.findByPublishedAtIsNull();
    pending.forEach(event -> {
        kafkaTemplate.send(topic, event.getPayload());
        event.setPublishedAt(now());
    });
}

Maven Dependencies

<!-- Local Events (Spring Framework core) -->
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context</artifactId>
</dependency>

<!-- Kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<!-- Spring Cloud Stream -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
    <version>4.0.4</version>
</dependency>

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
    <version>4.0.4</version>
</dependency>

<!-- Jackson for JSON -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

<dependency>
    <groupId>com.fasterxml.jackson.datatype</groupId>
    <artifactId>jackson-datatype-jsr310</artifactId>
</dependency>

Gradle Dependencies

dependencies {
    // Local Events
    implementation 'org.springframework:spring-context'

    // Kafka
    implementation 'org.springframework.kafka:spring-kafka'

    // Spring Cloud Stream
    implementation 'org.springframework.cloud:spring-cloud-stream:4.0.4'
    implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka:4.0.4'

    // Jackson
    implementation 'com.fasterxml.jackson.core:jackson-databind'
    implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
}

Event Ordering Guarantees

Kafka Partition Key Strategy

// Events with same product must be in same partition
kafkaTemplate.send(topic, 
    productId,  // Key: ensures ordering per product
    event);     // Value

// Consumer configuration
spring.kafka.consumer.properties.isolation.level=read_committed
spring.cloud.stream.kafka.binder.configuration.isolation.level=read_committed

Error Handling Patterns

Retry with Backoff

spring:
  cloud:
    stream:
      bindings:
        eventConsumer-in-0:
          consumer:
            max-attempts: 3
            back-off-initial-interval: 1000      # 1 second
            back-off-max-interval: 10000         # 10 seconds
            back-off-multiplier: 2.0             # Exponential
            default-retryable: true
            retryable-exceptions:
              com.example.RetryableException: true

Dead Letter Topic (DLT)

spring:
  cloud:
    stream:
      kafka:
        bindings:
          eventConsumer-in-0:
            consumer:
              enable-dlq: true
              dlq-name: product-events.dlq
              dlq-producer-properties:
                linger.ms: 5

Idempotency Patterns

Idempotent Consumer

@Component
public class IdempotentEventHandler {
    private final IdempotencyKeyRepository idempotencyRepository;
    private final EventProcessingService eventService;

    @EventListener
    public void handle(DomainEvent event) throws Exception {
        String idempotencyKey = event.getEventId().toString();
        
        // Check if already processed
        if (idempotencyRepository.exists(idempotencyKey)) {
            log.info("Event already processed: {}", idempotencyKey);
            return;
        }
        
        try {
            // Process event
            eventService.process(event);
            
            // Mark as processed
            idempotencyRepository.save(new IdempotencyKey(idempotencyKey));
        } catch (Exception e) {
            log.error("Event processing failed: {}", idempotencyKey, e);
            throw e;
        }
    }
}

Testing Event-Driven Systems

Local Event Testing

@SpringBootTest
class EventDrivenTest {
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @MockBean
    private EventHandler handler;

    @Test
    void shouldHandleEvent() {
        // Arrange
        ProductCreatedEvent event = new ProductCreatedEvent("123", "Laptop", BigDecimal.TEN);

        // Act
        eventPublisher.publishEvent(event);

        // Assert
        verify(handler).onProductCreated(event);
    }
}

Kafka Testing with Testcontainers

@SpringBootTest
@Testcontainers
class KafkaEventTest {
    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.5.0"));

    @DynamicPropertySource
    static void setupProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Test
    void shouldPublishEventToKafka() throws Exception {
        ProductCreatedEvent event = new ProductCreatedEvent("123", "Laptop", BigDecimal.TEN);
        kafkaTemplate.send("product-events", "123", event).get(5, TimeUnit.SECONDS);
        
        // Verify consumption
    }
}

Monitoring and Observability

Spring Boot Actuator Metrics

# Enable metrics
management.endpoints.web.exposure.include=metrics,health

# Kafka metrics
kafka.controller.metrics.topic.under_replication_count
kafka.log.leader_election.latency.avg

Custom Event Metrics

@Component
@RequiredArgsConstructor
public class EventMetrics {
    private final MeterRegistry meterRegistry;

    public void recordEventPublished(String eventType) {
        meterRegistry.counter("events.published", "type", eventType).increment();
    }

    public void recordEventProcessed(String eventType, long durationMs) {
        meterRegistry.timer("events.processed", "type", eventType).record(durationMs, TimeUnit.MILLISECONDS);
    }

    public void recordEventFailed(String eventType) {
        meterRegistry.counter("events.failed", "type", eventType).increment();
    }
}

Related Skills

• spring-boot-crud-patterns - Domain events in CRUD operations
• spring-boot-rest-api-standards - Event notifications via webhooks
• spring-boot-test-patterns - Testing event-driven systems
• spring-boot-dependency-injection - Dependency injection in event handlers

External Resources

Official Documentation

• Spring ApplicationContext
• Spring Cloud Stream
• Kafka Spring Integration

Patterns and Best Practices

• Event Sourcing Pattern
• Saga Pattern
• Transactional Outbox

Event Handling Patterns

Local Event Handling

Transactional Event Listener

import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
import lombok.RequiredArgsConstructor;

@Component
@RequiredArgsConstructor
public class ProductEventHandler {
    private final NotificationService notificationService;
    private final AuditService auditService;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onProductCreated(ProductCreatedEvent event) {
        auditService.logProductCreation(
            event.getProductId().getValue(),
            event.getName(),
            event.getPrice(),
            event.getCorrelationId()
        );

        notificationService.sendProductCreatedNotification(event.getName());
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onProductStockDecreased(ProductStockDecreasedEvent event) {
        notificationService.sendStockUpdateNotification(
            event.getProductId().getValue(),
            event.getQuantity()
        );
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
    public void onTransactionRollback(DomainEvent event) {
        log.error("Transaction rolled back for event: {}", event.getEventId());
    }
}

Async Event Listener

@Component
@RequiredArgsConstructor
public class AsyncEventHandler {
    private final EmailService emailService;

    @Async
    @EventListener
    public void handleOrderCreatedEvent(OrderCreatedEvent event) {
        // Executes asynchronously in a separate thread
        emailService.sendOrderConfirmationEmail(
            event.getCustomerId().getValue(),
            event.getOrderId().getValue()
        );
    }
}

Kafka Event Consumption

Kafka Listener

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Component
@RequiredArgsConstructor
@Slf4j
public class ProductEventConsumer {

    private final OrderService orderService;

    @KafkaListener(
        topics = "product-events",
        groupId = "order-service",
        properties = {
            "spring.json.value.default.type=com.example.events.ProductCreatedEventDto"
        }
    )
    public void handleProductCreated(ProductCreatedEventDto event) {
        log.info("Received ProductCreatedEvent: {}", event.getProductId());

        try {
            orderService.onProductCreated(event);
        } catch (Exception e) {
            log.error("Failed to handle ProductCreatedEvent", e);
            throw e; // Re-throw to trigger retry
        }
    }

    @KafkaListener(
        topics = "product-events",
        groupId = "order-service",
        properties = {
            "spring.json.value.default.type=com.example.events.ProductStockDecreasedEventDto"
        }
    )
    public void handleProductStockDecreased(ProductStockDecreasedEventDto event) {
        log.info("Received ProductStockDecreasedEvent: {}", event.getProductId());

        orderService.onProductStockDecreased(event);
    }
}

Manual Acknowledgment

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;

@Component
@Slf4j
public class ManualAckConsumer {

    @KafkaListener(
        topics = "product-events",
        groupId = "order-service",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleWithManualAck(
        @Payload ProductCreatedEventDto event,
        @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment
    ) {
        try {
            // Process event
            orderService.onProductCreated(event);

            // Manually acknowledge
            acknowledgment.acknowledge();

        } catch (Exception e) {
            log.error("Failed to process event", e);
            // Don't acknowledge - message will be redelivered
        }
    }
}

Error Handling with Dead Letter Queue

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.retry.annotation.Backoff;

@Component
@Slf4j
public class ResilientEventConsumer {

    @RetryableTopic(
        attempts = "3",
        backoff = @Backoff(delay = 1000, multiplier = 2),
        autoCreateTopics = "false",
        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE
    )
    @KafkaListener(
        topics = "product-events",
        groupId = "order-service"
    )
    public void handleProductEvent(ProductCreatedEventDto event) {
        log.info("Processing product event: {}", event.getProductId());

        // Process event
        orderService.onProductCreated(event);
    }

    @KafkaListener(
        topics = "product-events-dlt",
        groupId = "order-service-dlt"
    )
    public void handleDeadLetterEvent(ProductCreatedEventDto event) {
        log.error("Event moved to DLT: {}", event.getProductId());

        // Log to monitoring system
        monitoringService.alertDeadLetterEvent(event);

        // Store for manual inspection
        deadLetterRepository.save(event);
    }
}

Spring Cloud Stream Consumption

Functional Consumer

import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import java.util.function.Consumer;

@Component
@RequiredArgsConstructor
public class ProductEventStreamConsumer {
    private final OrderService orderService;

    @Bean
    public Consumer<ProductCreatedEventDto> productCreated() {
        return event -> {
            log.info("Received ProductCreatedEvent: {}", event.getProductId());
            orderService.onProductCreated(event);
        };
    }

    @Bean
    public Consumer<ProductStockDecreasedEventDto> productStockDecreased() {
        return event -> {
            log.info("Received ProductStockDecreasedEvent: {}", event.getProductId());
            orderService.onProductStockDecreased(event);
        };
    }
}

Consumer with Error Handling

import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
public class ErrorHandlingConsumer {

    @Bean
    public Consumer<Message<ProductCreatedEventDto>> productCreatedWithRetry() {
        return message -> {
            try {
                ProductCreatedEventDto event = message.getPayload();
                orderService.onProductCreated(event);
            } catch (Exception e) {
                log.error("Failed to process event", e);

                // Send to dead letter topic
                throw new RuntimeException("Failed to process event", e);
            }
        };
    }
}

Event Handler Best Practices

1. Idempotent Handlers

@Component
@RequiredArgsConstructor
public class IdempotentEventHandler {
    private final ProcessedEventRepository processedEventRepository;

    public void handleProductCreated(ProductCreatedEventDto event) {
        // Check if event was already processed
        if (processedEventRepository.existsByEventId(event.getEventId())) {
            log.info("Event already processed: {}", event.getEventId());
            return;
        }

        // Process event
        orderService.onProductCreated(event);

        // Mark as processed
        processedEventRepository.save(new ProcessedEvent(event.getEventId()));
    }
}

2. Event Handler with Validation

@Component
public class ValidatingEventHandler {

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Validate event
        if (event.getItems().isEmpty()) {
            throw new InvalidEventException("Order items cannot be empty");
        }

        if (event.getTotalAmount().compareTo(BigDecimal.ZERO) <= 0) {
            throw new InvalidEventException("Total amount must be positive");
        }

        // Process valid event
        inventoryService.reserveItems(event.getItems());
        paymentService.charge(event.getTotalAmount());
    }
}

3. Event Handler with Circuit Breaker

import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;

@Component
@RequiredArgsConstructor
public class ResilientEventHandler {
    private final ExternalServiceClient externalServiceClient;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    @CircuitBreaker(
        name = "externalService",
        fallbackMethod = "handleExternalServiceFailure"
    )
    public void handleOrderCreated(OrderCreatedEvent event) {
        externalServiceClient.notifyOrderCreated(event);
    }

    private void handleExternalServiceFailure(OrderCreatedEvent event, Exception ex) {
        log.error("External service unavailable for event: {}", event.getOrderId(), ex);

        // Store event for later retry
        outboxRepository.save(OutboxEvent.from(event));
    }
}

4. Event Handler with Timeout

import org.springframework.transaction.annotation.Transactional;
import java.util.concurrent.TimeUnit;

@Component
public class TimeoutEventHandler {

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleOrderCreated(OrderCreatedEvent event) {
        ExecutorService executor = Executors.newSingleThreadExecutor();

        try {
            Future<?> future = executor.submit(() -> {
                notificationService.sendOrderConfirmation(event);
            });

            future.get(5, TimeUnit.SECONDS); // Timeout after 5 seconds

        } catch (TimeoutException e) {
            log.error("Notification timed out for order: {}", event.getOrderId());
            // Handle timeout appropriately
        } catch (Exception e) {
            log.error("Failed to send notification", e);
            throw new EventHandlingException("Failed to handle event", e);
        } finally {
            executor.shutdown();
        }
    }
}

5. Batch Event Processing

import org.springframework.scheduling.annotation.Scheduled;
import java.util.List;

@Component
public class BatchEventHandler {
    private final List<DomainEvent> eventBuffer = new ArrayList<>();

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void bufferEvent(DomainEvent event) {
        synchronized (eventBuffer) {
            eventBuffer.add(event);

            if (eventBuffer.size() >= 100) {
                processBatch();
            }
        }
    }

    @Scheduled(fixedDelay = 5000)
    public synchronized void processBatch() {
        if (eventBuffer.isEmpty()) {
            return;
        }

        List<DomainEvent> batch = new ArrayList<>(eventBuffer);
        eventBuffer.clear();

        // Process batch
        batchProcessor.process(batch);
    }
}

Event Publishing Patterns

Local Event Publishing

Application Event Publisher

import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import lombok.RequiredArgsConstructor;

@Service
@RequiredArgsConstructor
public class ProductApplicationService {
    private final ProductRepository productRepository;
    private final ApplicationEventPublisher eventPublisher;

    @Transactional
    public ProductResponse createProduct(CreateProductRequest request) {
        Product product = Product.create(
            request.getName(),
            request.getPrice(),
            request.getStock()
        );

        productRepository.save(product);

        // Publish domain events
        product.getDomainEvents().forEach(eventPublisher::publishEvent);
        product.clearDomainEvents();

        return mapToResponse(product);
    }
}

Distributed Event Publishing

Kafka Event Publisher

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Component
@RequiredArgsConstructor
@Slf4j
public class ProductEventPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publishProductCreatedEvent(ProductCreatedEvent event) {
        ProductCreatedEventDto dto = mapToDto(event);

        kafkaTemplate.send("product-events", event.getProductId().getValue(), dto)
            .whenComplete((result, ex) -> {
                if (ex == null) {
                    log.info("Published ProductCreatedEvent: {}", event.getProductId());
                } else {
                    log.error("Failed to publish ProductCreatedEvent", ex);
                }
            });
    }

    private ProductCreatedEventDto mapToDto(ProductCreatedEvent event) {
        return new ProductCreatedEventDto(
            event.getEventId().toString(),
            event.getProductId().getValue(),
            event.getName(),
            event.getPrice(),
            event.getStock(),
            event.getOccurredAt(),
            event.getCorrelationId().toString()
        );
    }
}

Event Publisher with Retry

import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.retry.annotation.Retryable;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

@Component
@RequiredArgsConstructor
public class ResilientEventPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final RetryTemplate retryTemplate;

    public void publishEvent(String topic, String key, Object payload) {
        retryTemplate.execute(context -> {
            try {
                kafkaTemplate.send(topic, key, payload).get();
                return true;
            } catch (Exception e) {
                log.error("Failed to publish event to topic: {}", topic, e);
                throw new EventPublishingException("Failed to publish event", e);
            }
        });
    }

    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1000L);

        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);

        retryTemplate.setBackOffPolicy(backOffPolicy);
        retryTemplate.setRetryPolicy(retryPolicy);

        return retryTemplate;
    }
}

Spring Cloud Stream Publishing

Functional Publisher

import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Component;
import lombok.RequiredArgsConstructor;

@Component
@RequiredArgsConstructor
public class ProductEventStreamPublisher {
    private final StreamBridge streamBridge;

    public void publishProductCreatedEvent(ProductCreatedEvent event) {
        ProductCreatedEventDto dto = mapToDto(event);
        streamBridge.send("productCreated-out-0", dto);
    }

    private ProductCreatedEventDto mapToDto(ProductCreatedEvent event) {
        return new ProductCreatedEventDto(
            event.getEventId().toString(),
            event.getProductId().getValue(),
            event.getName(),
            event.getPrice(),
            event.getStock(),
            event.getOccurredAt(),
            event.getCorrelationId().toString()
        );
    }
}

Producer Configuration

spring:
  cloud:
    stream:
      bindings:
        productCreated-out-0:
          destination: product-events
          producer:
            partition-count: 3
            required-groups: order-service, inventory-service
      kafka:
        bindings:
          productCreated-out-0:
            producer:
              configuration:
                acks: all
                retries: 3
                enable.idempotence: true

Event Publishing from Event Listener

Publishing Events After Handling

@Component
@RequiredArgsConstructor
public class OrderEventHandler {
    private final OrderRepository orderRepository;
    private final ApplicationEventPublisher eventPublisher;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Process order creation
        Order order = orderRepository.findById(event.getOrderId())
            .orElseThrow();

        // Publish derived events
        eventPublisher.publishEvent(new InventoryReservationRequestedEvent(
            event.getOrderId(),
            event.getItems()
        ));

        eventPublisher.publishEvent(new PaymentRequestedEvent(
            event.getOrderId(),
            event.getTotalAmount()
        ));
    }
}

Batching Events

Batch Event Publisher

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import lombok.RequiredArgsConstructor;

@Component
@RequiredArgsConstructor
public class BatchEventPublisher {
    private final List<DomainEvent> eventBuffer = new ArrayList<>();
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public synchronized void addEvent(DomainEvent event) {
        eventBuffer.add(event);

        if (eventBuffer.size() >= 100) {
            flush();
        }
    }

    @Scheduled(fixedDelay = 5000)
    public synchronized void flush() {
        if (eventBuffer.isEmpty()) {
            return;
        }

        try {
            eventBuffer.forEach(event -> {
                kafkaTemplate.send("events", event);
            });
        } finally {
            eventBuffer.clear();
        }
    }
}

Event Publishing Best Practices

1. Publish After Transaction Commit

@Transactional
public void executeBusinessOperation() {
    // Perform business logic
    aggregateRoot.performAction();

    // Save to database
    repository.save(aggregateRoot);

    // Publish events AFTER transaction commits
    TransactionSynchronizationManager.registerSynchronization(
        new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                aggregateRoot.getDomainEvents().forEach(eventPublisher::publishEvent);
                aggregateRoot.clearDomainEvents();
            }
        }
    );
}

2. Use Transactional Event Listener

@Component
public class EventForwardingHandler {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void forwardToKafka(DomainEvent event) {
        kafkaTemplate.send("events", event);
    }
}

3. Handle Publishing Failures

public void publishEventWithFallback(DomainEvent event) {
    try {
        kafkaTemplate.send("events", event).get(5, TimeUnit.SECONDS);
    } catch (Exception e) {
        log.error("Failed to publish event, storing in outbox", e);
        outboxRepository.save(OutboxEvent.from(event));
    }
}

Spring Boot Event-Driven Patterns - Examples

Comprehensive examples demonstrating event-driven architecture from basic local events to advanced distributed messaging.

Example 1: Basic Domain Events

A simple product lifecycle with domain events.

// Domain event
public class ProductCreatedEvent extends DomainEvent {
    private final String productId;
    private final String name;
    private final BigDecimal price;

    public ProductCreatedEvent(String productId, String name, BigDecimal price) {
        super();
        this.productId = productId;
        this.name = name;
        this.price = price;
    }

    // Getters
}

// Aggregate publishing events
@Getter
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class Product {
    private String id;
    private String name;
    private BigDecimal price;
    
    @Transient
    private List<DomainEvent> domainEvents = new ArrayList<>();

    public static Product create(String name, BigDecimal price) {
        Product product = new Product();
        product.id = UUID.randomUUID().toString();
        product.name = name;
        product.price = price;
        
        // Publish domain event
        product.publishEvent(new ProductCreatedEvent(product.id, name, price));
        
        return product;
    }

    protected void publishEvent(DomainEvent event) {
        domainEvents.add(event);
    }

    public List<DomainEvent> getDomainEvents() {
        return new ArrayList<>(domainEvents);
    }

    public void clearDomainEvents() {
        domainEvents.clear();
    }
}

---

Example 2: Local Event Publishing

Using ApplicationEventPublisher for in-process events.

// Application service
@Service
@Slf4j
@RequiredArgsConstructor
@Transactional
public class ProductApplicationService {
    private final ProductRepository productRepository;
    private final ApplicationEventPublisher eventPublisher;

    public ProductResponse createProduct(CreateProductRequest request) {
        Product product = Product.create(request.getName(), request.getPrice());
        Product saved = productRepository.save(product);
        
        // Publish domain events
        saved.getDomainEvents().forEach(event -> {
            log.debug("Publishing event: {}", event.getClass().getSimpleName());
            eventPublisher.publishEvent(event);
        });
        saved.clearDomainEvents();
        
        return mapper.toResponse(saved);
    }
}

// Event listener
@Component
@Slf4j
@RequiredArgsConstructor
public class ProductEventHandler {
    private final NotificationService notificationService;
    private final InventoryService inventoryService;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onProductCreated(ProductCreatedEvent event) {
        log.info("Handling ProductCreatedEvent");
        
        // Send notification
        notificationService.sendProductCreatedNotification(
            event.getName(), event.getPrice()
        );
        
        // Update inventory
        inventoryService.registerProduct(event.getProductId());
    }
}

// Test
@SpringBootTest
class ProductEventTest {
    @Autowired
    private ProductApplicationService productService;
    
    @MockBean
    private NotificationService notificationService;
    
    @Autowired
    private ProductRepository productRepository;

    @Test
    void shouldPublishProductCreatedEvent() {
        // Act
        productService.createProduct(
            new CreateProductRequest("Laptop", BigDecimal.valueOf(999.99))
        );

        // Assert - Event was handled
        verify(notificationService).sendProductCreatedNotification(
            "Laptop", BigDecimal.valueOf(999.99)
        );
    }
}

---

Example 3: Transactional Outbox Pattern

Ensures reliable event publishing even on failures.

// Outbox entity
@Entity
@Table(name = "outbox_events")
@Getter
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class OutboxEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    private String aggregateId;
    private String eventType;
    
    @Column(columnDefinition = "TEXT")
    private String payload;
    
    private LocalDateTime createdAt;
    private LocalDateTime publishedAt;
    private Integer retryCount;
}

// Application service using outbox
@Service
@Slf4j
@RequiredArgsConstructor
@Transactional
public class ProductApplicationService {
    private final ProductRepository productRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;

    public ProductResponse createProduct(CreateProductRequest request) {
        Product product = Product.create(request.getName(), request.getPrice());
        Product saved = productRepository.save(product);
        
        // Store event in outbox (same transaction)
        saved.getDomainEvents().forEach(event -> {
            try {
                String payload = objectMapper.writeValueAsString(event);
                OutboxEvent outboxEvent = OutboxEvent.builder()
                    .aggregateId(saved.getId())
                    .eventType(event.getClass().getSimpleName())
                    .payload(payload)
                    .createdAt(LocalDateTime.now())
                    .retryCount(0)
                    .build();
                
                outboxRepository.save(outboxEvent);
                log.debug("Outbox event created: {}", event.getClass().getSimpleName());
            } catch (Exception e) {
                log.error("Failed to create outbox event", e);
                throw new RuntimeException(e);
            }
        });
        
        return mapper.toResponse(saved);
    }
}

// Scheduled publisher
@Component
@Slf4j
@RequiredArgsConstructor
public class OutboxEventPublisher {
    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

    @Scheduled(fixedDelay = 5000)
    @Transactional
    public void publishPendingEvents() {
        List<OutboxEvent> pending = outboxRepository.findByPublishedAtIsNull();
        
        for (OutboxEvent event : pending) {
            try {
                kafkaTemplate.send("product-events", 
                    event.getAggregateId(), event.getPayload());
                
                event.setPublishedAt(LocalDateTime.now());
                outboxRepository.save(event);
                
                log.info("Published outbox event: {}", event.getId());
            } catch (Exception e) {
                log.error("Failed to publish event: {}", event.getId(), e);
                event.setRetryCount(event.getRetryCount() + 1);
                outboxRepository.save(event);
            }
        }
    }
}

---

Example 4: Kafka Event Publishing

Distributed event publishing with Spring Cloud Stream.

// Application configuration
@Configuration
public class KafkaConfig {
    
    @Bean
    public ObjectMapper objectMapper() {
        return new ObjectMapper()
            .registerModule(new JavaTimeModule())
            .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
    }
}

// Event publisher
@Component
@Slf4j
@RequiredArgsConstructor
public class KafkaProductEventPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publishProductCreatedEvent(ProductCreatedEvent event) {
        log.info("Publishing ProductCreatedEvent to Kafka: {}", event.getProductId());
        
        kafkaTemplate.send("product-events", 
            event.getProductId(),
            event);
    }
}

// Event consumer
@Component
@Slf4j
@RequiredArgsConstructor
public class ProductEventStreamConsumer {
    private final InventoryService inventoryService;

    @Bean
    public java.util.function.Consumer<ProductCreatedEvent> productCreatedConsumer() {
        return event -> {
            log.info("Consumed ProductCreatedEvent: {}", event.getProductId());
            inventoryService.registerProduct(event.getProductId(), event.getName());
        };
    }

    @Bean
    public java.util.function.Consumer<ProductUpdatedEvent> productUpdatedConsumer() {
        return event -> {
            log.info("Consumed ProductUpdatedEvent: {}", event.getProductId());
            inventoryService.updateProduct(event.getProductId(), event.getPrice());
        };
    }
}

// Application properties

application.yml:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: product-service
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"

  cloud:
    stream:
      bindings:
        productCreatedConsumer-in-0:
          destination: product-events
          group: product-inventory-service
        productUpdatedConsumer-in-0:
          destination: product-events
          group: product-inventory-service

---

Example 5: Event Saga Pattern

Coordinating multiple services with events.

// Events
public class OrderPlacedEvent extends DomainEvent {
    private final String orderId;
    private final String productId;
    private final Integer quantity;
    // ...
}

public class OrderPaymentConfirmedEvent extends DomainEvent {
    private final String orderId;
    // ...
}

// Saga orchestrator
@Component
@Slf4j
@RequiredArgsConstructor
public class OrderFulfillmentSaga {
    private final OrderService orderService;
    private final PaymentService paymentService;
    private final InventoryService inventoryService;
    private final ApplicationEventPublisher eventPublisher;

    @Transactional
    @EventListener
    public void onOrderPlaced(OrderPlacedEvent event) {
        log.info("Starting order fulfillment saga for order: {}", event.getOrderId());
        
        try {
            // Step 1: Reserve inventory
            inventoryService.reserveStock(event.getProductId(), event.getQuantity());
            log.info("Inventory reserved for order: {}", event.getOrderId());
            
            // Step 2: Process payment
            paymentService.processPayment(event.getOrderId());
            log.info("Payment processed for order: {}", event.getOrderId());
            
            // Step 3: Publish confirmation
            eventPublisher.publishEvent(new OrderPaymentConfirmedEvent(event.getOrderId()));
            
            // Step 4: Update order status
            orderService.markAsConfirmed(event.getOrderId());
            log.info("Order confirmed: {}", event.getOrderId());
            
        } catch (PaymentFailedException e) {
            log.warn("Payment failed, releasing inventory");
            inventoryService.releaseStock(event.getProductId(), event.getQuantity());
            orderService.markAsFailed(event.getOrderId(), e.getMessage());
        }
    }
}

// Test
@SpringBootTest
class OrderFulfillmentSagaTest {
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @MockBean
    private InventoryService inventoryService;
    
    @MockBean
    private PaymentService paymentService;
    
    @MockBean
    private OrderService orderService;

    @Test
    void shouldCompleteOrderFulfillmentSaga() {
        // Arrange
        OrderPlacedEvent event = new OrderPlacedEvent("order-123", "product-456", 2);

        // Act
        eventPublisher.publishEvent(event);

        // Assert
        verify(inventoryService).reserveStock("product-456", 2);
        verify(paymentService).processPayment("order-123");
        verify(orderService).markAsConfirmed("order-123");
    }
}

---

Example 6: Event Sourcing Foundation

Storing state changes as events.

// Event store
@Repository
public interface EventStoreRepository extends JpaRepository<StoredEvent, UUID> {
    List<StoredEvent> findByAggregateIdOrderBySequenceAsc(String aggregateId);
}

// Stored event
@Entity
@Table(name = "event_store")
@Getter
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class StoredEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    private String aggregateId;
    private String eventType;
    private Integer sequence;
    
    @Column(columnDefinition = "TEXT")
    private String payload;
    
    private LocalDateTime occurredAt;
}

// Event sourcing service
@Service
@Slf4j
@RequiredArgsConstructor
public class EventSourcingService {
    private final EventStoreRepository eventStoreRepository;
    private final ObjectMapper objectMapper;

    @Transactional
    public void storeEvent(String aggregateId, DomainEvent event) {
        try {
            List<StoredEvent> existing = eventStoreRepository
                .findByAggregateIdOrderBySequenceAsc(aggregateId);
            
            Integer nextSequence = existing.isEmpty() ? 1 : 
                existing.get(existing.size() - 1).getSequence() + 1;
            
            StoredEvent storedEvent = StoredEvent.builder()
                .aggregateId(aggregateId)
                .eventType(event.getClass().getSimpleName())
                .sequence(nextSequence)
                .payload(objectMapper.writeValueAsString(event))
                .occurredAt(LocalDateTime.now())
                .build();
            
            eventStoreRepository.save(storedEvent);
            log.info("Event stored: {} for aggregate: {}", 
                event.getClass().getSimpleName(), aggregateId);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Failed to store event", e);
        }
    }

    public List<DomainEvent> getEventHistory(String aggregateId) {
        return eventStoreRepository
            .findByAggregateIdOrderBySequenceAsc(aggregateId)
            .stream()
            .map(this::deserializeEvent)
            .collect(Collectors.toList());
    }

    private DomainEvent deserializeEvent(StoredEvent stored) {
        try {
            Class<?> eventClass = Class.forName(
                "com.example.product.domain.event." + stored.getEventType());
            return (DomainEvent) objectMapper.readValue(stored.getPayload(), eventClass);
        } catch (Exception e) {
            throw new RuntimeException("Failed to deserialize event", e);
        }
    }
}

These examples cover local events, transactional outbox pattern, Kafka publishing, saga coordination, and event sourcing foundations for comprehensive event-driven architecture.

Transactional Outbox Pattern

Outbox Entity

Basic Outbox Event

import jakarta.persistence.*;
import lombok.*;
import java.time.LocalDateTime;
import java.util.UUID;

@Entity
@Table(name = "outbox_events")
@Getter
@Setter
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class OutboxEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(nullable = false)
    private String aggregateId;

    @Column(nullable = false)
    private String aggregateType;

    @Column(nullable = false)
    private String eventType;

    @Lob
    @Column(nullable = false)
    private String payload;

    @Column(nullable = false)
    private UUID correlationId;

    @Column(nullable = false)
    private LocalDateTime createdAt;

    private LocalDateTime publishedAt;

    @Column(nullable = false)
    @Builder.Default
    private Integer retryCount = 0;

    private String errorMessage;

    private LocalDateTime lastAttemptAt;
}

Outbox Repository

import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;
import java.util.List;

@Repository
public interface OutboxEventRepository extends JpaRepository<OutboxEvent, UUID> {

    List<OutboxEvent> findByPublishedAtNullOrderByCreatedAtAsc();

    List<OutboxEvent> findByPublishedAtNullAndRetryCountLessThanOrderByCreatedAtAsc(Integer maxRetries);

    @Query("""
        SELECT e FROM OutboxEvent e
        WHERE e.publishedAt IS NULL
        AND e.retryCount < :maxRetries
        AND (e.lastAttemptAt IS NULL OR e.lastAttemptAt < :threshold)
        ORDER BY e.createdAt ASC
    """)
    List<OutboxEvent> findPendingEvents(
        Integer maxRetries,
        LocalDateTime threshold
    );
}

Outbox Event Creation

Save Outbox Event with Aggregate

import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import lombok.RequiredArgsConstructor;

@Service
@RequiredArgsConstructor
public class OrderApplicationService {
    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;

    @Transactional
    public OrderResponse createOrder(CreateOrderRequest request) {
        Order order = Order.create(
            request.getCustomerId(),
            request.getItems()
        );

        orderRepository.save(order);

        // Create outbox events atomically with order
        order.getDomainEvents().forEach(domainEvent -> {
            try {
                OutboxEvent outboxEvent = OutboxEvent.builder()
                    .aggregateId(order.getId().getValue())
                    .aggregateType("Order")
                    .eventType(domainEvent.getClass().getSimpleName())
                    .payload(objectMapper.writeValueAsString(domainEvent))
                    .correlationId(domainEvent.getCorrelationId())
                    .createdAt(LocalDateTime.now())
                    .build();

                outboxRepository.save(outboxEvent);
            } catch (JsonProcessingException e) {
                throw new EventSerializationException("Failed to serialize event", e);
            }
        });

        order.clearDomainEvents();

        return mapToResponse(order);
    }
}

Outbox Event Processor

Scheduled Event Publisher

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Component
@RequiredArgsConstructor
@Slf4j
public class OutboxEventProcessor {
    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Scheduled(fixedDelay = 5000)
    @Transactional
    public void processPendingEvents() {
        List<OutboxEvent> pendingEvents = outboxRepository.findByPublishedAtNullOrderByCreatedAtAsc();

        for (OutboxEvent event : pendingEvents) {
            try {
                publishEvent(event);
                event.setPublishedAt(LocalDateTime.now());
                outboxRepository.save(event);

                log.info("Published outbox event: {}", event.getId());

            } catch (Exception e) {
                handlePublishFailure(event, e);
            }
        }
    }

    private void publishEvent(OutboxEvent event) throws JsonProcessingException {
        String topic = determineTopic(event.getEventType());

        kafkaTemplate.send(
            topic,
            event.getAggregateId(),
            event.getPayload()
        ).get(5, TimeUnit.SECONDS);
    }

    private void handlePublishFailure(OutboxEvent event, Exception e) {
        log.error("Failed to publish outbox event: {}", event.getId(), e);

        event.setRetryCount(event.getRetryCount() + 1);
        event.setLastAttemptAt(LocalDateTime.now());
        event.setErrorMessage(e.getMessage());

        outboxRepository.save(event);

        if (event.getRetryCount() >= 3) {
            log.error("Max retries exceeded for event: {}", event.getId());
            // Send alert to monitoring system
        }
    }

    private String determineTopic(String eventType) {
        return switch (eventType) {
            case "OrderCreatedEvent", "OrderPaidEvent" -> "order-events";
            case "ProductCreatedEvent", "ProductStockDecreasedEvent" -> "product-events";
            default -> "default-events";
        };
    }
}

Idempotent Event Publisher

@Component
@RequiredArgsConstructor
public class IdempotentOutboxProcessor {
    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Scheduled(fixedDelay = 5000)
    @Transactional
    public void processPendingEvents() {
        LocalDateTime threshold = LocalDateTime.now().minusMinutes(5);
        List<OutboxEvent> pendingEvents = outboxRepository.findPendingEvents(3, threshold);

        for (OutboxEvent event : pendingEvents) {
            if (shouldProcessEvent(event)) {
                publishEvent(event);
            }
        }
    }

    private boolean shouldProcessEvent(OutboxEvent event) {
        // Don't process if recently attempted
        if (event.getLastAttemptAt() != null &&
            event.getLastAttemptAt().isAfter(LocalDateTime.now().minusMinutes(1))) {
            return false;
        }

        return true;
    }

    private void publishEvent(OutboxEvent event) {
        try {
            kafkaTemplate.send(
                determineTopic(event.getEventType()),
                event.getAggregateId(),
                event.getPayload()
            ).get();

            event.setPublishedAt(LocalDateTime.now());
            outboxRepository.save(event);

        } catch (Exception e) {
            event.setRetryCount(event.getRetryCount() + 1);
            event.setLastAttemptAt(LocalDateTime.now());
            event.setErrorMessage(e.getMessage());
            outboxRepository.save(event);
        }
    }
}

Cleanup Strategy

Purge Published Events

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.annotation.Transactional;

@Component
@RequiredArgsConstructor
public class OutboxCleanupService {
    private final OutboxEventRepository outboxRepository;

    @Scheduled(cron = "0 0 2 * * ?") // 2 AM daily
    @Transactional
    public void purgePublishedEvents() {
        LocalDateTime cutoff = LocalDateTime.now().minusDays(7);

        List<OutboxEvent> eventsToDelete = outboxRepository
            .findByPublishedAtBeforeAndPublishedAtIsNotNull(cutoff);

        outboxRepository.deleteAll(eventsToDelete);

        log.info("Purged {} published outbox events", eventsToDelete.size());
    }
}

Archive Old Events

@Scheduled(cron = "0 0 3 * * ?") // 3 AM daily
@Transactional
public void archivePublishedEvents() {
    LocalDateTime cutoff = LocalDateTime.now().minusDays(30);

    List<OutboxEvent> eventsToArchive = outboxRepository
        .findByPublishedAtBeforeAndPublishedAtIsNotNull(cutoff);

    // Move to archive table or external storage
    archiveService.archiveEvents(eventsToArchive);

    outboxRepository.deleteAll(eventsToArchive);

    log.info("Archived {} outbox events", eventsToArchive.size());
}

Outbox Pattern Variations

Optimistic Locking

@Entity
@Table(name = "outbox_events")
@Getter
@Setter
public class OutboxEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Version
    private Long version;

    // ... other fields
}

@Component
public class OptimisticLockingProcessor {
    @Transactional
    public void processEvent(OutboxEvent event) {
        try {
            publishEvent(event);

            event.setPublishedAt(LocalDateTime.now());
            outboxRepository.save(event);

        } catch (ObjectOptimisticLockingFailureException e) {
            log.warn("Concurrent modification detected for event: {}", event.getId());
            // Retry after delay
        }
    }
}

Batch Processing

@Component
public class BatchOutboxProcessor {
    private static final int BATCH_SIZE = 100;

    @Scheduled(fixedDelay = 5000)
    @Transactional
    public void processPendingEvents() {
        int page = 0;
        List<OutboxEvent> batch;

        do {
            Pageable pageable = PageRequest.of(page, BATCH_SIZE);
            batch = outboxRepository.findByPublishedAtNullOrderByCreatedAtAsc(pageable);

            if (!batch.isEmpty()) {
                publishBatch(batch);
            }

            page++;
        } while (batch.size() == BATCH_SIZE);
    }

    private void publishBatch(List<OutboxEvent> batch) {
        batch.forEach(event -> {
            try {
                publishEvent(event);
                event.setPublishedAt(LocalDateTime.now());
            } catch (Exception e) {
                event.setRetryCount(event.getRetryCount() + 1);
                event.setErrorMessage(e.getMessage());
            }
        });

        outboxRepository.saveAll(batch);
    }
}

Monitoring and Alerts

Outbox Metrics

@Component
@RequiredArgsConstructor
public class OutboxMetricsReporter {
    private final OutboxEventRepository outboxRepository;
    private final MeterRegistry meterRegistry;

    @Scheduled(fixedDelay = 60000)
    public void reportMetrics() {
        long pendingCount = outboxRepository.countByPublishedAtNull();
        long failedCount = outboxRepository.countByRetryCountGreaterThanEqual(3);

        meterRegistry.gauge("outbox.pending.events", pendingCount);
        meterRegistry.gauge("outbox.failed.events", failedCount);

        if (pendingCount > 1000) {
            log.warn("High outbox backlog: {} pending events", pendingCount);
        }

        if (failedCount > 100) {
            log.error("Many failed outbox events: {}", failedCount);
            // Send alert
        }
    }
}

Testing Event-Driven Applications

Unit Testing Domain Events

Test Domain Event Publishing

import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.*;

class ProductTest {

    @Test
    void shouldPublishProductCreatedEventOnCreation() {
        // When
        Product product = Product.create(
            "Test Product",
            BigDecimal.TEN,
            100
        );

        // Then
        assertThat(product.getDomainEvents()).hasSize(1);
        assertThat(product.getDomainEvents().get(0))
            .isInstanceOf(ProductCreatedEvent.class);

        ProductCreatedEvent event = (ProductCreatedEvent) product.getDomainEvents().get(0);
        assertThat(event.getName()).isEqualTo("Test Product");
        assertThat(event.getPrice()).isEqualByComparingTo(BigDecimal.TEN);
        assertThat(event.getStock()).isEqualTo(100);
    }

    @Test
    void shouldPublishStockDecreasedEvent() {
        // Given
        Product product = Product.create("Product", BigDecimal.TEN, 100);
        product.clearDomainEvents();

        // When
        product.decreaseStock(10);

        // Then
        assertThat(product.getDomainEvents()).hasSize(1);
        assertThat(product.getDomainEvents().get(0))
            .isInstanceOf(ProductStockDecreasedEvent.class);
    }

    @Test
    void shouldClearDomainEvents() {
        // Given
        Product product = Product.create("Product", BigDecimal.TEN, 100);

        // When
        product.clearDomainEvents();

        // Then
        assertThat(product.getDomainEvents()).isEmpty();
    }
}

Test Event Handlers

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;

@ExtendWith(MockitoExtension.class)
class ProductEventHandlerTest {

    @Mock
    private NotificationService notificationService;

    @Mock
    private AuditService auditService;

    @InjectMocks
    private ProductEventHandler handler;

    @Test
    void shouldHandleProductCreatedEvent() {
        // Given
        ProductId productId = ProductId.of("123");
        ProductCreatedEvent event = new ProductCreatedEvent(
            productId,
            "Test Product",
            BigDecimal.TEN,
            100
        );

        // When
        handler.onProductCreated(event);

        // Then
        verify(notificationService).sendProductCreatedNotification("Test Product");
        verify(auditService).logProductCreation(
            eq("123"),
            eq("Test Product"),
            eq(BigDecimal.TEN),
            any(UUID.class)
        );
    }

    @Test
    void shouldHandleProductStockDecreasedEvent() {
        // Given
        ProductId productId = ProductId.of("123");
        ProductStockDecreasedEvent event = new ProductStockDecreasedEvent(
            productId,
            10,
            90
        );

        // When
        handler.onProductStockDecreased(event);

        // Then
        verify(notificationService).sendStockUpdateNotification("123", 10);
    }
}

Integration Testing with Testcontainers

Kafka Integration Test

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.TestPropertySource;
import java.util.concurrent.TimeUnit;
import static org.awaitility.Awaitility.await;

@SpringBootTest
@EmbeddedKafka(partitions = 1, brokerProperties = {
    "listeners=PLAINTEXT://localhost:9092",
    "port=9092"
})
class KafkaEventIntegrationTest {

    @Autowired
    private ProductApplicationService productService;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private ProductEventConsumer consumer;

    @Test
    void shouldPublishEventToKafka() throws Exception {
        // Given
        CreateProductRequest request = new CreateProductRequest(
            "Test Product",
            BigDecimal.valueOf(99.99),
            50
        );

        // When
        ProductResponse response = productService.createProduct(request);

        // Then
        await()
            .atMost(5, TimeUnit.SECONDS)
            .untilAsserted(() -> {
                verify(consumer).handleProductCreated(any(ProductCreatedEventDto.class));
            });
    }
}

Full Integration Test with Testcontainers

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

@SpringBootTest
@Testcontainers
class EventDrivenIntegrationTest {

    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.5.0")
    );

    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }

    @Autowired
    private ProductApplicationService productService;

    @Test
    void shouldProcessEndToEnd() {
        // Create product
        CreateProductRequest request = new CreateProductRequest(
            "Test Product",
            BigDecimal.valueOf(99.99),
            50
        );

        ProductResponse response = productService.createProduct(request);

        assertThat(response.getId()).isNotNull();
        assertThat(response.getName()).isEqualTo("Test Product");
    }
}

Testing Event Handlers with @TransactionalEventListener

Test Transactional Event Listener

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.transaction.annotation.Transactional;
import org.junit.jupiter.api.Test;
import static org.mockito.ArgumentCaptor.forClass;
import static org.mockito.Mockito.verify;

@SpringBootTest
class TransactionalEventListenerTest {

    @Autowired
    private ProductApplicationService productService;

    @Autowired
    private ProductEventHandler eventHandler;

    @Test
    @Transactional
    void shouldPublishEventAfterTransactionCommit() {
        // Given
        CreateProductRequest request = new CreateProductRequest(
            "Test Product",
            BigDecimal.TEN,
            100
        );

        // When
        productService.createProduct(request);

        // Then - Transaction will commit and event will be published
        // Use TransactionalEventListener test support
    }
}

Test Event Publishing with ApplicationEventPublisher

import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.context.ApplicationEventPublisher;
import org.junit.jupiter.api.Test;
import static org.mockito.Mockito.*;

@SpringBootTest
class ApplicationEventPublisherTest {

    @Autowired
    private ProductApplicationService productService;

    @MockBean
    private ApplicationEventPublisher eventPublisher;

    @Test
    void shouldPublishEvents() {
        // Given
        CreateProductRequest request = new CreateProductRequest(
            "Test Product",
            BigDecimal.TEN,
            100
        );

        // When
        productService.createProduct(request);

        // Then
        verify(eventPublisher).publishEvent(any(ProductCreatedEvent.class));
    }
}

Testing Kafka Consumers

Test Kafka Listener

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
@EmbeddedKafka
class KafkaConsumerTest {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private ProductEventConsumer consumer;

    @Test
    void shouldConsumeProductCreatedEvent() throws Exception {
        // Given
        ProductCreatedEventDto event = new ProductCreatedEventDto(
            UUID.randomUUID().toString(),
            "123",
            "Test Product",
            BigDecimal.TEN,
            100,
            LocalDateTime.now(),
            UUID.randomUUID().toString()
        );

        // When
        kafkaTemplate.send("product-events", "123", event).get();

        // Then
        // Wait for async processing
        TimeUnit.SECONDS.sleep(2);

        verify(consumer).handleProductCreated(event);
    }
}

Testing Spring Cloud Stream

Test Stream Functions

import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.Import;
import org.junit.jupiter.api.Test;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;

@SpringBootTest
@Import(TestChannelBinderConfiguration.class)
class StreamFunctionTest {

    @Autowired
    private InputDestination input;

    @Autowired
    private OutputDestination output;

    @Test
    void shouldProcessProductCreatedEvent() {
        // Given
        ProductCreatedEventDto event = new ProductCreatedEventDto(
            UUID.randomUUID().toString(),
            "123",
            "Test Product",
            BigDecimal.TEN,
            100,
            LocalDateTime.now(),
            UUID.randomUUID().toString()
        );

        Message<ProductCreatedEventDto> message = new GenericMessage<>(event);

        // When
        input.send(message, "productCreated-in-0");

        // Then
        Message<byte[]> result = output.receive(1000, "productCreated-out-0");
        assertThat(result).isNotNull();
    }
}

Testing Event Sourcing Scenarios

Test Event Reconstruction

class EventSourcingTest {

    @Test
    void shouldReconstructAggregateFromEvents() {
        // Given
        List<DomainEvent> events = List.of(
            new ProductCreatedEvent(ProductId.of("123"), "Product", BigDecimal.TEN, 100),
            new ProductStockDecreasedEvent(ProductId.of("123"), 10, 90),
            new ProductStockDecreasedEvent(ProductId.of("123"), 5, 85)
        );

        // When
        Product product = Product.replay(events);

        // Then
        assertThat(product.getStock()).isEqualTo(85);
    }

    @Test
    void shouldRebuildStateFromEventStream() {
        // Given
        EventStream eventStream = eventStore.getEvents(ProductId.of("123"));

        // When
        Product product = Product.rebuild(eventStream);

        // Then
        assertThat(product).isNotNull();
        assertThat(product.getStock()).isEqualTo(85);
    }
}

Testing Error Scenarios

Test Event Handler Failure

@Test
void shouldHandleEventProcessingFailure() {
    // Given
    ProductCreatedEvent event = new ProductCreatedEvent(
        ProductId.of("123"),
        "Test Product",
        BigDecimal.TEN,
        100
    );

    doThrow(new RuntimeException("Processing failed"))
        .when(orderService).onProductCreated(any());

    // When
    assertThatThrownBy(() -> consumer.handleProductCreated(event))
        .isInstanceOf(RuntimeException.class)
        .hasMessage("Processing failed");

    // Then
    verify(orderService, times(3)).onProductCreated(any()); // Retry attempts
}

Test Dead Letter Queue

@Test
void shouldSendFailedEventsToDLQ() {
    // Given
    ProductCreatedEventDto event = createEvent();

    doThrow(new RuntimeException("Max retries exceeded"))
        .when(orderService).onProductCreated(any());

    // When
    kafkaTemplate.send("product-events", event);

    // Then
    await().atMost(10, TimeUnit.SECONDS)
        .untilAsserted(() -> {
            Message<?> dltMessage = kafkaTemplate.receive("product-events-dlt");
            assertThat(dltMessage).isNotNull();
        });
}