Event-driven architecture has transformed how we build scalable, loosely-coupled systems. Here is my experience implementing Kafka-based solutions in enterprise environments.

Why Event-Driven?

Traditional request-response architecture creates tight coupling. Event-driven architecture provides:

  • Loose coupling between services

  • Better scalability

  • Improved fault tolerance

  • Natural audit trail

Kafka Concepts

Producer → Topic → Partition → Consumer Group → Consumer

Order Service (Producer)
       ↓
   [order-events] Topic
   ├── Partition 0: [event1, event4, event7...]
   ├── Partition 1: [event2, event5, event8...]
   └── Partition 2: [event3, event6, event9...]
       ↓
Consumer Group: notification-service
├── Consumer 1 → Partition 0
├── Consumer 2 → Partition 1
└── Consumer 3 → Partition 2

Spring Kafka Configuration

@Configuration
public class KafkaConfig {

    @Bean
    public ProducerFactory<String, OrderEvent> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        config.put(ProducerConfig.RETRIES_CONFIG, 3);
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, OrderEvent> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public ConsumerFactory<String, OrderEvent> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service");
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        return new DefaultKafkaConsumerFactory<>(config,
            new StringDeserializer(),
            new JsonDeserializer<>(OrderEvent.class));
    }
}

Event Publishing

@Service
@RequiredArgsConstructor
public class OrderService {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    private final OrderRepository orderRepository;

    @Transactional
    public Order createOrder(CreateOrderRequest request) {
        Order order = Order.builder()
            .customerId(request.getCustomerId())
            .items(request.getItems())
            .status(OrderStatus.CREATED)
            .build();

        Order savedOrder = orderRepository.save(order);

        // Publish event
        OrderEvent event = OrderEvent.builder()
            .eventId(UUID.randomUUID().toString())
            .eventType("ORDER_CREATED")
            .orderId(savedOrder.getId())
            .customerId(savedOrder.getCustomerId())
            .timestamp(Instant.now())
            .build();

        kafkaTemplate.send("order-events", savedOrder.getId().toString(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to publish event", ex);
                } else {
                    log.info("Event published: {}", event.getEventId());
                }
            });

        return savedOrder;
    }
}

Event Consumption

@Service
@RequiredArgsConstructor
public class NotificationConsumer {

    private final NotificationService notificationService;

    @KafkaListener(
        topics = "order-events",
        groupId = "notification-service",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            Acknowledgment ack) {

        try {
            log.info("Received event: {} for order: {}",
                event.getEventType(), event.getOrderId());

            switch (event.getEventType()) {
                case "ORDER_CREATED" ->
                    notificationService.sendOrderConfirmation(event);
                case "ORDER_SHIPPED" ->
                    notificationService.sendShippingNotification(event);
                case "ORDER_DELIVERED" ->
                    notificationService.sendDeliveryNotification(event);
            }

            ack.acknowledge();
        } catch (Exception e) {
            log.error("Error processing event", e);
            // Don not acknowledge - message will be redelivered
        }
    }
}

Error Handling with Dead Letter Topic

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
        kafkaListenerContainerFactory() {

    ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setCommonErrorHandler(new DefaultErrorHandler(
        new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, ex) -> new TopicPartition(
                record.topic() + ".DLT",
                record.partition())),
        new FixedBackOff(1000L, 3)));
    return factory;
}

Best Practices

  1. Idempotent Consumers: Handle duplicate messages gracefully
  2. Schema Registry: Use Avro/Protobuf for schema evolution
  3. Monitoring: Track lag, throughput, and error rates
  4. Partition Strategy: Choose keys wisely for ordering
  5. Exactly-Once Semantics: Use transactions when needed

Event-driven architecture requires careful design but provides immense scalability benefits.