Schedule reactive retries in Netty event loop thread pool.

By default, Project Reactor schedules delayed operations on a parallel scheduler, which is created lazily when needed.
We can avoid creating such an extra resource by schedule the delayed work on our existing netty event loop thread pool.
This behaviour is exactly the same as our existing async retries.

Added tests to verify that the Parallel scheduler is not created.

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/internal/retry/ExponentialBackoffRetryLogic.java

@@ -23,6 +23,7 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.util.context.Context;
 import reactor.util.function.Tuples;
 
@@ -33,16 +34,16 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Supplier;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
-import org.neo4j.driver.internal.util.Clock;
-import org.neo4j.driver.internal.util.Futures;
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.Logging;
 import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.exceptions.SessionExpiredException;
 import org.neo4j.driver.exceptions.TransientException;
+import org.neo4j.driver.internal.util.Clock;
+import org.neo4j.driver.internal.util.Futures;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -179,12 +180,15 @@ private Function<Flux<Throwable>,Publisher<Context>> retryRxCondition()
                 nextDelayMs = (long) (nextDelayMs * multiplier);
                 errors = recordError( lastError, errors );
 
+                // retry on netty event loop thread
+                EventExecutor eventExecutor = eventExecutorGroup.next();
                 return Mono.just( ctx.put( "errors", errors ).put( "startTime", startTime ).put( "nextDelayMs", nextDelayMs ) )
-                        .delayElement( Duration.ofMillis( delayWithJitterMs ) );
+                        .delayElement( Duration.ofMillis( delayWithJitterMs ), Schedulers.fromExecutorService( eventExecutor ) );
             }
             else
             {
                 addSuppressed( lastError, errors );
+
                 return Mono.error( lastError );
             }
         } );

driver/src/test/java/org/neo4j/driver/integration/reactive/RxSessionIT.java

@@ -27,6 +27,7 @@
 
 import java.util.Arrays;
 import java.util.Iterator;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.neo4j.driver.Session;
@@ -46,6 +47,8 @@
 
 import static java.util.Collections.emptyIterator;
 import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.CoreMatchers.startsWith;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.neo4j.driver.internal.util.Neo4jFeature.BOLT_V4;
@@ -119,6 +122,7 @@ void shouldRunAsyncTransactionWithRetriesOnAsyncFailures()
 
         assertEquals( 4, work.invocationCount() );
         assertEquals( 1, countNodesByLabel( "Node" ) );
+        assertNoParallelScheduler();
     }
 
     @Test
@@ -134,6 +138,7 @@ void shouldRunAsyncTransactionWithRetriesOnSyncFailures()
 
         assertEquals( 3, work.invocationCount() );
         assertEquals( 1, countNodesByLabel( "Test" ) );
+        assertNoParallelScheduler();
     }
 
     @Test
@@ -150,6 +155,7 @@ void shouldRunAsyncTransactionThatCanNotBeRetried()
 
         assertEquals( 1, work.invocationCount() );
         assertEquals( 0, countNodesByLabel( "Hi" ) );
+        assertNoParallelScheduler();
     }
 
     @Test
@@ -173,6 +179,17 @@ void shouldRunAsyncTransactionThatCanNotBeRetriedAfterATransientFailure()
 
         assertEquals( 2, work.invocationCount() );
         assertEquals( 0, countNodesByLabel( "Person" ) );
+        assertNoParallelScheduler();
+    }
+
+    private void assertNoParallelScheduler()
+    {
+        Set<Thread> threadSet = Thread.getAllStackTraces().keySet();
+        for ( Thread t : threadSet )
+        {
+            String name = t.getName();
+            assertThat( name, not( startsWith( "parallel" ) ) );
+        }
     }
 
     private long countNodesByLabel( String label )