I need to write code that will:
I created a method like below:
@WithSession
public Uni<Void> retrieveDataAndSaveInLocalDB() {
final Sample timer = Timer.start(meterRegistry);
return Uni.createFrom()
.deferred(() -> Uni.createFrom()
.item(dbService::connectToDB))
.onFailure()
.retry()
.withBackOff(Duration.ofSeconds(3), Duration.ofMinutes(5))
.atMost(3)
.map(this::getListOfEntities)
.flatMap(entityRepository::persist)
.call(entityRepository::flush)
.invoke(() -> timer.stop(meterRegistry.timer("timer")))
.invoke(() -> Log.info("logged info"));
}
When dbService::connectToDB throws an exception, I want to give it another go in a few seconds. The problem I have is that when dbService::connectToDB throws an error, retries are done on another thread. Here are the logs:
2023-04-18 12:09:42,926 INFO [DbService] (vert.x-eventloop-thread-1) CONNECTED! org.postgresql.jdbc.PgConnection@b931fa7
2023-04-18 12:09:46,142 INFO [DbService] (executor-thread-1) CONNECTED! org.postgresql.jdbc.PgConnection@73903696
2023-04-18 12:09:51,210 INFO [DbService] (executor-thread-1) CONNECTED! org.postgresql.jdbc.PgConnection@379aed2d
2023-04-18 12:09:51,237 ERROR [io.qua.ver.htt.run.QuarkusErrorHandler] (vert.x-eventloop-thread-1) HTTP Request to /sampleendpoint failed, error id: 8dc81df2-e526-4a64-a67d-bf04cbb9e580-1: io.smallrye.mutiny.CompositeException: Multiple exceptions caught:
[Exception 0] java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [107]: 'vert.x-eventloop-thread-1' current Thread [99]: 'executor-thread-1'
[Exception 1] java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [107]: 'vert.x-eventloop-thread-1' current Thread [99]: 'executor-thread-1'
at io.smallrye.mutiny.groups.UniOnItemOrFailure.lambda$call$1(UniOnItemOrFailure.java:75)
at io.smallrye.context.impl.wrappers.SlowContextualBiFunction.apply(SlowContextualBiFunction.java:21)
at io.smallrye.mutiny.operators.uni.UniOnItemOrFailureFlatMap$UniOnItemOrFailureFlatMapProcessor.performInnerSubscription(UniOnItemOrFailureFlatMap.java:86)
at io.smallrye.mutiny.operators.uni.UniOnItemOrFailureFlatMap$UniOnItemOrFailureFlatMapProcessor.onFailure(UniOnItemOrFailureFlatMap.java:65)
at io.smallrye.mutiny.operators.uni.UniOnTermination$UniOnTerminationProcessor.onFailure(UniOnTermination.java:52)
at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onFailure(UniOperatorProcessor.java:55)
at io.smallrye.mutiny.helpers.EmptyUniSubscription.propagateFailureEvent(EmptyUniSubscription.java:40)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage.subscribe(UniCreateFromCompletionStage.java:26)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniRunSubscribeOn.lambda$subscribe$0(UniRunSubscribeOn.java:27)
at org.hibernate.reactive.context.impl.VertxContext.execute(VertxContext.java:90)
at io.smallrye.mutiny.operators.uni.UniRunSubscribeOn.subscribe(UniRunSubscribeOn.java:25)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniOnTermination.subscribe(UniOnTermination.java:21)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniOnItemOrFailureFlatMap.subscribe(UniOnItemOrFailureFlatMap.java:27)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniOnItemOrFailureFlatMap$UniOnItemOrFailureFlatMapProcessor.performInnerSubscription(UniOnItemOrFailureFlatMap.java:99)
at io.smallrye.mutiny.operators.uni.UniOnItemOrFailureFlatMap$UniOnItemOrFailureFlatMapProcessor.onFailure(UniOnItemOrFailureFlatMap.java:65)
at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onFailure(UniOperatorProcessor.java:55)
at io.smallrye.mutiny.operators.uni.UniOnItemConsume$UniOnItemComsumeProcessor.onFailure(UniOnItemConsume.java:65)
at io.smallrye.mutiny.operators.uni.UniOnItemConsume$UniOnItemComsumeProcessor.onFailure(UniOnItemConsume.java:65)
at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onFailure(UniOperatorProcessor.java:55)
at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onFailure(UniOperatorProcessor.java:55)
at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onFailure(UniOperatorProcessor.java:55)
at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onFailure(UniOperatorProcessor.java:55)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage$CompletionStageUniSubscription.forwardResult(UniCreateFromCompletionStage.java:58)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture.uniWhenCompleteStage(CompletableFuture.java:887)
at java.base/java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:2325)
at java.base/java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:144)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage$CompletionStageUniSubscription.forward(UniCreateFromCompletionStage.java:51)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage.subscribe(UniCreateFromCompletionStage.java:35)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniRunSubscribeOn.lambda$subscribe$0(UniRunSubscribeOn.java:27)
at org.hibernate.reactive.context.impl.VertxContext.execute(VertxContext.java:90)
at io.smallrye.mutiny.operators.uni.UniRunSubscribeOn.subscribe(UniRunSubscribeOn.java:25)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.performInnerSubscription(UniOnItemTransformToUni.java:81)
at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:57)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromKnownItem$KnownItemSubscription.forward(UniCreateFromKnownItem.java:38)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromKnownItem.subscribe(UniCreateFromKnownItem.java:23)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni.subscribe(UniOnItemTransformToUni.java:25)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.performInnerSubscription(UniOnItemTransformToUni.java:81)
at io.smallrye.mutiny.operators.uni.UniOnItemTransformToUni$UniOnItemTransformToUniProcessor.onItem(UniOnItemTransformToUni.java:57)
at io.smallrye.mutiny.operators.uni.UniOnItemTransform$UniOnItemTransformProcessor.onItem(UniOnItemTransform.java:43)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromPublisher$PublisherSubscriber.onNext(UniCreateFromPublisher.java:70)
at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:30)
at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:84)
at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)
at io.smallrye.mutiny.subscription.SerializedSubscriber.onItem(SerializedSubscriber.java:74)
at io.smallrye.mutiny.operators.multi.MultiRetryWhenOp$RetryWhenOperator.onItem(MultiRetryWhenOp.java:111)
at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)
at io.smallrye.mutiny.converters.uni.UniToMultiPublisher$UniToMultiSubscription.onItem(UniToMultiPublisher.java:92)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromItemSupplier.subscribe(UniCreateFromItemSupplier.java:29)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromDeferredSupplier.subscribe(UniCreateFromDeferredSupplier.java:36)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.converters.uni.UniToMultiPublisher$UniToMultiSubscription.request(UniToMultiPublisher.java:73)
at io.smallrye.mutiny.subscription.SwitchableSubscriptionSubscriber.setOrSwitchUpstream(SwitchableSubscriptionSubscriber.java:205)
at io.smallrye.mutiny.subscription.SwitchableSubscriptionSubscriber.onSubscribe(SwitchableSubscriptionSubscriber.java:107)
at io.smallrye.mutiny.converters.uni.UniToMultiPublisher.subscribe(UniToMultiPublisher.java:25)
at io.smallrye.mutiny.groups.MultiCreate$1.subscribe(MultiCreate.java:165)
at io.smallrye.mutiny.operators.multi.MultiRetryWhenOp$RetryWhenOperator.resubscribe(MultiRetryWhenOp.java:157)
at io.smalliny.operators.multi.MultiRetryWhenOp$TriggerSubscriber.onNext(MultiRetryWhenOp.java:188)
at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:30)
at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:84)
at io.smallrye.mutiny.operators.multi.MultiConcatMapOp$ConcatMapMainSubscriber.tryEmit(MultiConcatMapOp.java:182)
at io.smallrye.mutiny.operators.multi.MultiConcatMapOp$ConcatMapInner.onItem(MultiConcatMapOp.java:285)
at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)
at io.smallrye.mutiny.converters.uni.UniToMultiPublisher$UniToMultiSubscription.onItem(UniToMultiPublisher.java:92)
at io.smallrye.mutiny.operators.uni.UniDelayOnItem$UniDelayOnItemProcessor.lambda$onItem$0(UniDelayOnItem.java:53)
at org.jboss.threads.EnhancedQueueExecutor$RunnableScheduledFuture.performTask(EnhancedQueueExecutor.java:2892)
at org.jboss.threads.EnhancedQueueExecutor$RunnableScheduledFuture.performTask(EnhancedQueueExecutor.java:2883)
at org.jboss.threads.EnhancedQueueExecutor$AbstractScheduledFuture.run(EnhancedQueueExecutor.java:2741)
at io.quarkus.vertx.core.runtime.VertxCoreRecorder$14.runWith(VertxCoreRecorder.java:576)
at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2513)
at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1538)
at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:29)
at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:29)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:833)
Suppressed: java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [107]: 'vert.x-eventloop-thread-1' current Thread [99]: 'executor-thread-1'
at org.hibernate.reactive.common.InternalStateAssertions.assertCurrentThreadMatches(InternalStateAssertions.java:46)
at org.hibernate.reactive.session.impl.ReactiveSessionImpl.threadCheck(ReactiveSessionImpl.java:190)
at org.hibernate.reactive.session.impl.ReactiveSessionImpl.checkOpen(ReactiveSessionImpl.java:1786)
at org.hibernate.internal.AbstractSharedSessionContract.checkOpenOrWaitingForAutoClose(AbstractSharedSessionContract.java:447)
at org.hibernate.internal.SessionImpl.checkOpenOrWaitingForAutoClose(SessionImpl.java:616)
at org.hibernate.internal.SessionImpl.closeWithoutOpenChecks(SessionImpl.java:410)
at org.hibernate.internal.SessionImpl.close(SessionImpl.java:397)
at org.hibernate.reactive.session.impl.ReactiveSessionImpl.reactiveClose(ReactiveSessionImpl.java:1738)
at io.smallrye.context.impl.wrappers.SlowContextualSupplier.get(SlowContextualSupplier.java:21)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage.subscribe(UniCreateFromCompletionStage.java:24)
... 76 more
Caused by: java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [107]: 'vert.x-eventloop-thread-1' current Thread [99]: 'executor-thread-1'
at org.hibernate.reactive.common.InternalStateAssertions.assertCurrentThreadMatches(InternalStateAssertions.java:46)
at org.hibernate.reactive.session.impl.ReactiveSessionImpl.threadCheck(ReactiveSessionImpl.java:190)
at org.hibernate.reactive.session.impl.ReactiveSessionImpl.checkOpen(ReactiveSessionImpl.java:1786)
at org.hibernate.reactive.session.impl.ReactiveSessionImpl.reactivePersist(ReactiveSessionImpl.java:833)
at org.hibernate.reactive.util.impl.CompletionStages.lambda$loop$2(CompletionStages.java:178)
at org.hibernate.reactive.util.impl.CompletionStages.lambda$loop$7(CompletionStages.java:409)
at org.hibernate.reactive.util.impl.CompletionStages$ArrayLoop.next(CompletionStages.java:483)
at org.hibernate.reactive.util.async.impl.AsyncTrampoline.lambda$asyncWhile$1(AsyncTrampoline.java:215)
at org.hibernate.reactive.util.async.impl.AsyncTrampoline$TrampolineInternal.unroll(AsyncTrampoline.java:121)
at org.hibernate.reactive.util.async.impl.AsyncTrampoline$TrampolineInternal.trampoline(AsyncTrampoline.java:102)
at org.hibernate.reactive.util.async.impl.AsyncTrampoline.asyncWhile(AsyncTrampoline.java:197)
at org.hibernate.reactive.util.async.impl.AsyncTrampoline.asyncWhile(AsyncTrampoline.java:215)
at org.hibernate.reactive.util.impl.CompletionStages.loop(CompletionStages.java:410)
at org.hibernate.reactive.util.impl.CompletionStages.loop(CompletionStages.java:381)
at org.hibernate.reactive.util.impl.CompletionStages.loop(CompletionStages.java:178)
at org.hibernate.reactive.util.impl.CompletionStages.applyToAll(CompletionStages.java:505)
at org.hibernate.reactive.mutiny.impl.MutinySessionImpl.lambda$persistAll$10(MutinySessionImpl.java:229)
at io.smallrye.context.impl.wrappers.SlowContextualSupplier.get(SlowContextualSupplier.java:21)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromCompletionStage.subscribe(UniCreateFromCompletionStage.java:24)
... 51 more
The "Connected" line in logs comes from the body of dbService::connectToDB method.
Logs show that the first execution is on the correct thread, which in my case is 'vert.x-eventloop-thread-4', but then the retries are executed on 'executor-thread-1'. If an exception is not thrown in the first execution, all processing goes well. If the first exection fails, but one of the retries succeeds, the exception as above is thrown.
What's wrong with my code? Starting a new session for each retry is not possible; Quarkus complains that a new session is not on a vert.x-eventloop-thread thread.
Maybe it's also worth pointing out that when calling serviceImpactRepository::persist I actually want to persist a list of entities, not a single entity.
I am using a new version of Quarkus (3.0.0.CR2) with Reactive Hibernate.
The problem is caused by .withBackoff.
Hibernate Reactive checks two things when the session is used:
In this case, the first check is failing because of a limitation (or bug) in Mutiny: when retrying using the .withBackoff option, the uni runs in a worker thread (and not in the event-loop thread the session was created).
Note that even if you create a new session, the check will fail because the code it's not running in a worker-thread (and not a event-loop thread).
This works with Quarkus 2.16 because there is a check that makes sure that everything runs in an event-loop thread. This check has been removed in Quarkus 3, this is probably a bug.
At the moment, I don't have a workaround, but it should work if you avoid the .withBackoff option.