Intermittent Issues with ReactiveNeo4jClient or Driver not cleaning up resources

[seabamirum]

I have a neo4j/redis REST API that is built with Spring WebFlux and Spring Security. When I enable Blockhound, it's difficult to run my application for very long without encountering an error, sometimes even on sign-in. These errors sometimes lead to deadlocks and application hangs, even without Blockhound running. The problem seems to occur when I run a query with the ReactiveNeo4jClient, and then do something else async after, like logging to a Redis Stream with spring-data-redis. I have been able to reproduce with just two threads hitting "login" ( the success handler logs to a Redis stream). The stack traces suggest it could be something internal, either with the Reactive Client or perhaps the Neo4j driver itself. There are several variants of the error; however all are related to being unable to release a lock on a thread. I will keep working on making a reproducible test case, but wanted to put the info here for now.

Spring-Data-Neo4j 7.1.1
Neo4j Version: Community 5.9
Neo4j Mode: Single instance
Driver Version: 5.9 Java
Operating System: Kubuntu 22.04

Sometimes I just see:

java.lang.IllegalMonitorStateException: attempt to unlock read lock, not locked by current thread

The "null" variant:

java.lang.IllegalMonitorStateException: null
	at java.base/java.util.concurrent.locks.ReentrantReadWriteLock$Sync.tryRelease(ReentrantReadWriteLock.java:372) ~[na:na]
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 

The most common, "unsafe park", variant:

2023-06-29T13:19:15.357-04:00 ERROR 8573 --- [oundedElastic-5] a.w.r.e.AbstractErrorWebExceptionHandler : [43ea9a76-6]  500 Server Error for HTTP POST "/login"

reactor.blockhound.BlockingOperationError: Blocking call! jdk.internal.misc.Unsafe#park
	at java.base/jdk.internal.misc.Unsafe.park(Unsafe.java) ~[na:na]
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
	*__checkpoint ⇢ org.springframework.security.web.server.authentication.AuthenticationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ HTTP POST "/login" [ExceptionHandlingWebHandler]
Original Stack Trace:
		at java.base/jdk.internal.misc.Unsafe.park(Unsafe.java) ~[na:na]
		at java.base/java.util.concurrent.locks.LockSupport.park(LockSupport.java:221) ~[na:na]
		at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:754) ~[na:na]
		at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireShared(AbstractQueuedSynchronizer.java:1079) ~[na:na]
		at java.base/java.util.concurrent.locks.ReentrantReadWriteLock$ReadLock.lock(ReentrantReadWriteLock.java:738) ~[na:na]
		at org.neo4j.driver.internal.util.LockUtil.executeWithLock(LockUtil.java:37) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.async.pool.ConnectionPoolImpl.getOrCreatePool(ConnectionPoolImpl.java:276) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.async.pool.ConnectionPoolImpl.acquire(ConnectionPoolImpl.java:126) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.DirectConnectionProvider.acquirePooledConnection(DirectConnectionProvider.java:96) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.DirectConnectionProvider.acquireConnection(DirectConnectionProvider.java:52) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.async.NetworkSession.lambda$acquireConnection$23(NetworkSession.java:308) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at java.base/java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2341) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:144) ~[na:na]
		at org.neo4j.driver.internal.async.NetworkSession.acquireConnection(NetworkSession.java:303) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.async.NetworkSession.lambda$buildResultCursorFactory$17(NetworkSession.java:256) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at java.base/java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1187) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2341) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:144) ~[na:na]
		at org.neo4j.driver.internal.async.NetworkSession.buildResultCursorFactory(NetworkSession.java:256) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.async.NetworkSession.runRx(NetworkSession.java:124) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.reactivestreams.InternalReactiveSession.run(InternalReactiveSession.java:88) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.reactivestreams.InternalReactiveSession.run(InternalReactiveSession.java:81) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.reactivestreams.BaseReactiveQueryRunner.run(BaseReactiveQueryRunner.java:38) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.neo4j.driver.internal.reactivestreams.BaseReactiveQueryRunner.run(BaseReactiveQueryRunner.java:46) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DelegatingQueryRunner.run(DefaultReactiveNeo4jClient.java:148) ~[spring-data-neo4j-7.1.1.jar:7.1.1]
		at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.executeWith(DefaultReactiveNeo4jClient.java:418) ~[spring-data-neo4j-7.1.1.jar:7.1.1]
		at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DefaultRecordFetchSpec.lambda$one$5(DefaultReactiveNeo4jClient.java:427) ~[spring-data-neo4j-7.1.1.jar:7.1.1]

Here's another stacktrace I've only seen once:

java.lang.RuntimeException: Async resource cleanup failed after onComplete
	at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:520) ~[reactor-core-3.5.7.jar:3.5.7]
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
	*__checkpoint ⇢ org.springframework.security.web.server.authentication.AuthenticationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ HTTP POST "/login" [ExceptionHandlingWebHandler]
Original Stack Trace:
		at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:520) ~[reactor-core-3.5.7.jar:3.5.7]
		at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onError(MonoPeekTerminal.java:258) ~[reactor-core-3.5.7.jar:3.5.7]
		at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onComplete(MonoPeekTerminal.java:292) ~[reactor-core-3.5.7.jar:3.5.7]
		at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89) ~[reactor-core-3.5.7.jar:3.5.7]
		at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89) ~[reactor-core-3.5.7.jar:3.5.7]
		at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:140) ~[reactor-core-3.5.7.jar:3.5.7]
		at org.neo4j.driver.internal.reactive.RxUtils.lambda$createEmptyPublisher$0(RxUtils.java:44) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179) ~[na:na]
		at org.neo4j.driver.internal.handlers.ChannelReleasingResetResponseHandler.lambda$resetCompleted$2(ChannelReleasingResetResponseHandler.java:63) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[na:na]
		at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179) ~[na:na]
		at org.neo4j.driver.internal.util.Futures.lambda$asCompletionStage$0(Futures.java:73) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
		at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:557) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.concurrent.DefaultPromise.access$200(DefaultPromise.java:35) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.concurrent.DefaultPromise$1.run(DefaultPromise.java:503) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
		at java.base/java.lang.Thread.run(Thread.java:1623) ~[na:na]
Caused by: java.lang.IllegalMonitorStateException: null
	at java.base/java.util.concurrent.locks.ReentrantReadWriteLock$Sync.tryRelease(ReentrantReadWriteLock.java:372) ~[na:na]
	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.release(AbstractQueuedSynchronizer.java:1059) ~[na:na]
	at java.base/java.util.concurrent.locks.ReentrantReadWriteLock$WriteLock.unlock(ReentrantReadWriteLock.java:1147) ~[na:na]
	at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient.lambda$getQueryRunner$2(DefaultReactiveNeo4jClient.java:112) ~[spring-data-neo4j-7.1.1.jar:7.1.1]
	at org.springframework.data.neo4j.core.DefaultReactiveNeo4jClient$DelegatingQueryRunner.lambda$close$0(DefaultReactiveNeo4jClient.java:135) ~[spring-data-neo4j-7.1.1.jar:7.1.1]
	at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onComplete(MonoPeekTerminal.java:289) ~[reactor-core-3.5.7.jar:3.5.7]
	at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89) ~[reactor-core-3.5.7.jar:3.5.7]
	at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:89) ~[reactor-core-3.5.7.jar:3.5.7]
	at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:140) ~[reactor-core-3.5.7.jar:3.5.7]
	at org.neo4j.driver.internal.reactive.RxUtils.lambda$createEmptyPublisher$0(RxUtils.java:44) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[na:na]
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[na:na]
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[na:na]
	at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179) ~[na:na]
	at org.neo4j.driver.internal.handlers.ChannelReleasingResetResponseHandler.lambda$resetCompleted$2(ChannelReleasingResetResponseHandler.java:63) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[na:na]
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[na:na]
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[na:na]
	at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179) ~[na:na]
	at org.neo4j.driver.internal.util.Futures.lambda$asCompletionStage$0(Futures.java:73) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:557) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.access$200(DefaultPromise.java:35) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise$1.run(DefaultPromise.java:503) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at java.base/java.lang.Thread.run(Thread.java:1623) ~[na:na]

[seabamirum]

Here is a test case that prints out a number of elementIds, and then produces several IllegalMonitorStateExceptions (with Blockhound enabled). Using concatMap instead of flatMap or flatMapSequential avoids the blocking errors, but ideally both operators would still work.

@SpringBootTest(properties = "spring.profiles.active=dev")
public class ReactiveNeo4jClientTest {

    @Autowired
    private ReactiveNeo4jClient neo4jClient;

    @Test
    public void testQueryExecution() throws InterruptedException {
        String cypher = "MATCH (n) RETURN elementId(n)";        
        String cypher2 = "MATCH (n) WHERE elementId(n) = $elementId RETURN elementId(n)";
     
        Flux<String> elementIdFlux = neo4jClient.query(cypher).fetchAs(String.class).all();                	 
        elementIdFlux.flatMap(elementId -> 
        {
        	Mono<String> elementIdMono = neo4jClient.query(cypher2).bindAll(Map.of("elementId",elementId)).fetchAs(String.class).one();        	
        	return elementIdMono.doOnNext(returnedElementId -> System.out.println("elementId is " + returnedElementId));     	
        }).subscribe();
        
        Thread.sleep(10000);
        
    }
}

Stack trace:

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalMonitorStateException
Caused by: java.lang.IllegalMonitorStateException: null
	at java.base/java.util.concurrent.locks.ReentrantReadWriteLock$Sync.tryRelease(ReentrantReadWriteLock.java:372) ~[na:na]
	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.release(AbstractQueuedSynchronizer.java:1059) ~[na:na]
	at java.base/java.util.concurrent.locks.ReentrantReadWriteLock$WriteLock.unlock(ReentrantReadWriteLock.java:1147) ~[na:na]
	at org.neo4j.driver.internal.async.pool.NettyChannelTracker.doInWriteLock(NettyChannelTracker.java:69) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at org.neo4j.driver.internal.async.pool.NettyChannelTracker.channelAcquired(NettyChannelTracker.java:95) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at io.netty.channel.pool.SimpleChannelPool.notifyConnect(SimpleChannelPool.java:212) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.pool.SimpleChannelPool.access$000(SimpleChannelPool.java:43) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.pool.SimpleChannelPool$2.operationComplete(SimpleChannelPool.java:184) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.pool.SimpleChannelPool$2.operationComplete(SimpleChannelPool.java:181) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:557) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:492) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:636) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:625) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setSuccess(DefaultPromise.java:97) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.DefaultChannelPromise.setSuccess(DefaultChannelPromise.java:78) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.DefaultChannelPromise.setSuccess(DefaultChannelPromise.java:73) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at org.neo4j.driver.internal.async.pool.NettyChannelPool$1.lambda$connectChannel$0(NettyChannelPool.java:108) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:557) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:492) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:636) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:625) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setSuccess(DefaultPromise.java:97) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.DefaultChannelPromise.setSuccess(DefaultChannelPromise.java:78) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.DefaultChannelPromise.setSuccess(DefaultChannelPromise.java:73) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at org.neo4j.driver.internal.messaging.v53.BoltProtocolV53.initializeChannel(BoltProtocolV53.java:70) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at org.neo4j.driver.internal.async.connection.HandshakeCompletedListener.operationComplete(HandshakeCompletedListener.java:88) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at org.neo4j.driver.internal.async.connection.HandshakeCompletedListener.operationComplete(HandshakeCompletedListener.java:34) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:583) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:559) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:492) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:636) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:625) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.DefaultPromise.setSuccess(DefaultPromise.java:97) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.DefaultChannelPromise.setSuccess(DefaultChannelPromise.java:78) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.DefaultChannelPromise.setSuccess(DefaultChannelPromise.java:73) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at org.neo4j.driver.internal.async.connection.HandshakeHandler.protocolSelected(HandshakeHandler.java:119) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at org.neo4j.driver.internal.async.connection.HandshakeHandler.decode(HandshakeHandler.java:102) ~[neo4j-java-driver-5.9.0.jar:5.9.0-1250b2d94997ae49e5a623c52f61636505185057]
	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:529) ~[netty-codec-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.handler.codec.ReplayingDecoder.callDecode(ReplayingDecoder.java:366) ~[netty-codec-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290) ~[netty-codec-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286) ~[netty-handler-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.channel.nio.NioEventLoop.run(Unknown Source) ~[netty-transport-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.94.Final.jar:4.1.94.Final]
	at java.base/java.lang.Thread.run(Unknown Source) ~[na:na]

[michael-simons]

Can you have a look at the driver stack trace, please? @injectives

[meistermeier]

Thanks for reporting this. I could replace your SDN operations with direct driver operations and also reproduce the problem. I created an issue at neo4j/neo4j-java-driver#1444 but will keep this open for the upgrade of the driver.

[seabamirum]

Even with the driver code fixed, the DefaultReactiveNeo4jClient has the same blocking issue with the threads parking at java.util.concurrent.locks.LockSupport.park for access to the bookmarks set. By removing the ReentrantReadWriteLock code completely, and setting bookmarks to ConcurrentHashMap.newKeySet() I was able to run my JMeter tests on my application for the first time with absolutely no blocking errors.

private final Set<Bookmark> bookmarks = ConcurrentHashMap.newKeySet();

...

@Override
	public Mono<ReactiveQueryRunner> getQueryRunner(Mono<DatabaseSelection> databaseSelection, Mono<UserSelection> userSelection) {

		return databaseSelection.zipWith(userSelection)
				.flatMap(targetDatabaseAndUser ->
						ReactiveNeo4jTransactionManager.retrieveReactiveTransaction(driver, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())
								.map(ReactiveQueryRunner.class::cast)
								.zipWith(Mono.just(Collections.<Bookmark>emptySet()))
								.switchIfEmpty(Mono.fromSupplier(() -> {
										Set<Bookmark> lastBookmarks = new HashSet<>(bookmarks);
										return Tuples.of(driver.session(ReactiveSession.class, Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())), lastBookmarks);
									
								})))
				.map(t -> new DelegatingQueryRunner(t.getT1(), t.getT2(), (usedBookmarks, newBookmarks) -> {
						bookmarks.removeAll(usedBookmarks);
						bookmarks.addAll(newBookmarks);
				}));
	}

[meistermeier]

I went down the same road with SynchronizedSet but additionally also added a ReactiveDefaultBookmarkManager for the operations on higher abstraction levels. There is a branch right now issue/gh-2755 but need to wait for the driver fix before merging.
Thanks for the effort, you already put in the analysis and the possible solution.