Spring Builders: Ivan Garcia Sainz-Aja

Externalize Spring-Modulith Events with Spring Cloud Stream

Ivan Garcia Sainz-Aja — Sat, 08 Mar 2025 14:47:07 +0000

Spring Modulith enables developers to build well-structured modular monoliths with built-in event capabilities. This allows teams to leverage Event-Driven Architecture patterns without immediately committing to a distributed system.

It provides multiple event externalizers out-of-the-box:

Kafka: spring-modulith-events-kafka
AMQP: spring-modulith-events-amqp
JMS: spring-modulith-events-jms
AWS SQS: spring-modulith-events-aws-sqs
AWS SNS: spring-modulith-events-aws-sns
Spring Messaging: spring-modulith-events-messaging

While these built-in externalizers cover common use cases, there are scenarios where you need more flexibility and control.

I'm happy to introduce a new library featuring a new Spring Modulith event externalizer for Spring Cloud Stream.

Why a New Library?

This library addresses several key needs by:

Leverage Spring Cloud Stream to support multiple message brokers at once, even inside the same application.
Providing enhanced control over message headers and metadata.
Supporting flexible payload serialization with both JSON and Avro.

GitHub Repository

Getting Started

Using this library is straightforward. Here's what you need to do:

Add the Spring-Modulith Events Externalizer dependency to your project
Include your preferred Spring Cloud Stream binder (Kafka, RabbitMQ, etc.)
Configure Spring Cloud Stream bindings in application.yml
Enable externalization with @EnableSpringCloudStreamEventExternalization
Use ApplicationEventPublisher as normal, to publish POJOs, Avro or Message<?> events

1. Add Core Dependency

<dependency>
    <groupId>io.zenwave360.sdk</groupId>
    <artifactId>spring-modulith-events-scs</artifactId>
    <version>1.0.0-RC1</version>
</dependency>

2. Add Spring Cloud Stream Message Broker Binder

Choose your preferred message broker. For Kafka:

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

3. Configure Bindings

Configure your output bindings in application.yml. We are going to configure two output bindings for different payload types:

spring:
  cloud:
    stream:
      bindings:
        # JSON events binding
        customers-json-out-0:
          destination: customers-json-topic
        # Avro events binding
        customers-avro-out-0:
          destination: customers-avro-topic
          content-type: application/*+avro
  # Kafka-specific configuration
  kafka:
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer

A key advantage of using Spring Cloud Stream is the ability to configure multiple message brokers simultaneously:

Use different brokers (Kafka, RabbitMQ, etc.) in the same application
Route different events to different brokers through configuration

Basic Configuration

Enable Spring Cloud Stream externalization by adding the @EnableSpringCloudStreamEventExternalization annotation:

@Configuration
@EnableSpringCloudStreamEventExternalization
public class EventsConfiguration { }

Sending Events

POJO Events

Use ApplicationEventPublisher as you normally would in Spring Modulith:

@Externalized("customers-json-out-0::#{#this.getId()}")  // binding name and routing key
public class CustomerEvent {
    // Your POJO implementation
}

@Service
@Transactional
public class CustomerEventsProducer {
    private final ApplicationEventPublisher publisher;

    public CustomerEventsProducer(ApplicationEventPublisher publisher) {
        this.publisher = publisher;
    }

    public void publishCustomerEvent(CustomerEvent event) {
        publisher.publishEvent(event); // <-- Sending the event
    }
}

Avro Events

Add the Avro dependency:

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-avro</artifactId>
</dependency>

Define your Avro event:

@Externalized("customers-avro-out-0::#{#this.getId()}")  // binding name and routing key
public class CustomerEvent extends SpecificRecordBase implements SpecificRecord {
    // Your Avro implementation
}

Publish events the same way as POJOs:

@Service
@Transactional
public class CustomerEventsProducer {
    private final ApplicationEventPublisher publisher;

    public void publishAvroEvent(CustomerEvent event) {
        publisher.publishEvent(event); // <-- Sending the event
    }
}

Routing Key Header

The SpringCloudStreamEventExternalizer automatically maps routing keys to the appropriate message header based on your message broker:

Message Broker	Header Name
Kafka	`kafka_messageKey`
RabbitMQ	`rabbit_routingKey`
Kinesis	`partitionKey`
Google PubSub	`pubsub_orderingKey`
Azure Event Hubs	`partitionKey`
Solace	`solace_messageKey`
Apache Pulsar	`pulsar_key`

Event Serialization for Spring Modulith Publication Log

Spring Modulith's Transactional Event Publication Log requires events to be serialized for database storage. This presents two challenges:

Type Information: The default JacksonEventSerializer loses generic type information for Message<?> payloads
Format Support: If you need to support Avro payloads, JacksonEventSerializer does not play well with Avro GenericRecord/SpecificRecord

This library addresses these challenges by:

Adding a _class field to preserve complete type information for Message<?> payloads
Supporting both POJO (JSON) and Avro serialization formats, through AvroMapper
Enabling full deserialization back to original types

Avro serialization requires the com.fasterxml.jackson.dataformat.avro.AvroMapper class to be present in the classpath. In order to use Avro serialization, you need to add com.fasterxml.jackson.dataformat:jackson-dataformat-avro dependency to your project, as stated above

Sending Spring Message Events

For advanced control over message headers, you can send Message<?> objects by including the spring.cloud.stream.sendto.destination routing header. This header should point to your intended Spring Cloud Stream output binding.

@Service
public class CustomerEventsProducer {
    private final ApplicationEventPublisher applicationEventPublisher;

    public CustomerEventsProducer(ApplicationEventPublisher applicationEventPublisher) {
        this.applicationEventPublisher = applicationEventPublisher;
    }

    @Transactional
    public void sendCustomerEvent(CustomerEvent event) {
        Message<CustomerEvent> message = MessageBuilder
            .withPayload(event)                     // supports both POJO and Avro payloads
            .setHeader(
                SpringCloudStreamEventExternalizer.SPRING_CLOUD_STREAM_SENDTO_DESTINATION_HEADER,
                "customers-json-out-0"              // target binding name
            )
            .build();
        applicationEventPublisher.publishEvent(message);
    }
}

This header is automatically set when using ZenWave SDK AsyncAPI Generator.

Event Externalization and API Management with AsyncAPI

While Spring Modulith's @Externalized annotation provides a quick and convenient way to publish events, teams building event-driven systems often need additional capabilities:

API Documentation: No built-in support for formal API documentation
Schema Management: No friction to prevent breaking changes in event schemas that could impact consumers
API Governance: No standardized way to enforce API design standards: naming conventions, versioning, headers/metadata...

API-First Approach with AsyncAPI

For teams following API-First practices, AsyncAPI offers a better approach to describe your Event-Driven Architecture:

Formal API documentation
Schema validation (Avro, JSON Schema)
API governance and versioning
Contract-first development

Code Generation with ZenWave SDK

The ZenWave SDK AsyncAPI Generator can generate full SDKs from AsyncAPI definitions, including:

Event Models/DTOs with full type safety
Strongly-typed header objects with runtime population support
Spring Cloud Stream event producers/consumers with transactional support via Spring Modulith
Zero boilerplate code

NOTE: Already using @Externalized? We're developing a tool to reverse engineer your events into AsyncAPI specifications.

Example Implementation

See it in action with this complete example:

Playground Project: Full implementation with AsyncAPI + Avro
Implementation Guide: Step-by-step tutorial

Benefits Over Built-in Externalization

Broker Flexibility: Connect to any message broker supported by Spring Cloud Stream
Enhanced Header Control: Simple configuration of message headers
Multiple Serialization Formats: Built-in support for JSON and Avro
AsyncAPI Integration: Seamless integration with Spring Modulith Events and Spring Cloud Stream through ZenWave SDK AsyncAPI Generator

Get Involved

Visit GitHub repository

We welcome contributions and feedback from the community! 🚀

Originally published at: https://www.zenwave360.io/posts/Spring-Modulith-Events-Spring-Cloud-Stream-Externalizer/

Implementing a Transactional OutBox With AsyncAPI, SpringModulith and ZenWaveSDK (No Code Required)

Ivan Garcia Sainz-Aja — Sat, 08 Mar 2025 14:38:37 +0000

Introduction

When processes span multiple services without shared transaction boundaries, ensuring atomicity and consistency in distributed systems is challenging.

Distributed transactions with Two-Phase Commits (2PC) are complex, introduce significant performance overhead, and may not even be feasible for certain services.

The Outbox Pattern solves this by using a Database Transaction to store events in a dedicated "outbox" table within the same transaction as a database update. These events are then published to external systems, such as an email service or message broker, ensuring eventual consistency without the need for distributed transactions.

In this post, we’ll explore how we can implement a Transactional Outbox Pattern to:

Persist data to a supported transactional database (e.g., SQL or MongoDB).
Send events to an external message broker like Kafka or RabbitMQ using Spring Cloud Stream.
Leverage Spring Modulith Events transactional features.
Use ZenWaveSDK Code Generator for AsyncAPI so you don’t need to write a single line of boilerplate code for the transactional outbox and event publishing.

Playground Project

Because working software is worth more than a thousand words, we’ll use a fully functional playground project that you can explore and test yourself.

We'll use the following project as our playground: EDA-TransactionalOutbox-Modulith-JPA.

This project provides a simple API for managing customer details, including REST endpoints for CRUD operations. These operations publish event notifications to a Kafka topic, using Avro as the payload format.

Key components of the project include:

OpenAPI definition file: openapi.yml
AsyncAPI definition file: asyncapi.yml
Avro schemas for events: Avro schemas

Distributed Transactions Problem and The Outbox Pattern

When managing a Customer entity, we encounter a significant challenge: ensuring the atomicity of persisting customer details in the database and publishing a related event to an external message broker. Since the database and the message broker do not share a transaction boundary, this can lead to potential inconsistencies.

Key Scenarios Highlighting the Problem:

Event Sent Before Database Transaction Commits:
If the event is published before the database transaction is committed, a rollback of the transaction would leave the system in an inconsistent state, as the published event cannot be undone.
Event Sent After Database Transaction Commits:
If the event is published after the database transaction is committed, there’s no guarantee the event will actually be sent. A service crash or network failure could prevent the event from being published, resulting in data inconsistency.

@Service
public class CustomerService {
    // ...

    @Transactional
    public Customer createCustomer(Customer input) {
        log.debug("Request to save Customer: {}", input);
        var customer = mapper.update(new Customer(), input);
        customer = customerRepository.save(customer); // Persist to DB

        var customerEvent = eventsMapper.asCustomerEvent(customer);
        eventsProducer.onCustomerEvent(customerEvent); // Emit Event to external Broker
        return customer;
    }
}

With the transactional outbox pattern, instead of directly sending the event to the external system, the call to eventsProducer.onCustomerEvent(customerEvent) stores the event in a dedicated outbox table as part of the same transaction as the database update. An external process then reads from the outbox table and publishes the event to the message broker.

However, implementing this external process, managing the outbox table, and ensuring that events are published once and in the correct order can be complex and error-prone.

Fortunately, tools like Spring Modulith’s Event Publication Registry and the ZenWaveSDK Code Generator for AsyncAPI simplify this process, so you can focus only on implementing your business logic.

ZenWaveSDK Code Generation for AsyncAPI and Spring Cloud Streams

ZenWaveSDK generates all the boilerplate code needed to send and receive events using Spring Cloud Stream from an AsyncAPI definition file.

From your AsyncAPI definition file, ZenWaveSDK generates:

DTO/Models for your event payloads.
Typed Header Objects to ensure consistency.
"Intention-revealing" Java interfaces with methods named explicitly after your operationIds.
A lightweight implementation using Spring Cloud Streams that can be @Autowired directly into your services.

You can then configure Spring Cloud Stream to send and receive messages using any of the supported binders, such as Kafka or RabbitMQ. With ZenWaveSDK, you don’t need to write any boilerplate code, allowing you to focus entirely on your business logic.

ZenWave SDK Maven Plugin Configuration

  <plugin>
    <groupId>io.github.zenwave360.zenwave-sdk</groupId>
    <artifactId>zenwave-sdk-maven-plugin</artifactId>
    <version>${zenwave.version}</version>
    <configuration>
      <inputSpec>${project.basedir}/src/main/resources/public/apis/asyncapi.yml</inputSpec>
      <addCompileSourceRoot>true</addCompileSourceRoot>
      <addTestCompileSourceRoot>true</addTestCompileSourceRoot>
    </configuration>
    <executions>
      <!-- DTOs -->
      <!-- we skip DTOs generation in this case b/c we are using Avro for that -->

      <!-- Generate PROVIDER -->
      <execution>
        <id>generate-asyncapi</id>
        <phase>generate-sources</phase>
        <goals>
          <goal>generate</goal>
        </goals>
        <configuration>
          <generatorName>spring-cloud-streams3</generatorName>
          <configOptions>
            <role>provider</role>
            <style>imperative</style>
            <transactionalOutbox>modulith</transactionalOutbox><!-- using Spring Modulith implementation -->
            <modelPackage>${asyncApiModelPackage}</modelPackage>
            <producerApiPackage>${asyncApiProducerApiPackage}</producerApiPackage>
            <consumerApiPackage>${asyncApiConsumerApiPackage}</consumerApiPackage>
          </configOptions>
        </configuration>
      </execution>
    </executions>
    <dependencies>
      // ...
    </dependencies>
  </plugin>

*Avro Maven Plugin Configuration

<plugin>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-maven-plugin</artifactId>
  <version>1.11.1</version>
  <executions>
    <execution>
      <goals>
        <goal>schema</goal>
      </goals>
      <phase>generate-sources</phase>
    </execution>
  </executions>
  <configuration>
    <sourceDirectory>${project.basedir}/src/main/resources/public/apis/avro</sourceDirectory>
    <outputDirectory>${project.basedir}/target/generated-sources/avro</outputDirectory>
    <imports>
      <import>${project.basedir}/src/main/resources/public/apis/avro/PaymentMethodType.avsc</import>
      <import>${project.basedir}/src/main/resources/public/apis/avro/PaymentMethod.avsc</import>
      <import>${project.basedir}/src/main/resources/public/apis/avro/Address.avsc</import>
    </imports>
  </configuration>
</plugin>

To achieve this, we configure the zenwave-sdk-maven-plugin and, in this case, also the avro-maven-plugin in the pom.xml file. This setup ensures that the necessary code is generated in the target/generated-sources folder as part of your build process.

Since this configuration runs automatically during the build, every time you update your AsyncAPI definition file, you can be sure that your code remains in sync with your API definition.

And because we configured the transactionalOutbox option to modulith, ZenWaveSDK will generate the code to use Spring Modulith Events Publication Registry to manage the transactional outbox for you.

Spring Modulith Events Publication Registry

Because Event Externalization is enabled, and Message<?> objects are configured for externalization (see below), these events are also stored transactionally in the Spring Modulith Event Publication Registry.

spring:
  modulith.events.externalization.enabled: true
  modulith.events.jdbc.schema-initialization.enabled: true
  modulith.events.republish-outstanding-events-on-restart: true

Using Spring Cloud Streams to Externalize Modulith Events

Now that Spring Modulith is managing our Message<?> events, we need to configure one of the many supported event externalizers.

If we were publishing POJOs in JSON format, we could use spring-modulith-events-kafka to externalize events to a Kafka topic. However, since we want to externalize Message<?> objects with Avro payloads, we will use io.zenwave360.sdk:spring-modulith-events-scs, which supports serializing and deserializing Message<?> objects, with or without Avro payloads.

For more details, see ZenWave360 Spring Modulith Events for Spring Cloud Stream.

To enable this, add the following dependencies to your pom.xml file:

<dependency><!-- Spring Cloud Stream Externalization for Message<?> -->
    <groupId>io.zenwave360.sdk</groupId>
    <artifactId>spring-modulith-events-scs</artifactId>
    <version>${spring-modulith-events-scs.version}</version>
</dependency>
<dependency><!-- Needed for serializing Avro payloads to db storage as json -->
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-avro</artifactId>
</dependency>

And @EnableSpringCloudStreamEventExternalization to our Spring Boot Configuration:

import io.zenwave360.modulith.events.scs.config.EnableSpringCloudStreamEventExternalization;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableSpringCloudStreamEventExternalization
class ExternalizationConfiguration {
}

This will route programmatically route our Message<?> objects to the correct Spring Cloud Stream binding, but you don't need to worry about any of this as it is handled automatically behind the scenes.

// this is some of what you get when adding @EnableSpringCloudStreamEventExternalization
@AutoConfiguration
@AutoConfigureAfter(EventExternalizationAutoConfiguration.class)
@ConditionalOnProperty(name = "spring.modulith.events.externalization.enabled", havingValue = "true",
        matchIfMissing = true)
public class MessageExternalizationConfiguration {

    @Bean
    EventExternalizationConfiguration eventExternalizationConfiguration() {
        return EventExternalizationConfiguration.externalizing()
            .select(event -> annotatedAsExternalized().test(event)
                    || event instanceof Message<?> && getTarget(event) != null)
            .route(Message.class, event -> RoutingTarget.forTarget(getTarget(event)).withoutKey())
            .build();
    }

    private String getTarget(Object event) {
        if (event instanceof Message<?> message) {
            return message.getHeaders()
                .get(SpringCloudStreamEventExternalizer.SPRING_CLOUD_STREAM_SENDTO_DESTINATION_HEADER, String.class);
        }
        return null;
    }
}

Automagic Transactional OutBox Implementation

Our original code is now automagically implementing the Transactional OutBox pattern using Spring Modulith Events Publication Registry and Spring Cloud Stream, all made possible by the ZenWaveSDK AsyncAPI Code Generator. 🚀🚀🚀

@Service
public class CustomerService {
    // ...

    @Transactional
    public Customer createCustomer(Customer input) {
        log.debug("Request to save Customer: {}", input);
        var customer = mapper.update(new Customer(), input);
        customer = customerRepository.save(customer); // Persist to DB

        var customerEvent = eventsMapper.asCustomerEvent(customer);
        eventsProducer.onCustomerEvent(customerEvent); // Emit Event to external Broker via Tx OutBox
        return customer;
    }
}

Originally published at: https://www.zenwave360.io/posts/TransactionalOutBoxWithAsyncAPIAndSpringModulith/