Migrate domain event bus from RabbitMQ to Kafka

[maalayat]

Spec: Migrate Domain Event Bus from RabbitMQ to Kafka

Problem Statement

The current domain event infrastructure uses RabbitMQ with a reflection-based subscriber discovery and serialization layer that is overengineered for the project's needs. The org.reflections library scans the classpath at startup to find subscribers annotated with @DomainEventSubscriber, and the DomainEventJsonDeserializer creates null no-arg instances of each event class to invoke fromPrimitives() via reflection. Every domain event subclass requires a meaningless no-arg constructor, three serialization methods (toPrimitives(), toJson(), fromPrimitives()), plus equals/hashCode — roughly 40 lines of boilerplate per event.

The RabbitMQ topology itself is complex: 3 topic exchanges (main, retry, dead_letter), 3 queues per subscriber, retry queues with 1s TTL, and manual redelivery counting in message headers. A Guava dependency exists solely for CaseFormat to convert subscriber class names to queue names.

This infrastructure makes adding new domain events more expensive than it should be, couples the domain layer to serialization concerns, and requires developers to understand the reflection pipeline to debug event flow.

Solution

Replace RabbitMQ with Kafka using per-type topics and explicit infrastructure-level consumers. Keep the domain layer pure — no Jackson annotations, no framework dependencies. Serialization and deserialization live entirely in the infrastructure layer using Jackson mix-ins, not reflection.

The EventBus domain interface stays unchanged. Application services and domain aggregates are untouched. Only the infrastructure layer's messaging implementation changes.

User Stories

1. As a developer, I want domain events to be published to Kafka topics, so that the system reliably distributes events to subscribers.
2. As a developer, I want each domain event type to have its own Kafka topic, so that consumers only receive events they care about and topic naming matches domain language.
3. As a developer, I want subscribers to use explicit @KafkaListener annotations on infrastructure-layer classes, so that the event handling pipeline is transparent, debuggable, and doesn't require reflection.
4. As a developer, I want the DomainEvent base class to be free of infrastructure annotations and serialization methods, so that the domain layer remains pure DDD.
5. As a developer, I want serialization and deserialization logic to live in the infrastructure layer, so that domain classes don't depend on Jackson, Serializable, or any framework.
6. As a developer, I want the no-arg constructors removed from domain event classes, so that event objects are always fully initialized with valid state.
7. As a developer, I want to remove the Guava dependency from the project, so that the project has fewer transitive library constraints.
8. As a developer, I want to remove the entire RabbitMQ infrastructure package and its test dependencies, so that the codebase doesn't maintain two messaging implementations with different topologies.
9. As a developer, I want events that fail Kafka delivery after retries to be persisted to MySQL, so that no events are silently lost and operators have a dead-letter store for manual inspection.
10. As a developer, I want the EventBus domain interface to remain unchanged, so that existing application services don't need modification.
11. As a developer, I want new domain events to require only a new topic declaration and a new infrastructure consumer class, so that adding an event is a mechanical, low-ceremony process.
12. As the system, I want Kafka to reliably deliver each event to at least one consumer, so that business invariants are eventually consistent.
13. As a developer, I want the Spring in-process event bus retained as an optional escape hatch, so that synchronous in-process event handling remains available for future use cases without adding infrastructure dependencies.
14. As a developer, I want infrastructure consumer classes to inject application services directly via constructor injection, so that event flow is type-safe and does not use reflection or string-based bean lookups.

Implementation Decisions

Module Architecture

The implementation is split across two Gradle modules — shared (infrastructure building blocks) and mooc (bounded context with the first consumer).

New modules (shared):

• KafkaEventBus — implements EventBus. Receives List<DomainEvent>, serializes each event to the JSON envelope via DomainEventKafkaSerializer, publishes to the Kafka topic matching the event's eventName(). On KafkaException after producer retries are exhausted, falls back to MySqlEventBus for dead-letter storage.

• DomainEventKafkaSerializer — converts a DomainEvent to a byte array in the JSON envelope format: {"data":{"id":"...","type":"course.created","occurred_on":"...","attributes":{...}},"meta":{}}. Uses Jackson ObjectMapper with an infrastructure-level mix-in. No reflection, no domain layer dependencies.

• DomainEventKafkaDeserializer — converts a byte array back to a concrete DomainEvent subclass. Unwraps the JSON envelope, resolves the concrete class via the type field using a Jackson mix-in, deserializes directly without reflection or fromPrimitives().

• KafkaEventBusConfiguration — Spring @Configuration that declares: Kafka ProducerFactory/KafkaTemplate, consumer ConsumerFactory with the custom deserializer, NewTopic beans for each event type (one topic per event type, replicated and partitioned appropriately), and the Jackson ObjectMapper bean with the mix-in registered.

• DomainEventMixIn — infrastructure-level Jackson mix-in abstract class (not on DomainEvent itself) annotated with @JsonTypeInfo(use = Id.NAME, property = "type") and @JsonSubTypes listing known domain events. This file is the single registry of event-type-to-class mappings and must be updated when new domain events are added.

New modules (mooc):

• CourseCreatedKafkaConsumer — @Service + @KafkaListener(topics = "course.created"). Receives CourseCreatedDomainEvent, calls IncrementCoursesCounterOnCourseCreated.on(). Serves as the pattern for all future domain event consumers.

Modified modules (shared):

• DomainEvent — removes four abstract methods: toPrimitives(), toJson(), fromPrimitives(). Removes implements Serializable. Retains: aggregateId, eventId, occurredOn, eventName().

Modified modules (mooc):

• CourseCreatedDomainEvent — removes no-arg constructor, toPrimitives(), toJson(), fromPrimitives(), equals(), hashCode(). Becomes a simple final class with one constructor and accessors only.
• IncrementCoursesCounterOnCourseCreated — removes @DomainEventSubscriber and @EventListener annotations. Retains @Service and the on() method.

Deleted modules (shared):

• All 7 RabbitMQ infrastructure files
• DomainEventJsonSerializer, DomainEventJsonDeserializer
• DomainEventsInformation, DomainEventSubscribersInformation, DomainEventSubscriberInformation
• DomainEventSubscriber annotation
• NamingUtil

Retained unchanged:

• EventBus interface — no changes to its void publish(List<DomainEvent>) contract
• AggregateRoot — no changes to record() or pullDomainEvents()
• MySqlEventBus — kept as dead-letter store, invoked by KafkaEventBus on persistent Kafka failure
• SpringApplicationEventBus — kept for optional in-process event handling
• All application services (CourseCreator, CoursesCounterIncrementer, etc.) — no changes
• CommandHandlersInformation, QueryHandlersInformation — still use org.reflections

Build Dependencies

• Add: org.springframework.kafka:spring-kafka
• Remove: org.springframework.boot:spring-boot-starter-amqp, com.google.guava:guava:33.0.0-jre
• Keep: org.reflections:reflections:0.10.2 (still used by CQRS infrastructure)
• Remove test deps: spring-rabbit-test, org.testcontainers:rabbitmq
• Add test deps: org.testcontainers:kafka (for integration tests)

Topic Strategy

• One Kafka topic per domain event type
• Topic name equals the event name returned by eventName() (e.g., "course.created")
• Topics are created programmatically via NewTopic beans in KafkaEventBusConfiguration
• Replication factor and partition count are configurable per topic

Envelope Format

{
  "data": {
    "id": "uuid-of-event",
    "type": "course.created",
    "occurred_on": "2026-05-27T12:00:00",
    "attributes": {
      "id": "aggregate-id",
      ...event-specific fields...
    }
  },
  "meta": {}
}

The type field enables future consumers that subscribe to multiple topics or replay historical events. The envelope is constructed and consumed entirely in the infrastructure layer — domain classes never see it.

Testing Decisions

The Primary Seam

The highest existing seam is the EventBus interface — void publish(List<DomainEvent>). This single seam spans the entire producer-to-consumer pipeline without exposing Kafka, Jackson, RabbitMQ, or serialization details. We will test the migration by driving behavior through this seam and observing side effects at the other end.

Producer-side seam test

Unit-test KafkaEventBus through the EventBus seam:

• Inject a mock KafkaTemplate and a stub MySqlEventBus
• Call eventBus.publish(List.of(courseCreatedEvent))
• Verify KafkaTemplate.send("course.created", ...) was called once with a serialized JSON envelope containing the correct type, aggregateId, and event attributes
• Force KafkaTemplate.send() to throw KafkaException and verify MySqlEventBus receives the event for dead-letter storage

This test does not assert on the exact byte contents of the envelope string. It asserts on the external contract: correct topic and a deserializable envelope.

Consumer-side seam test

Unit-test CourseCreatedKafkaConsumer through its own seam — the on(CourseCreatedDomainEvent) method of the application service it delegates to:

• Construct a CourseCreatedDomainEvent directly (no broker, no JSON)
• Call the consumer with the event
• Verify the application service's on() method is called exactly once with the same event

This isolates the wiring between infrastructure and application layers.

End-to-end seam test

One integration test drives the full flow through the EventBus seam using an embedded Kafka broker:

• Invoke CourseCreator.create(...) — this calls eventBus.publish(pullDomainEvents())
• Assert that the CoursesCounter read model is eventually incremented, or that IncrementCoursesCounterOnCourseCreated.on() was invoked

This single test proves serialization, topic routing, Kafka delivery, deserialization, and consumer invocation work together without exposing any of those layers in the assertion.

What makes a good test

Tests should verify external behavior at a seam, not implementation details. Do not assert on:

• Exact JSON string formatting
• Whether a specific private method was called
• Reflection-based discovery state

Do assert on:

• Events arrive at the expected Kafka topic
• The envelope round-trips correctly
• Dead-letter fallback fires on persistent failure
• The downstream business effect occurs when an event is consumed

Prior Art

• MysqlEventBusTest in src/mooc/test/ shows the existing integration test pattern (Spring Boot test, autowire the bus, publish and verify)
• CoursesCounterModuleUnitTestCase and CoursesModuleUnitTestCase show the unit test base class pattern with mocked collaborators
• Object mothers (CourseCreatedDomainEventMother) should be retained for test data generation

Out of Scope

• Migration of the CQRS command and query buses to Kafka or removal of org.reflections — those remain unchanged and are outside this spec
• Removal of SpringApplicationEventBus — retained for future in-process event use cases
• Adding Kafka to the Docker Compose or local development setup — the developer must provide a Kafka broker; this spec covers code changes only
• Schema registry or Avro/Protobuf serialization — Jackson JSON is sufficient for the current event volume and complexity
• Migration of existing RabbitMQ messages from the broker — this is a clean cutover; no message replay from RabbitMQ queues is planned
• Retrofitting tests for existing code uncovered by this migration

Further Notes

• This is a pure infrastructure swap. No domain logic, application service, aggregate, or value object changes are required beyond removing serialization boilerplate from event classes.
• The KafkaEventBus is @Primary (replacing RabbitMqEventBus in that role) so all existing EventBus injection points resolve to it without code changes.
• New domain events follow this pattern: (1) create the domain event class with fields only, (2) add a @JsonSubTypes.Type entry in the mix-in class, (3) add a NewTopic bean in KafkaEventBusConfiguration, (4) create an infrastructure consumer with @KafkaListener.