← Back to the dispatchesSystem Design · 14 min read

System Design

Event-Driven Architecture with Apache Kafka

Learn how to implement event-driven architecture using Apache Kafka and Spring Boot, with practical examples for building scalable, resilient systems.

Utsab Dahal

· Backend Developer — Spring Boot & DevOps28 December 2023 · 14 min read

Event-driven architecture (EDA) is a powerful paradigm for building scalable, loosely coupled, and resilient distributed systems. This comprehensive guide explores EDA concepts, Apache Kafka fundamentals, and practical implementation using Spring Boot. With detailed code examples and best practices, you'll learn how to design and deploy event-driven systems for real-world applications.

Event-Driven Architecture

Event-driven architecture enables services to communicate asynchronously through events, reducing coupling and improving scalability. Events represent significant changes in the system, such as a user registration or an order placement, and are processed by interested services.

Core Concepts

Events: Immutable records of something that happened (e.g., "OrderCreated").
Producers: Services that generate and publish events to a message broker.
Consumers: Services that subscribe to and process events.
Event Store: A persistent storage system (like Kafka) for events.
Message Broker: A system that routes events between producers and consumers.

Benefits

01Loose Coupling: Services communicate via events, not direct API calls, reducing dependencies.
02Scalability: Producers and consumers can scale independently.
03Resilience: Asynchronous processing ensures the system remains operational if a service fails.
04Auditability: Events provide a historical record of system activities.
05Flexibility: New consumers can be added without modifying producers.

Real-World Use Case

Consider an e-commerce platform where an order placement triggers multiple actions: updating inventory, sending a confirmation email, and processing payment. In a traditional synchronous system, these actions would be tightly coupled, increasing complexity. With EDA, the order service publishes an "OrderCreated" event, and independent services (inventory, notification, payment) consume it asynchronously, improving scalability and fault tolerance.

Apache Kafka Basics

Apache Kafka is a distributed streaming platform designed for high-throughput, fault-tolerant, and scalable event processing.

Key Components

Topics: Categories where events are published (e.g., "order-events").
Partitions: Subdivisions of topics for parallel processing and scalability.
Producers: Applications that send events to topics.
Consumers: Applications that read events from topics.
Brokers: Kafka servers that store and manage events.
Consumer Groups: Groups of consumers that share the load of processing events.

Setting Up Kafka with Spring Boot

Add the necessary dependency to your pom.xml:

xml

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

Configure Kafka in application.yml:

yaml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: notification-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: com.example.event
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

Spring Kafka Integration

Below is an example of integrating Kafka with Spring Boot to produce and consume events in an e-commerce system.

Event Definition

Define event classes for serialization:

java

public class OrderCreatedEvent {
    private Long orderId;
    private Long customerId;
    private Double total;
    private Instant timestamp;

    // Constructors, getters, and setters
    public OrderCreatedEvent(Long orderId, Long customerId, Double total, Instant timestamp) {
        this.orderId = orderId;
        this.customerId = customerId;
        this.total = total;
        this.timestamp = timestamp;
    }
}

Producing Events

Create a service to publish events when an order is created:

java

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderEventProducer {
    private static final Logger logger = LoggerFactory.getLogger(OrderEventProducer.class);
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public OrderEventProducer(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishOrderCreated(Order order) {
        OrderCreatedEvent event = new OrderCreatedEvent(
                order.getId(),
                order.getCustomerId(),
                order.getTotal(),
                Instant.now()
        );
        kafkaTemplate.send("order-events", order.getId().toString(), event)
                .addCallback(
                        result -> logger.info("Sent order event for order ID: {}", order.getId()),
                        ex -> logger.error("Failed to send order event: {}", ex.getMessage())
                );
    }
}

Consuming Events

Create a consumer to process order events:

java

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class OrderEventConsumer {
    private static final Logger logger = LoggerFactory.getLogger(OrderEventConsumer.class);
    private final NotificationService notificationService;

    public OrderEventConsumer(NotificationService notificationService) {
        this.notificationService = notificationService;
    }

    @KafkaListener(topics = "order-events", groupId = "notification-service")
    public void handleOrderCreated(OrderCreatedEvent event) {
        logger.info("Received order created event for order ID: {}", event.getOrderId());
        notificationService.sendOrderConfirmation(event.getCustomerId(), event.getOrderId());
    }
}

Order Service Integration

Integrate the event producer into the order service:

java

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

@Service
@Transactional
public class OrderService {
    private final OrderRepository orderRepository;
    private final OrderEventProducer eventProducer;

    public OrderService(OrderRepository orderRepository, OrderEventProducer eventProducer) {
        this.orderRepository = orderRepository;
        this.eventProducer = eventProducer;
    }

    public Order createOrder(Order order) {
        Order savedOrder = orderRepository.save(order);
        eventProducer.publishOrderCreated(savedOrder);
        return savedOrder;
    }
}

Entity:

java

import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Table;

@Entity
@Table(name = "orders")
public class Order {
    @Id
    private Long id;
    private Long customerId;
    private Double total;
    private String status;

    // Constructors, getters, and setters
}

Repository:

java

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

public interface OrderRepository extends JpaRepository<Order, Long> {
    List<Order> findByCustomerId(Long customerId);
}

Controller:

java

import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/orders")
public class OrderController {
    private final OrderService orderService;

    public OrderController(OrderService orderService) {
        this.orderService = orderService;
    }

    @PostMapping
    public ResponseEntity<Order> createOrder(@RequestBody Order order) {
        Order createdOrder = orderService.createOrder(order);
        return ResponseEntity.status(HttpStatus.CREATED).body(createdOrder);
    }
}

Event Patterns

Event-driven systems leverage patterns like event sourcing, CQRS, and Saga to address complex requirements.

Event Sourcing

Event sourcing stores the state of an application as a sequence of events, enabling auditability and state reconstruction.

java

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderEventSourcingService {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public OrderEventSourcingService(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void saveOrderEvent(OrderEvent event) {
        kafkaTemplate.send("order-state-events", event.getOrderId().toString(), event);
    }
}

public class OrderEvent {
    private Long orderId;
    private String eventType; // e.g., CREATED, UPDATED, CANCELLED
    private Map<String, Object> payload;
    private Instant timestamp;

    // Constructors, getters, and setters
}

CQRS (Command Query Responsibility Segregation)

CQRS separates read and write operations, optimizing performance for each.

java

@Service
public class OrderQueryService {
    private final OrderReadRepository readRepository;

    public OrderQueryService(OrderReadRepository readRepository) {
        this.readRepository = readRepository;
    }

    public OrderSummary getOrderSummary(Long orderId) {
        return readRepository.findSummaryById(orderId)
                .orElseThrow(() -> new RuntimeException("Order summary not found"));
    }
}

public interface OrderReadRepository extends JpaRepository<OrderSummary, Long> {
    Optional<OrderSummary> findSummaryById(Long orderId);
}

public class OrderSummary {
    private Long id;
    private Long customerId;
    private Double total;
    private String status;

    // Constructors, getters, and setters
}

Saga Pattern

The Saga pattern manages distributed transactions across microservices using a series of local transactions.

java

@Service
public class OrderSagaOrchestrator {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public OrderSagaOrchestrator(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void startOrderSaga(Order order) {
        OrderSagaEvent event = new OrderSagaEvent(order.getId(), "ORDER_CREATED", order);
        kafkaTemplate.send("order-saga", order.getId().toString(), event);
    }

    @KafkaListener(topics = "order-saga", groupId = "saga-orchestrator")
    public void handleSagaEvent(OrderSagaEvent event) {
        switch (event.getStatus()) {
            case "ORDER_CREATED":
                // Trigger inventory check
                kafkaTemplate.send("inventory-saga", event.getOrderId().toString(), event);
                break;
            case "INVENTORY_CONFIRMED":
                // Trigger payment processing
                kafkaTemplate.send("payment-saga", event.getOrderId().toString(), event);
                break;
            case "PAYMENT_COMPLETED":
                // Finalize order
                completeOrderSaga(event.getOrderId());
                break;
        }
    }
}

Error Handling

Robust error handling ensures reliability in event-driven systems.

Dead Letter Queue (DLQ)

Handle failed events by sending them to a DLQ for later analysis.

java

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

    public OrderEventConsumer(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @KafkaListener(topics = "order-events", groupId = "notification-service")
    public void handleOrderCreated(ConsumerRecord<String, Object> record) {
        try {
            OrderCreatedEvent event = (OrderCreatedEvent) record.value();
            notificationService.sendOrderConfirmation(event.getCustomerId(), event.getOrderId());
        } catch (Exception e) {
            kafkaTemplate.send("order-events-dlq", record.key(), record.value());
            logger.error("Failed to process event, sent to DLQ: {}", e.getMessage());
        }
    }
}

Retry Mechanism

Configure retries for transient failures:

java

@Component
public class OrderEventConsumer {
    private static final Logger logger = LoggerFactory.getLogger(OrderEventConsumer.class);

    @KafkaListener(topics = "order-events", groupId = "notification-service",
                   errorHandler = "kafkaListenerErrorHandler")
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Process event
        notificationService.sendOrderConfirmation(event.getCustomerId(), event.getOrderId());
    }

    @Bean
    public KafkaListenerErrorHandler kafkaListenerErrorHandler() {
        return (message, exception) -> {
            logger.error("Retrying event processing: {}", message.getPayload());
            throw new RetryableException("Retrying event processing", exception.getCause());
        };
    }
}

Retry Configuration (in application.yml):

yaml

spring:
  kafka:
    listener:
      retry:
        max-attempts: 3
        initial-interval: 1000
        multiplier: 2

Monitoring and Observability

Monitor Kafka clusters and Spring Boot applications to ensure performance and reliability.

Kafka Monitoring

Use tools like Confluent Control Center or Prometheus with Kafka Exporter.

Prometheus Configuration (in application.yml):

yaml

management:
  endpoints:
    web:
      exposure:
        include: prometheus, health, info

Prometheus Scrape Config (prometheus.yml):

yaml

scrape_configs:
  - job_name: 'order-service'
    metrics_path: '/actuator/prometheus'
    static_configs:
    - targets: ['order-service:8082']

Centralized Logging

Aggregate logs using ELK Stack or Fluentd:

java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Component
public class LogService {
    private static final Logger logger = LoggerFactory.getLogger(LogService.class);

    public void logEvent(String eventType, String details) {
        logger.info("Event processed: type={}, details={}", eventType, details);
    }
}

Best Practices

01Event Schema Design:

Use clear, self-descriptive event schemas with versioning.
Example:

java

     public class OrderEvent {
         private String version = "1.0";
         private String eventType;
         private Map<String, Object> payload;

         // Constructors, getters, and setters
     }

01Idempotent Consumers:

Ensure consumers can handle duplicate events.
Example:

java

     @Component
     public class IdempotentConsumer {
         private final Set<String> processedEvents = new HashSet<>();

         @KafkaListener(topics = "order-events", groupId = "notification-service")
         public void handleOrderCreated(OrderCreatedEvent event) {
             String eventId = event.getOrderId() + "-" + event.getTimestamp();
             if (processedEvents.contains(eventId)) {
                 logger.info("Duplicate event ignored: {}", eventId);
                 return;
             }
             processedEvents.add(eventId);
             notificationService.sendOrderConfirmation(event.getCustomerId(), event.getOrderId());
         }
     }

01Message Ordering:

Use partition keys to ensure order within a partition.
Example: Use order.getId().toString() as the key in kafkaTemplate.send.

01Monitoring and Alerts:

Set up alerts for consumer lag and broker health.
Use Kafka Exporter with Prometheus for metrics.

01Security:

Secure Kafka with SSL/TLS and SASL.
Example configuration:

yaml

     spring:
       kafka:
         properties:
           security.protocol: SSL
           ssl.truststore.location: /path/to/truststore.jks
           ssl.truststore.password: password

01Testing:

Use Testcontainers for integration testing with Kafka.
Example:

java

     @SpringBootTest
     @Testcontainers
     class OrderEventProducerTest {
         @Container
         private static final KafkaContainer kafka = new KafkaContainer(
             DockerImageName.parse("confluentinc/cp-kafka:7.3.0")
         );

         @Autowired
         private OrderEventProducer producer;

         @Test
         void shouldPublishOrderEvent() {
             Order order = new Order(1L, 100L, 99.99, "PENDING");
             producer.publishOrderCreated(order);
             // Verify event in Kafka topic
         }
     }

Conclusion

Event-driven architecture with Apache Kafka enables scalable, resilient, and loosely coupled systems. By integrating Kafka with Spring Boot, you can build robust applications that handle high-throughput event processing. Key takeaways:

Use Kafka for asynchronous, event-driven communication.
Implement event sourcing, CQRS, and Saga patterns for complex workflows.
Handle errors with DLQs and retries.
Monitor Kafka clusters and Spring Boot services with Prometheus and centralized logging.
Follow best practices like idempotent consumers and secure configurations.

Call to Action: Start building your event-driven application using the examples above. Experiment with Kafka locally using Docker and scale to production with proper monitoring. Share your experiences or questions in the comments or on X to join the developer community!