GH-1370 - Support for serialized event externalization.

Author: odrotbohm

spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/DefaultEventExternalizationConfiguration.java

@@ -33,6 +33,7 @@ class DefaultEventExternalizationConfiguration implements EventExternalizationCo
 	private final Function<Object, Object> mapper;
 	private final Function<Object, RoutingTarget> router;
 	private final Function<Object, Map<String, Object>> headers;
+	private final boolean serializeExternalization;
 
 	/**
 	 * Creates a new {@link DefaultEventExternalizationConfiguration}
@@ -43,7 +44,8 @@ class DefaultEventExternalizationConfiguration implements EventExternalizationCo
 	 * @param headers must not be {@literal null}.
 	 */
 	DefaultEventExternalizationConfiguration(Predicate<Object> filter, Function<Object, Object> mapper,
-			Function<Object, RoutingTarget> router, Function<Object, Map<String, Object>> headers) {
+			Function<Object, RoutingTarget> router, Function<Object, Map<String, Object>> headers,
+			boolean serializeExternalization) {
 
 		Assert.notNull(filter, "Filter must not be null!");
 		Assert.notNull(mapper, "Mapper must not be null!");
@@ -53,6 +55,7 @@ class DefaultEventExternalizationConfiguration implements EventExternalizationCo
 		this.mapper = mapper;
 		this.router = router;
 		this.headers = headers;
+		this.serializeExternalization = serializeExternalization;
 	}
 
 	/**
@@ -111,4 +114,13 @@ public Map<String, Object> getHeadersFor(Object event) {
 
 		return headers.apply(event);
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.modulith.events.EventExternalizationConfiguration#serializeExternalization()
+	 */
+	@Override
+	public boolean serializeExternalization() {
+		return serializeExternalization;
+	}
 }

spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/EventExternalizationConfiguration.java

@@ -199,6 +199,15 @@ public static Function<Object, RoutingTarget> byFullyQualifiedTypeName() {
 	 */
 	Map<String, Object> getHeadersFor(Object event);
 
+	/**
+	 * Returns whether the event externalization should be serialized to make sure the broker sees the events in the order
+	 * they were published in the application. By default, this is not guaranteed as multiple threads might trigger events
+	 * and the externalization of one event might overtake the one of another.
+	 *
+	 * @since 2.0
+	 */
+	boolean serializeExternalization();
+
 	/**
 	 * API to define which events are supposed to be selected for externalization.
 	 *
@@ -372,13 +381,26 @@ private static <T extends Annotation> T findAnnotation(Object event, Class<T> an
 		}
 	}
 
+	public static abstract class BaseConfiguration<T extends BaseConfiguration<T>> {
+
+		protected boolean serializeExternalization;
+
+		@SuppressWarnings("unchecked")
+		public T serializeExternalization(boolean serializeExternalization) {
+
+			this.serializeExternalization = serializeExternalization;
+
+			return (T) this;
+		}
+	}
+
 	/**
 	 * API to define the event routing.
 	 *
 	 * @author Oliver Drotbohm
 	 * @since 1.1
 	 */
-	public static class Router {
+	public static class Router extends BaseConfiguration<Router> {
 
 		private static final Function<Object, RoutingTarget> DEFAULT_ROUTER = it -> {
 			return mergeWithExternalizedAnnotation(it, byFullyQualifiedTypeName().apply(it));
@@ -398,7 +420,7 @@ public static class Router {
 		 * @param headers must not be {@literal null}.
 		 */
 		Router(Predicate<Object> filter, Function<Object, Object> mapper, Function<Object, RoutingTarget> router,
-				Function<Object, Map<String, Object>> headers) {
+				Function<Object, Map<String, Object>> headers, boolean serializeExternalization) {
 
 			Assert.notNull(filter, "Selector must not be null!");
 			Assert.notNull(mapper, "Mapper must not be null!");
@@ -409,6 +431,7 @@ public static class Router {
 			this.mapper = mapper;
 			this.router = router;
 			this.headers = headers;
+			this.serializeExternalization = serializeExternalization;
 		}
 
 		/**
@@ -417,7 +440,7 @@ public static class Router {
 		 * @param filter must not be {@literal null}.
 		 */
 		Router(Predicate<Object> filter) {
-			this(filter, Function.identity(), DEFAULT_ROUTER, it -> Collections.emptyMap());
+			this(filter, Function.identity(), DEFAULT_ROUTER, it -> Collections.emptyMap(), false);
 		}
 
 		/**
@@ -431,7 +454,7 @@ public Router mapping(Function<Object, Object> mapper) {
 
 			Assert.notNull(mapper, "Mapper must not be null!");
 
-			return new Router(filter, mapper, router, headers);
+			return new Router(filter, mapper, router, headers, serializeExternalization);
 		}
 
 		/**
@@ -453,7 +476,7 @@ public <T> Router mapping(Class<T> type, Function<T, Object> mapper) {
 					.map(mapper::apply)
 					.orElse(it);
 
-			return new Router(filter, this.mapper.compose(combined), router, headers);
+			return new Router(filter, this.mapper.compose(combined), router, headers, serializeExternalization);
 		}
 
 		/**
@@ -469,7 +492,7 @@ public Router headers(Function<Object, Map<String, Object>> extractor) {
 
 			Assert.notNull(extractor, "Headers extractor must not be null!");
 
-			return new Router(filter, mapper, router, extractor);
+			return new Router(filter, mapper, router, extractor, serializeExternalization);
 		}
 
 		/**
@@ -488,7 +511,7 @@ public <T> Router headers(Class<T> type, Function<T, Map<String, Object>> extrac
 					.map(extractor::apply)
 					.orElseGet(() -> this.headers.apply(it));
 
-			return new Router(filter, mapper, router, combined);
+			return new Router(filter, mapper, router, combined, serializeExternalization);
 		}
 
 		/**
@@ -497,7 +520,7 @@ public <T> Router headers(Class<T> type, Function<T, Map<String, Object>> extrac
 		 * @return will never be {@literal null}.
 		 */
 		public Router routeMapped() {
-			return new Router(filter, mapper, router.compose(mapper), headers);
+			return new Router(filter, mapper, router.compose(mapper), headers, serializeExternalization);
 		}
 
 		/**
@@ -510,7 +533,7 @@ public Router routeAll(Function<Object, RoutingTarget> router) {
 
 			Assert.notNull(router, "Router must not be null!");
 
-			return new Router(filter, mapper, router, headers);
+			return new Router(filter, mapper, router, headers, serializeExternalization);
 		}
 
 		/**
@@ -530,7 +553,7 @@ public <T> Router route(Class<T> type, Function<T, RoutingTarget> router) {
 					.map(router::apply)
 					.orElseGet(() -> this.router.apply(it));
 
-			return new Router(filter, mapper, adapted, headers);
+			return new Router(filter, mapper, adapted, headers, serializeExternalization);
 		}
 
 		/**
@@ -553,7 +576,7 @@ public <T> Router routeKey(Class<T> type, Function<T, String> extractor) {
 					.map(t -> this.router.apply(t).withKey(extractor.apply(t)))
 					.orElseGet(() -> this.router.apply(it));
 
-			return new Router(filter, mapper, adapted, headers);
+			return new Router(filter, mapper, adapted, headers, serializeExternalization);
 		}
 
 		/**
@@ -569,7 +592,7 @@ public EventExternalizationConfiguration routeOptional(Function<Object, Optional
 
 			Function<Object, RoutingTarget> adapted = it -> router.apply(it).orElseGet(() -> this.router.apply(it));
 
-			return new Router(filter, mapper, adapted, headers).build();
+			return new Router(filter, mapper, adapted, headers, serializeExternalization).build();
 		}
 
 		/**
@@ -597,7 +620,7 @@ public Router routeAllByType(Function<Class<?>, RoutingTarget> router) {
 
 			Assert.notNull(router, "Router must not be null!");
 
-			return new Router(filter, mapper, it -> router.apply(it.getClass()), headers);
+			return new Router(filter, mapper, it -> router.apply(it.getClass()), headers, serializeExternalization);
 		}
 
 		/**
@@ -606,7 +629,7 @@ public Router routeAllByType(Function<Class<?>, RoutingTarget> router) {
 		 * @return will never be {@literal null}.
 		 */
 		public EventExternalizationConfiguration build() {
-			return new DefaultEventExternalizationConfiguration(filter, mapper, router, headers);
+			return new DefaultEventExternalizationConfiguration(filter, mapper, router, headers, serializeExternalization);
 		}
 
 		private static <T> Optional<T> toOptional(Class<T> type, Object source) {

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/config/EventExternalizationAutoConfiguration.java

@@ -31,6 +31,7 @@
 import org.springframework.context.annotation.Role;
 import org.springframework.context.event.EventListenerFactory;
 import org.springframework.core.Ordered;
+import org.springframework.core.env.Environment;
 import org.springframework.modulith.events.EventExternalizationConfiguration;
 import org.springframework.modulith.events.core.ConditionalEventListener;
 import org.springframework.transaction.event.TransactionalApplicationListenerMethodAdapter;
@@ -61,14 +62,24 @@ static EventListenerFactory filteringEventListenerFactory(EventExternalizationCo
 	@Bean
 	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 	@ConditionalOnMissingBean
-	static EventExternalizationConfiguration eventExternalizationConfiguration(BeanFactory factory) {
+	static EventExternalizationConfiguration eventExternalizationConfiguration(BeanFactory factory,
+			Environment environment) {
 
 		var packages = AutoConfigurationPackages.get(factory);
 
-		LOG.debug("Configuring event externalization to export events annotated with @Externalized in packages: {}",
-				packages);
+		var configuration = EventExternalizationConfiguration.defaults(packages);
 
-		return EventExternalizationConfiguration.defaults(packages).build();
+		var serialize = environment.getProperty("spring.modulith.events.externalization.serialize-externalization",
+				boolean.class);
+
+		if (serialize != null) {
+			configuration = configuration.serializeExternalization(serialize);
+		}
+
+		LOG.debug("Configuring {}event externalization to export events annotated with @Externalized in packages: {}",
+				Boolean.TRUE.equals(serialize) ? "serialized " : "", packages);
+
+		return configuration.build();
 	}
 
 	/**

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/EventExternalizationSupport.java

@@ -16,6 +16,7 @@
 package org.springframework.modulith.events.support;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -39,6 +40,7 @@ abstract class EventExternalizationSupport implements ConditionalEventListener {
 	private static final Logger logger = LoggerFactory.getLogger(EventExternalizationSupport.class);
 
 	private final EventExternalizationConfiguration configuration;
+	private final Semaphore semaphore = new Semaphore(1);
 
 	/**
 	 * Creates a new {@link EventExternalizationSupport} for the given {@link EventExternalizationConfiguration}.
@@ -85,8 +87,9 @@ public CompletableFuture<?> externalize(Object event) {
 			logger.debug("Externalizing event of type {} to {}.", event.getClass(), target);
 		}
 
-		return externalize(mapped, target)
-				.thenApply(it -> new EventExternalized<>(event, mapped, target, it));
+		return configuration.serializeExternalization()
+				? doExternalizeSerialized(event, mapped, target)
+				: doExternalize(event, mapped, target);
 	}
 
 	/**
@@ -97,4 +100,26 @@ public CompletableFuture<?> externalize(Object event) {
 	 * @return the externalization result, will never be {@literal null}.
 	 */
 	protected abstract CompletableFuture<?> externalize(Object payload, RoutingTarget target);
+
+	private CompletableFuture<?> doExternalizeSerialized(Object event, Object mapped, RoutingTarget target) {
+
+		try {
+
+			semaphore.acquire();
+
+			return doExternalize(event, mapped, target)
+					.whenComplete((__, ___) -> semaphore.release());
+
+		} catch (InterruptedException o_O) {
+
+			semaphore.release();
+			throw new RuntimeException(o_O);
+		}
+	}
+
+	private CompletableFuture<?> doExternalize(Object event, Object mapped, RoutingTarget target) {
+
+		return externalize(mapped, target)
+				.thenApply(it -> new EventExternalized<>(event, mapped, target, it));
+	}
 }

spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/config/EventExternalizationAutoConfigurationIntegrationTests.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.modulith.events.config;
+
+import static org.assertj.core.api.Assertions.*;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.modulith.events.EventExternalizationConfiguration;
+
+/**
+ * Integration tests for {@link EventExternalizationAutoConfiguration}.
+ *
+ * @author Oliver Drotbohm
+ * @since 2.0
+ */
+class EventExternalizationAutoConfigurationIntegrationTests {
+
+	@Test // GH-1370
+	void doesNotExternalizationSerializationByDefault() {
+		assertSerializationEnabled(null, false);
+	}
+
+	@Test // GH-1370
+	void configurationPropertyEnablesOrDisablesSerialization() {
+
+		assertSerializationEnabled(true, true);
+		assertSerializationEnabled(false, false);
+	}
+
+	private static void assertSerializationEnabled(Boolean propertyValue, boolean expected) {
+
+		var runner = new ApplicationContextRunner()
+				.withUserConfiguration(SampleApplication.class)
+				.withConfiguration(AutoConfigurations.of(EventExternalizationAutoConfiguration.class));
+
+		if (propertyValue != null) {
+
+			runner = runner
+					.withPropertyValues("spring.modulith.events.externalization.serialize-externalization=" + propertyValue);
+		}
+
+		runner.run(ctx -> {
+
+			assertThat(ctx).getBean(EventExternalizationConfiguration.class).satisfies(it -> {
+				assertThat(it.serializeExternalization()).isEqualTo(expected);
+			});
+		});
+	}
+
+	@AutoConfigurationPackage
+	static class SampleApplication {}
+}