Повторная попытка Quarkus Mutiny не удалась, потому что в другом потоке

Мне нужно написать код, который будет:

• подключиться к внешней БД
• запросите эту БД для данных
• обрабатывать эти данные, то есть создавать несколько объектов в локальной БД

Я создал метод, как показано ниже:

  @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"));
  }

Когда dbService::connectToDB выдает исключение, я хочу повторить попытку через несколько секунд. У меня проблема в том, что когда dbService::connectToDB выдает ошибку, повторные попытки выполняются в другом потоке. Вот журналы:

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

Строка «Подключено» в логах исходит из тела метода dbService::connectToDB.

Журналы показывают, что первое выполнение выполняется в правильном потоке, в моем случае это «vert.x-eventloop-thread-4», но затем повторные попытки выполняются в «executor-thread-1». Если при первом выполнении не возникает исключение, вся обработка проходит успешно. Если первое выполнение завершается ошибкой, но одна из попыток завершается успешно, выдается исключение, как указано выше.

Что не так с моим кодом? Запуск нового сеанса для каждой повторной попытки невозможен; Quarkus жалуется, что новый сеанс находится не в потоке vert.x-eventloop-thread.

Возможно, также стоит отметить, что при вызове serviceImpactRepository::persist я действительно хочу сохранить список сущностей, а не одну сущность.

Я использую новую версию Quarkus (3.0.0.CR2) с Reactive Hibernate.