reactor.core.publisher.Sinks$EmissionException: Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.

[chibenwa]

reactor.core.publisher.Sinks$EmissionException: Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.
	at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:56)
	at org.apache.james.task.eventsourcing.distributed.RabbitMQTerminationSubscriber.addEvent(RabbitMQTerminationSubscriber.java:130)
	at org.apache.james.task.eventsourcing.TerminationSubscriber.handle(TerminationSubscriber.scala:31)
	at org.apache.james.task.eventsourcing.TerminationSubscriber.$anonfun$handleReactive$1(TerminationSubscriber.scala:28)
	at reactor.core.publisher.MonoRunnable.subscribe(MonoRunnable.java:49)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4568)
	at reactor.core.scala.publisher.SMono.subscribe(SMono.scala:1136)
	at reactor.core.scala.publisher.SMono.subscribe$(SMono.scala:1136)
	at reactor.core.scala.publisher.ReactiveSMono.subscribe(ReactiveSMono.scala:8)
	at reactor.core.publisher.FluxSource.subscribe(FluxSource.java:71)
	at reactor.core.publisher.Flux.subscribe(Flux.java:8840)
	at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:452)
	at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerComplete(FluxConcatMap.java:296)
	at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onComplete(FluxConcatMap.java:889)
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onComplete(Operators.java:2231)
	at reactor.core.publisher.MonoNext$NextSubscriber.onComplete(MonoNext.java:102)
	at reactor.core.publisher.MonoNext$NextSubscriber.onComplete(MonoNext.java:102)
	at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
	at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
	at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89)
	at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89)
	at reactor.core.publisher.FluxSubscribeOnCallable$CallableSubscribeOnSubscription.run(FluxSubscribeOnCallable.java:254)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.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)

MU customer.

Investigate. Likely a data race?

[chibenwa]

Need a lock to emmit serially somewhere?