IllegalStateException during inTransaction(…) with inner Mono.error(…)

[mp911de]

The following code causes an IllegalStateException:

client
    .inTransaction(
        dc ->
            passwordHistoryRepository
                .save(new PasswordHistory(1))
                .then(Mono.error(new RuntimeException("6")))
                .then(passwordHistoryRepository.save(new PasswordHistory(2))
    )

Trace:

RuntimeException, trace=java.lang.RuntimeException: Async resource cleanup failed after onComplete
	at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:519)
...
Caused by: java.lang.IllegalStateException: Connection is closed!
	at org.springframework.data.r2dbc.function.connectionfactory.SingletonConnectionFactory.create(SingletonConnectionFactory.java:54)
	at org.springframework.data.r2dbc.function.DefaultTransactionalDatabaseClient.lambda$cleanup$7(DefaultTransactionalDatabaseClient.java:142)

Related ticket: #44.

[ToffeeLu]

One thing to correct is this IllegalStateException happens only after commenting the code of Mono.error(new RuntimeException("6")), which means it doesn't work when no error happens.

[mp911de]

I wasn't able to reproduce the issue (see reproducer code. Can you provide a complete, minimal, verifiable sample that reproduces the problem, ideally as GitHub repository or Gist?

[ToffeeLu]

@mp911de check this reproducer code
Just as I said above, error only happens without Mono.error(),
here is what's different between me and your reproduce code:

    LegoSet legoSet1 = new LegoSet(null, "SCHAUFELRADBAGGER", 12);
    LegoSet legoSet2 = new LegoSet(null, "SCHAUFELRADBAGGER", 13);

    Flux<Map<String, Object>> transactional = client.inTransaction(db ->
      legoSetRepository.save(legoSet1)
          .then(legoSetRepository.save(legoSet2))
          .then(Mono.empty()));

and the error - Caused by: java.lang.IllegalStateException: Connection is closed! happens:

java.lang.AssertionError: expectation "expectError(Class)" failed (expected error of type: MyException; actual type: java.lang.RuntimeException: Async resource cleanup failed after onComplete)

	at reactor.test.ErrorFormatter.assertionError(ErrorFormatter.java:105)
	at reactor.test.ErrorFormatter.failPrefix(ErrorFormatter.java:94)
	at reactor.test.ErrorFormatter.fail(ErrorFormatter.java:64)
	at reactor.test.ErrorFormatter.failOptional(ErrorFormatter.java:79)
	at reactor.test.DefaultStepVerifierBuilder.lambda$expectError$6(DefaultStepVerifierBuilder.java:357)
	at reactor.test.DefaultStepVerifierBuilder$SignalEvent.test(DefaultStepVerifierBuilder.java:2112)
	at reactor.test.DefaultStepVerifierBuilder$DefaultVerifySubscriber.onSignal(DefaultStepVerifierBuilder.java:1408)
	at reactor.test.DefaultStepVerifierBuilder$DefaultVerifySubscriber.onExpectation(DefaultStepVerifierBuilder.java:1356)
	at reactor.test.DefaultStepVerifierBuilder$DefaultVerifySubscriber.onError(DefaultStepVerifierBuilder.java:1030)
	at reactor.core.publisher.FluxContextStart$ContextStartSubscriber.onError(FluxContextStart.java:117)
	at reactor.core.publisher.FluxUsingWhen$UsingWhenSubscriber.deferredError(FluxUsingWhen.java:397)
	at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:520)
	...
	at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:310)
	at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:141)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:648)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:583)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:500)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:462)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:897)
	at java.lang.Thread.run(Thread.java:748)
	Suppressed: java.lang.RuntimeException: Async resource cleanup failed after onComplete
		at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:519)
		... 129 more
	Caused by: java.lang.IllegalStateException: Connection is closed!
		at org.springframework.data.r2dbc.function.connectionfactory.SingletonConnectionFactory.create(SingletonConnectionFactory.java:54)
		at org.springframework.data.r2dbc.function.DefaultTransactionalDatabaseClient.lambda$cleanup$7(DefaultTransactionalDatabaseClient.java:142)
		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:118)
		... 128 more


Process finished with exit code -1

[mp911de]

Thanks a lot, I'll have a look.

[mp911de]

The initial trace results in unexpected behavior of our resource cleanup. The cleanup happens twice and that's why the refCounter in the bound connection factory is decremented twice. This results in prematurely closing the connection. Analysis ongoing.

[mp911de]

See also reactor/reactor-core#1486.

[mp911de]

That's fixed now.