Reactive Resource Cleanup Fails After Query Completion in SDN 6.3.9 Using FluxUsingWhen

[kmmraju]

I am encountering an issue with Spring Data Neo4j 6.3.9 in a reactive environment where a RuntimeException is thrown during the cleanup phase after a query completes. This occurs when using Flux.usingWhen for resource management, which is invoked internally by SDN for reactive database interactions. The error seems related to resource cleanup (sessions or transactions) failing after onComplete.

The stack trace points to the DefaultReactiveNeo4jClient and ReactiveNeo4jTemplate in SDN. It seems to occur even for simple findById operations, which should typically not cause resource cleanup failures. I am not managing sessions manually, and all session and transaction handling is being done via the reactive flow provided by SDN.

Environment:
Spring Data Neo4j Version: 6.3.9
Neo4j Java Driver Version: 4.4.10
Reactor Version: 3.4.x
Neo4j Server Version: (Neo4j 5.24 on a hosted cloud instance)
Java Version: (OpenJDK 11)

Stack Trace:
java.lang.RuntimeException: Async resource cleanup failed after onComplete
at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:520)
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Assembly trace from producer [reactor.core.publisher.FluxUsingWhen] :
reactor.core.publisher.Flux.usingWhen
org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.executeWith(DefaultReactiveNeo4jClient.java:318)
Error has been observed at the following site(s):
*__________Flux.usingWhen ⇢ at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.executeWith(DefaultReactiveNeo4jClient.java:318)
*_______Mono.flatMapMany ⇢ at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.lambda$one$6(DefaultReactiveNeo4jClient.java:327)
| Flux.singleOrEmpty ⇢ at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.lambda$one$6(DefaultReactiveNeo4jClient.java:327)
*_______Mono.usingWhen ⇢ at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient.doInQueryRunnerForMono(DefaultReactiveNeo4jClient.java:163)
| Mono.onErrorMap ⇢ at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.one(DefaultReactiveNeo4jClient.java:328)
| Mono.map ⇢ at org.springframework.data.neo4j.core.ReactiveNeo4jTemplate.lambda$null$46(ReactiveNeo4jTemplate.java:661)
| Mono.expand ⇢ at org.springframework.data.neo4j.core.ReactiveNeo4jTemplate.lambda$null$46(ReactiveNeo4jTemplate.java:669)
*____________Flux.flatMap ⇢ at org.springframework.data.neo4j.core.ReactiveNeo4jTemplate.lambda$createNodesAndRelationshipsByIdStatementProvider$48(ReactiveNeo4jTemplate.java:649)
*______________Flux.then ⇢ at org.springframework.data.neo4j.core.ReactiveNeo4jTemplate.lambda$createNodesAndRelationshipsByIdStatementProvider$48(ReactiveNeo4jTemplate.java:671)
*Mono.deferContextual ⇢ at org.springframework.data.neo4j.core.ReactiveNeo4jTemplate.createNodesAndRelationshipsByIdStatementProvider(ReactiveNeo4jTemplate.java:642)
| Mono.contextWrite ⇢ at org.springframework.data.neo4j.core.ReactiveNeo4jTemplate.createNodesAndRelationshipsByIdStatementProvider(ReactiveNeo4jTemplate.java:673)
| Mono.flatMap ⇢ at org.springframework.data.neo4j.core.ReactiveNeo4jTemplate.createExecutableQuery(ReactiveNeo4jTemplate.java:631)
| Mono.flatMap ⇢ at org.springframework.data.neo4j.core.ReactiveNeo4jTemplate.findById(ReactiveNeo4jTemplate.java:272)
| ⇢ at org.springframework.data.neo4j.repository.support.SimpleReactiveNeo4jRepository.findById(SimpleReactiveNeo4jRepository.java:70)
| ⇢ at org.springframework.data.repository.core.support.RepositoryMethodInvoker$ReactiveInvocationListenerDecorator.lambda$decorate$1(RepositoryMethodInvoker.java:229)
*________Mono.usingWhen ⇢ at org.springframework.data.repository.core.support.RepositoryMethodInvoker$ReactiveInvocationListenerDecorator.decorate(RepositoryMethodInvoker.java:226)
| Mono.onErrorMap ⇢ at org.springframework.data.neo4j.repository.support.ReactivePersistenceExceptionTranslationInterceptor.invoke(ReactivePersistenceExceptionTranslationInterceptor.java:81)
| ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.lambda$null$0(TransactionAspectSupport.java:918)
*_________Mono.usingWhen ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.lambda$null$3(TransactionAspectSupport.java:914)
| Mono.onErrorResume ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.lambda$null$3(TransactionAspectSupport.java:927)
*_______________Mono.then ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.lambda$null$2(TransactionAspectSupport.java:928)
*____________Mono.flatMap ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.lambda$invokeWithinTransaction$4(TransactionAspectSupport.java:911)
*_________Mono.flatMap ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.invokeWithinTransaction(TransactionAspectSupport.java:910)
| Mono.contextWrite ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.invokeWithinTransaction(TransactionAspectSupport.java:934)
| Mono.contextWrite ⇢ at org.springframework.transaction.interceptor.TransactionAspectSupport$ReactiveTransactionSupport.invokeWithinTransaction(TransactionAspectSupport.java:935)
| ⇢ at com.example.service.SelectionStrategyService.findById(SelectionStrategyService.java:93)

... Followed by the original stack trace might give additional details.
Original Stack Trace:
at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:520)
at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onError(FluxPeekFuseable.java:234)
at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:205)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.StrictSubscriber.onNext(StrictSubscriber.java:89)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:165)
at org.neo4j.driver.internal.reactive.InternalRxResult.lambda$null$4(InternalRxResult.java:141)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
at java.base/java.util.concurrent.CompletableFuture.uniWhenCompleteStage(CompletableFuture.java:883)
at java.base/java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:2251)
at java.base/java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:143)
at org.neo4j.driver.internal.reactive.InternalRxResult.lambda$null$5(InternalRxResult.java:137)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
at java.base/java.util.concurrent.CompletableFuture.uniWhenCompleteStage(CompletableFuture.java:883)
at java.base/java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:2251)
at java.base/java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:143)
at org.neo4j.driver.internal.reactive.InternalRxResult.lambda$consume$6(InternalRxResult.java:134)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:57)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.Mono.subscribe(Mono.java:4400)
at reactor.core.publisher.FluxSource.subscribe(FluxSource.java:67)
at reactor.core.publisher.Flux.subscribe(Flux.java:8469)
at reactor.core.publisher.FluxUsingWhen$UsingWhenSubscriber.onComplete(FluxUsingWhen.java:384)
at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onComplete(FluxHandleFuseable.java:229)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.StrictSubscriber.onComplete(StrictSubscriber.java:123)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.FluxCreate$BaseSink.complete(FluxCreate.java:439)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.FluxCreate$SerializedFluxSink.drainLoop(FluxCreate.java:240)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.FluxCreate$SerializedFluxSink.drain(FluxCreate.java:206)
at org.neo4j.driver.internal.shaded.reactor.core.publisher.FluxCreate$SerializedFluxSink.complete(FluxCreate.java:197)
at org.neo4j.driver.internal.reactive.InternalRxResult.lambda$createRecordConsumer$3(InternalRxResult.java:103)
at org.neo4j.driver.internal.handlers.pulln.BasicPullResponseHandler.complete(BasicPullResponseHandler.java:221)
at org.neo4j.driver.internal.handlers.pulln.BasicPullResponseHandler.completeWithSuccess(BasicPullResponseHandler.java:117)
at org.neo4j.driver.internal.handlers.pulln.BasicPullResponseHandler$State$2.onSuccess(BasicPullResponseHandler.java:293)
at org.neo4j.driver.internal.handlers.pulln.BasicPullResponseHandler.onSuccess(BasicPullResponseHandler.java:75)
at org.neo4j.driver.internal.handlers.RoutingResponseHandler.onSuccess(RoutingResponseHandler.java:56)
at org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher.handleSuccessMessage(InboundMessageDispatcher.java:84)
at org.neo4j.driver.internal.messaging.common.CommonMessageReader.unpackSuccessMessage(CommonMessageReader.java:75)
at org.neo4j.driver.internal.messaging.common.CommonMessageReader.read(CommonMessageReader.java:56)
at org.neo4j.driver.internal.async.inbound.InboundMessageHandler.channelRead0(InboundMessageHandler.java:83)
at org.neo4j.driver.internal.async.inbound.InboundMessageHandler.channelRead0(InboundMessageHandler.java:35)
at org.neo4j.driver.internal.shaded.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)
at org.neo4j.driver.internal.async.inbound.MessageDecoder.channelRead(MessageDecoder.java:47)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.neo4j.driver.internal.shaded.io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1371)
at org.neo4j.driver.internal.shaded.io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1234)
at org.neo4j.driver.internal.shaded.io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1283)
at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:510)
at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:449)
at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:279)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at org.neo4j.driver.internal.shaded.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
at org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
at org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
at org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
at org.neo4j.driver.internal.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
at org.neo4j.driver.internal.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at org.neo4j.driver.internal.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.NullPointerException: null
at org.springframework.data.neo4j.core.ResultSummaries.format(ResultSummaries.java:90)
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Assembly trace from producer [reactor.core.publisher.FluxPeekFuseable] :
reactor.core.publisher.Flux.doOnNext
org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.lambda$executeWith$4(DefaultReactiveNeo4jClient.java:320)
Error has been observed at the following site(s):
*__Flux.doOnNext ⇢ at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.lambda$executeWith$4(DefaultReactiveNeo4jClient.java:320)