Event publication is marked completed even when event externalization fails

[xsreality]

When using Event Publication Registry in combination with Event externalization to Kafka, the entry in the registration for externalize listener is marked completed even when the message fails to be produced to Kafka.

Setup details

Event publication registry is enabled with spring-modulith-starter-jpa library. Event externalization is setup with Kafka using spring-modulith-events-kafka library.

To simulate error scenario, the bootstrap URL is misconfigured.

Expected Behaviour

When the Kafka producer times out, the transaction associated with "externalize" is rolled back. The associated entry in the publication registry should remain incomplete.

Actual Behaviour

When the Kafka producer times out after 60s (max.block.ms), the transaction associated with "externalize" is rolled back. A new transaction starts for marking the associated entry in the publication registry completed. See logs below.

If more information or source code is required, let me know and will add it.

Versions
Spring Modulith 1.1.0
Spring Boot 3.2.0
JDK 21

Logs

[nio-8080-exec-1] : Creating new transaction with name [example.order.OrderManagement.create]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
[nio-8080-exec-1] : Opened new EntityManager [SessionImpl(260294974<open>)] for JPA transaction
[nio-8080-exec-1] : Exposing JPA transaction as JDBC [org.springframework.orm.jpa.vendor.HibernateJpaDialect$HibernateConnectionHandle@5ef6a9c3]
[nio-8080-exec-1] : Found thread-bound EntityManager [SessionImpl(260294974<open>)] for JPA transaction
[nio-8080-exec-1] : Participating in existing transaction
Hibernate: insert into orders (product,status) values (?,?)
[nio-8080-exec-1] : Registering publication of example.order.OrderCreated for org.springframework.modulith.events.support.DelegatingEventExternalizer.externalize(java.lang.Object).
[nio-8080-exec-1] : Found thread-bound EntityManager [SessionImpl(260294974<open>)] for JPA transaction
[nio-8080-exec-1] : Participating in existing transaction
[nio-8080-exec-1] : Registered transaction synchronization for org.springframework.context.PayloadApplicationEvent[source=org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@17134190, started on Mon Nov 27 16:42:51 GMT 2023]
[nio-8080-exec-1] : Order created
[nio-8080-exec-1] : Initiating transaction commit
[nio-8080-exec-1] : Committing JPA transaction on EntityManager [SessionImpl(260294974<open>)]
Hibernate: insert into event_publication (completion_date,event_type,listener_id,publication_date,serialized_event,id) values (?,?,?,?,?,?)
[         task-1] : Creating new transaction with name [org.springframework.modulith.events.support.DelegatingEventExternalizer.externalize]: PROPAGATION_REQUIRES_NEW,ISOLATION_DEFAULT
[         task-1] : Opened new EntityManager [SessionImpl(566016861<open>)] for JPA transaction
[         task-1] : Exposing JPA transaction as JDBC [org.springframework.orm.jpa.vendor.HibernateJpaDialect$HibernateConnectionHandle@2f7b511a]
[nio-8080-exec-1] : Closing JPA EntityManager [SessionImpl(260294974<open>)] after transaction
[         task-1] : ProducerConfig values:
	acks = -1
	auto.include.jmx.reporter = true
	batch.size = 16384
	bootstrap.servers = [kafka:9095]
	buffer.memory = 33554432
	client.dns.lookup = use_all_dns_ips
	client.id = producer-1
	compression.type = none
	connections.max.idle.ms = 540000
	delivery.timeout.ms = 120000
	enable.idempotence = true
	interceptor.classes = []
	key.serializer = class org.apache.kafka.common.serialization.StringSerializer
	linger.ms = 0
	max.block.ms = 60000
	max.in.flight.requests.per.connection = 5
	max.request.size = 1048576
	metadata.max.age.ms = 300000
	metadata.max.idle.ms = 300000
	metric.reporters = [io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.OpenTelemetryMetricsReporter]
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partitioner.adaptive.partitioning.enable = true
	partitioner.availability.timeout.ms = 0
	partitioner.class = null
	partitioner.ignore.keys = false
	receive.buffer.bytes = 32768
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 2147483647
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.connect.timeout.ms = null
	sasl.login.read.timeout.ms = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.login.retry.backoff.max.ms = 10000
	sasl.login.retry.backoff.ms = 100
	sasl.mechanism = GSSAPI
	sasl.oauthbearer.clock.skew.seconds = 30
	sasl.oauthbearer.expected.audience = null
	sasl.oauthbearer.expected.issuer = null
	sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
	sasl.oauthbearer.jwks.endpoint.url = null
	sasl.oauthbearer.scope.claim.name = scope
	sasl.oauthbearer.sub.claim.name = sub
	sasl.oauthbearer.token.endpoint.url = null
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	transaction.timeout.ms = 60000
	transactional.id = null
	value.serializer = class org.springframework.kafka.support.serializer.JsonSerializer
outbox-demo-order-monolith-1        |
[         task-1] : [Producer clientId=producer-1] Instantiated an idempotent producer.
[         task-1] : Kafka version: 3.6.0
[         task-1] : Kafka commitId: 60e845626d8a465a
[         task-1] : Kafka startTimeMs: 1701103429656
[ad | producer-1] : [Producer clientId=producer-1] Node -1 disconnected.
[ad | producer-1] : [Producer clientId=producer-1] Connection to node -1 (kafka/172.18.0.2:9095) could not be established. Broker may not be available.
[ad | producer-1] : [Producer clientId=producer-1] Bootstrap broker kafka:9095 (id: -1 rack: null) disconnected

< repeating logs for 60s (max.block.ms) >

[ad | producer-1] : [Producer clientId=producer-1] Connection to node -1 (kafka/172.18.0.2:9095) could not be established. Broker may not be available.
[ad | producer-1] : [Producer clientId=producer-1] Bootstrap broker kafka:9095 (id: -1 rack: null) disconnected
[         task-1] : Exception thrown when sending a message with key='2' and payload='OrderCreated[id=2, product=product=CoffeeCheck]' to topic order-created:
org.apache.kafka.common.errors.TimeoutException: Topic order-created not present in metadata after 60000 ms.
[         task-1] : Transactional code has requested rollback
[         task-1] : Initiating transaction rollback
[         task-1] : Rolling back JPA transaction on EntityManager [SessionImpl(566016861<open>)]
[         task-1] : Closing JPA EntityManager [SessionImpl(566016861<open>)] after transaction
[         task-1] : Creating new transaction with name [org.springframework.modulith.events.core.DefaultEventPublicationRegistry.markCompleted]: PROPAGATION_REQUIRES_NEW,ISOLATION_DEFAULT
[         task-1] : Opened new EntityManager [SessionImpl(354507834<open>)] for JPA transaction
[         task-1] : Exposing JPA transaction as JDBC [org.springframework.orm.jpa.vendor.HibernateJpaDialect$HibernateConnectionHandle@2acf1ea4]
[         task-1] : Marking publication of event example.order.OrderCreated to listener org.springframework.modulith.events.support.DelegatingEventExternalizer.externalize(java.lang.Object) completed.
[         task-1] : Found thread-bound EntityManager [SessionImpl(354507834<open>)] for JPA transaction
[         task-1] : Participating in existing transaction
[ad | producer-1] : [Producer clientId=producer-1] Node -1 disconnected.
[ad | producer-1] : [Producer clientId=producer-1] Connection to node -1 (kafka/172.18.0.2:9095) could not be established. Broker may not be available.
[ad | producer-1] : [Producer clientId=producer-1] Bootstrap broker kafka:9095 (id: -1 rack: null) disconnected
Hibernate: update event_publication set completion_date=? where serialized_event=? and listener_id=?
[         task-1] : Initiating transaction commit
[         task-1] : Committing JPA transaction on EntityManager [SessionImpl(354507834<open>)]
[         task-1] : Closing JPA EntityManager [SessionImpl(354507834<open>)] after transaction
[         task-1] : Unexpected error occurred in asynchronous listener
java.util.concurrent.CompletionException: org.springframework.kafka.KafkaException: Send failed
	at org.springframework.util.concurrent.FutureUtils.lambda$toSupplier$0(FutureUtils.java:78)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.springframework.kafka.KafkaException: Send failed
	at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:798)
	at org.springframework.kafka.core.KafkaTemplate.observeSend(KafkaTemplate.java:756)
	at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:545)
	at org.springframework.modulith.events.kafka.KafkaEventExternalizerConfiguration.lambda$kafkaEventExternalizer$0(KafkaEventExternalizerConfiguration.java:64)
	at org.springframework.modulith.events.support.DelegatingEventExternalizer.externalize(DelegatingEventExternalizer.java:76)
	at org.springframework.modulith.events.support.EventExternalizationSupport.externalize(EventExternalizationSupport.java:87)
	at org.springframework.modulith.events.support.DelegatingEventExternalizer.externalize(DelegatingEventExternalizer.java:64)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(Unknown Source)
	at java.base/java.lang.reflect.Method.invoke(Unknown Source)
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:352)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:196)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:765)
	at org.springframework.transaction.interceptor.TransactionInterceptor$1.proceedWithInvocation(TransactionInterceptor.java:123)
	at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:385)
	at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:765)
	at org.springframework.modulith.events.support.CompletionRegisteringAdvisor$CompletionRegisteringMethodInterceptor.invoke(CompletionRegisteringAdvisor.java:169)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:765)
	at org.springframework.aop.interceptor.AsyncExecutionInterceptor.lambda$invoke$0(AsyncExecutionInterceptor.java:115)
	at org.springframework.util.concurrent.FutureUtils.lambda$toSupplier$0(FutureUtils.java:74)
	... 4 common frames omitted
Caused by: org.apache.kafka.common.errors.TimeoutException: Topic order-created not present in metadata after 60000 ms.
[ad | producer-1] : [Producer clientId=producer-1] Node -1 disconnected.
[ad | producer-1] : [Producer clientId=producer-1] Connection to node -1 (kafka/172.18.0.2:9095) could not be established. Broker may not be available.
[ad | producer-1] : [Producer clientId=producer-1] Bootstrap broker kafka:9095 (id: -1 rack: null) disconnected

[xsreality]

I think the introduction of CompletableFuture results in this issue.

The result of method invocation in CompletionRegisteringAdvisor should check if the CompletableFuture completed with an exception and leave the publication incomplete.

[odrotbohm]

That's a great catch and thank you for the PR. I wonder if we shouldn't rather register the completion on the CompletableFuture if we receive one. If I understand correctly, your suggested change would require the CF having already failed when the result is returned. In case it'd still be processing it might pass that spot, mark the publication completed and fail eventually. Decorating the CF with a "mark completed on success" should avoid that scenario, shouldn't it?

[odrotbohm]

I have a working draft for described, alternative approach. Polishing things up.

[odrotbohm]

I've pushed the fix to the 1.1 branch. Would you mind giving the snapshots a try? If everything looks good now, I'd port the fix back to 1.0.x.

[xsreality]

Thanks! Your fix is much better :) I will give it a try and report back.

[xsreality]

Works correctly! Both for success and error scenario.