Week 10: Microservices & Messaging (Part 2) — Event-Driven Messaging

1. Sync vs Async Communication

In Week 9, marking a task complete meant the Orders service made a direct RestClient call to the Notification service and waited for a response, with a Resilience4j circuit breaker so a struggling Notification service couldn't take Orders down with it. That design is synchronous coupling: the caller blocks on the callee, and — circuit breaker or not — the caller's success is still tied to the callee being reachable at all, right now.

Event-driven messaging removes that coupling. Instead of Orders asking "please send a notification," Orders announces a fact that already happened — TaskCompleted — onto a broker, and walks away. It doesn't know or care who's listening. The Notification service picks the event up whenever it can and reacts to it; if it's down for five minutes, the event is still there when it comes back. A future service — say, an analytics service tracking completion rates — can start listening to the exact same event stream without Orders changing a single line.

When each is the right choice

Synchronous calls are still correct when the caller genuinely needs an answer before it can continue — checking inventory before confirming an order, or validating a payment before shipping. Asynchronous events fit "fire this off and let interested parties react in their own time" work: notifications, audit logging, analytics, cache invalidation, kicking off a downstream workflow. Marking a task complete and notifying someone is squarely in the second bucket — Orders doesn't need Notifications' answer to consider its own job done, which is exactly why Week 9's direct call was more coupling than the use case actually required.

Put side by side, the shift is: a synchronous call blocks the caller until it gets a response or a timeout, and ties the caller's own success to the callee being reachable right now — Week 9's circuit breaker only softens that second part, it doesn't remove it. An async event never blocks the publisher, lets the broker buffer the message until a consumer is ready, and can be picked up by zero, one, or many consumers the publisher never has to know about. Use a synchronous call when the caller needs an immediate answer to keep going; use an event when it doesn't.

2. Event-Driven Messaging with Kafka

Apache Kafka models messaging as a distributed, append-only log rather than a queue. Producers write records to a topic; a topic is split into partitions for throughput and ordering (records with the same key always land on the same partition, so per-key ordering is guaranteed). Consumers read from a topic as part of a consumer group — Kafka spreads a topic's partitions across the group's members, so adding consumers scales throughput, and each partition is only ever read by one consumer within a given group at a time.

Crucially, consuming a record doesn't delete it. Kafka retains records for a configured period regardless of whether anyone's read them, and each consumer group tracks its own read position (its offset) independently. That's what lets a brand-new consumer group — analytics, say — start reading the same task-events topic from the beginning without touching what Notifications has already consumed.

Producing with KafkaTemplate

package com.codeverse.week10.event;

import java.time.Instant;

public record TaskCompleted(
    String taskId,
    String assigneeEmail,
    String taskTitle,
    Instant completedAt
) {}

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.kafka.consumer.group-id=notification-service
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=com.codeverse.week10.event

package com.codeverse.week10.task;

import com.codeverse.week10.event.TaskCompleted;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class TaskEventPublisher {

    private static final String TOPIC = "task-events";

    private final KafkaTemplate<String, TaskCompleted> kafkaTemplate;

    public TaskEventPublisher(KafkaTemplate<String, TaskCompleted> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishTaskCompleted(TaskCompleted event) {
        // Key by taskId so all events for one task stay ordered on one partition.
        kafkaTemplate.send(TOPIC, event.taskId(), event);
    }
}

Consuming with @KafkaListener

package com.codeverse.week10.notification;

import com.codeverse.week10.event.TaskCompleted;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class TaskCompletedListener {

    private final NotificationSender notificationSender;

    public TaskCompletedListener(NotificationSender notificationSender) {
        this.notificationSender = notificationSender;
    }

    @KafkaListener(topics = "task-events", groupId = "notification-service")
    public void onTaskCompleted(TaskCompleted event) {
        // Sending twice for the same taskId must be harmless -- see "at-least-once" below.
        notificationSender.sendCompletionEmail(event.assigneeEmail(), event.taskTitle());
    }
}

At-least-once delivery

Kafka's default delivery guarantee is at-least-once: if a consumer crashes after processing a record but before committing its offset, that record is redelivered on restart. That's the safe failure mode — you never silently lose an event — but it means onTaskCompleted above can run twice for the same TaskCompleted. Consumers must be idempotent: sending the same completion email twice should be harmless (or de-duplicated by taskId), because "exactly once" is not the default and shouldn't be assumed without deliberately engineering for it.

3. Messaging with RabbitMQ

RabbitMQ models messaging as a message broker with queues, not a log. A producer publishes a message to an exchange; a binding routes it from that exchange into one or more queues based on rules (a routing key, for the common direct exchange type); a consumer reads from a queue, and once a message is acknowledged, it's removed. There's no replay-from-the-beginning the way a Kafka topic offers — once every bound queue has consumed and acked a message, it's gone.

Kafka vs RabbitMQ: log vs queue

The mental model difference matters more than the API difference. Kafka retains history and lets multiple independent consumer groups replay the same stream at their own pace — a natural fit for event streaming, analytics, and audit trails where you might add a new consumer of old data later. RabbitMQ is built around message routing and consumption: once processed, a message is gone, and it offers richer routing (topic exchanges, fanout, dead-letter queues, per-message priority) that's a natural fit for task distribution and work queues where you want a message handled once and done. For a TaskCompleted notification, either genuinely works; teams often pick based on what's already running in their infrastructure rather than a hard technical requirement.

Producing with RabbitTemplate

package com.codeverse.week10.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    public static final String EXCHANGE = "task.events.exchange";
    public static final String QUEUE = "notification.task-completed.queue";
    public static final String ROUTING_KEY = "task.completed";

    @Bean
    DirectExchange taskEventsExchange() {
        return new DirectExchange(EXCHANGE);
    }

    @Bean
    Queue taskCompletedQueue() {
        return new Queue(QUEUE, true);
    }

    @Bean
    Binding taskCompletedBinding(Queue taskCompletedQueue, DirectExchange taskEventsExchange) {
        return BindingBuilder.bind(taskCompletedQueue)
            .to(taskEventsExchange)
            .with(ROUTING_KEY);
    }
}

package com.codeverse.week10.task;

import com.codeverse.week10.config.RabbitConfig;
import com.codeverse.week10.event.TaskCompleted;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

@Component
public class RabbitTaskEventPublisher {

    private final RabbitTemplate rabbitTemplate;

    public RabbitTaskEventPublisher(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void publishTaskCompleted(TaskCompleted event) {
        rabbitTemplate.convertAndSend(
            RabbitConfig.EXCHANGE, RabbitConfig.ROUTING_KEY, event);
    }
}

Consuming with @RabbitListener

package com.codeverse.week10.notification;

import com.codeverse.week10.config.RabbitConfig;
import com.codeverse.week10.event.TaskCompleted;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class RabbitTaskCompletedListener {

    private final NotificationSender notificationSender;

    public RabbitTaskCompletedListener(NotificationSender notificationSender) {
        this.notificationSender = notificationSender;
    }

    @RabbitListener(queues = RabbitConfig.QUEUE)
    public void onTaskCompleted(TaskCompleted event) {
        notificationSender.sendCompletionEmail(event.assigneeEmail(), event.taskTitle());
    }
}

4. Eventual Consistency & the Outbox Pattern

Here's the problem publishing introduces. Marking a task complete involves two separate systems: your database (update the task's status) and your broker (publish TaskCompleted). Naively, you'd do this:

@Transactional
public void completeTask(String taskId) {
    Task task = taskRepository.findById(taskId).orElseThrow();
    task.markComplete();
    taskRepository.save(task);

    // What if the process crashes right here, after commit but before publish?
    // What if this publish call throws, after the DB transaction already committed?
    kafkaTemplate.send("task-events", taskId, new TaskCompleted(/* ... */));
}

The database write and the broker publish are two independent operations against two independent systems, and there's no single transaction spanning both. If the process crashes between them, or the broker is briefly unreachable, the database says the task is complete but the event was never published — Notifications never hears about it. Publish first and save second, and you can end up announcing an event for a change that then fails to commit. Either ordering has a window where the two systems disagree.

The transactional outbox pattern

The fix is to never let the event publish be a separate operation in the first place. Instead, you write the event to an outbox table in the same database, inside the same @Transactional boundary you already know from Week 5, as the business change it describes. A database transaction is atomic by definition, so the task update and the outbox row either both commit or neither does — there's no window where one happened without the other.

package com.codeverse.week10.outbox;

import jakarta.persistence.*;
import java.time.Instant;

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private String id;

    private String aggregateType;   // "Task"
    private String aggregateId;     // the taskId
    private String eventType;       // "TaskCompleted"

    @Lob
    private String payload;         // JSON-serialized TaskCompleted

    private Instant createdAt = Instant.now();
    private boolean published = false;

    // getters and setters omitted
}

@Transactional
public void completeTask(String taskId) {
    Task task = taskRepository.findById(taskId).orElseThrow();
    task.markComplete();
    taskRepository.save(task);

    TaskCompleted event = new TaskCompleted(
        taskId, task.getAssigneeEmail(), task.getTitle(), Instant.now());

    OutboxEvent outboxEvent = new OutboxEvent();
    outboxEvent.setAggregateType("Task");
    outboxEvent.setAggregateId(taskId);
    outboxEvent.setEventType("TaskCompleted");
    outboxEvent.setPayload(toJson(event));
    outboxRepository.save(outboxEvent);
    // Both the Task update and this outbox row commit -- or roll back -- together.
}

A separate poller (or relay) — a scheduled job running in the same service — then reads unpublished outbox rows, publishes each one to Kafka, and marks it published. If the publish step fails partway through, the row is simply left unpublished and retried on the next poll; the worst case is a duplicate publish (handled by the idempotent consumers from Section 2), never a lost one.

package com.codeverse.week10.outbox;

import com.codeverse.week10.event.TaskCompleted;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

@Component
public class OutboxPoller {

    private final OutboxRepository outboxRepository;
    private final KafkaTemplate<String, TaskCompleted> kafkaTemplate;
    private final ObjectMapperWrapper objectMapper;

    public OutboxPoller(OutboxRepository outboxRepository,
                         KafkaTemplate<String, TaskCompleted> kafkaTemplate,
                         ObjectMapperWrapper objectMapper) {
        this.outboxRepository = outboxRepository;
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = objectMapper;
    }

    @Scheduled(fixedDelay = 2000)
    @Transactional
    public void relayUnpublishedEvents() {
        List<OutboxEvent> pending = outboxRepository.findTop50ByPublishedFalseOrderByCreatedAtAsc();

        for (OutboxEvent row : pending) {
            TaskCompleted event = objectMapper.fromJson(row.getPayload(), TaskCompleted.class);
            kafkaTemplate.send("task-events", row.getAggregateId(), event);
            row.setPublished(true);
            outboxRepository.save(row);
        }
    }
}

This trades instant publishing for a small, bounded delay (the poll interval) in exchange for a guarantee that matters far more: the database and the event stream can never permanently disagree about whether TaskCompleted happened.

5. Hands-on Exercise

6. Knowledge Check

Four quick questions. Expand each to check your answer.