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 )

driver/src/test/java/org/neo4j/driver/internal/retry/ExponentialBackoffRetryLogicTest.java

@@ -22,21 +22,19 @@
 import org.mockito.ArgumentCaptor;
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
-import reactor.util.function.Tuple2;
 
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Supplier;
 
-import org.neo4j.driver.internal.util.Clock;
-import org.neo4j.driver.internal.util.ImmediateSchedulingEventExecutor;
 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.ImmediateSchedulingEventExecutor;
 
 import static java.lang.Long.MAX_VALUE;
 import static java.util.concurrent.CompletableFuture.completedFuture;
@@ -751,53 +749,44 @@ public CompletionStage<Void> get()
     }
 
     @Test
-    void shouldRetryWithBackOff() {
+    void shouldRetryWithBackOff()
+    {
         Exception exception = new TransientException( "Unknown", "Retry this error." );
-        List<Long> elapsedList = new ArrayList<>();
-
         ExponentialBackoffRetryLogic retryLogic = new ExponentialBackoffRetryLogic( 1000, 100, 2, 0,
-                null, Clock.SYSTEM, DEV_NULL_LOGGING );
+                eventExecutor, Clock.SYSTEM, DEV_NULL_LOGGING );
 
         Flux<Integer> source = Flux.concat( Flux.range( 0, 2 ), Flux.error( exception ) );
-        StepVerifier.withVirtualTime( () -> // This test uses a virtual time. So do not panic if you saw this test runs faster than it should be.
-                Flux.from( retryLogic.retryRx( source ) )
-                        .elapsed()
-                        .doOnNext( elapsed -> { if ( elapsed.getT2() == 0 ) elapsedList.add( elapsed.getT1() ); } )
-                        .map( Tuple2::getT2 ) )
-                .thenAwait( Duration.ofSeconds( 2 ) )
+        Flux<Integer> retriedSource = Flux.from( retryLogic.retryRx( source ) );
+        StepVerifier.create( retriedSource )
                 .expectNext( 0, 1 ) // first run
                 .expectNext( 0, 1, 0, 1, 0, 1, 0, 1 ) //4 retry attempts
                 .verifyErrorSatisfies( e -> assertThat( e, equalTo( exception ) ) );
 
-        assertThat( elapsedList.size(), equalTo( 5 ) );
-        assertThat( elapsedList, contains( 0L, 100L, 200L, 400L, 800L ) );
+        List<Long> delays = eventExecutor.scheduleDelays();
+        assertThat( delays.size(), equalTo( 4 ) );
+        assertThat( delays, contains( 100L, 200L, 400L, 800L ) );
     }
 
     @Test
-    void shouldRetryWithRandomBackOff() {
-        List<Long> elapsedList = new ArrayList<>();
+    void shouldRetryWithRandomBackOff()
+    {
         Exception exception = new TransientException( "Unknown", "Retry this error." );
-
         ExponentialBackoffRetryLogic retryLogic = new ExponentialBackoffRetryLogic( 1000, 100, 2, 0.1,
-                null, Clock.SYSTEM, DEV_NULL_LOGGING );
+                eventExecutor, Clock.SYSTEM, DEV_NULL_LOGGING );
 
         Flux<Integer> source = Flux.concat( Flux.range( 0, 2 ), Flux.error( exception ) );
-        StepVerifier.withVirtualTime( () -> // This test uses a virtual time. So do not panic if you saw this test runs faster than it should be.
-                Flux.from( retryLogic.retryRx( source ) )
-                        .elapsed()
-                        .doOnNext( elapsed -> { if ( elapsed.getT2() == 0 ) elapsedList.add( elapsed.getT1() ); } )
-                        .map( Tuple2::getT2 ) )
-                .thenAwait( Duration.ofSeconds( 2 ) )
+        Flux<Integer> retriedSource = Flux.from( retryLogic.retryRx( source ) );
+        StepVerifier.create( retriedSource )
                 .expectNext( 0, 1 ) // first run
                 .expectNext( 0, 1, 0, 1, 0, 1, 0, 1 ) // 4 retry attempts
                 .verifyErrorSatisfies( e -> assertThat( e, equalTo( exception ) ) );
 
-        assertThat( elapsedList.size(), equalTo( 5 ) );
-        assertThat( elapsedList.get( 0 ), equalTo( 0L ) );
-        assertThat( elapsedList.get( 1 ), allOf( greaterThanOrEqualTo( 90L ), lessThanOrEqualTo( 110L ) ) );
-        assertThat( elapsedList.get( 2 ), allOf( greaterThanOrEqualTo( 180L ), lessThanOrEqualTo( 220L ) ) );
-        assertThat( elapsedList.get( 3 ), allOf( greaterThanOrEqualTo( 260L ), lessThanOrEqualTo( 440L ) ) );
-        assertThat( elapsedList.get( 4 ), allOf( greaterThanOrEqualTo( 720L ), lessThanOrEqualTo( 880L ) ) );
+        List<Long> delays = eventExecutor.scheduleDelays();
+        assertThat( delays.size(), equalTo( 4 ) );
+        assertThat( delays.get( 0 ), allOf( greaterThanOrEqualTo( 90L ), lessThanOrEqualTo( 110L ) ) );
+        assertThat( delays.get( 1 ), allOf( greaterThanOrEqualTo( 180L ), lessThanOrEqualTo( 220L ) ) );
+        assertThat( delays.get( 2 ), allOf( greaterThanOrEqualTo( 260L ), lessThanOrEqualTo( 440L ) ) );
+        assertThat( delays.get( 3 ), allOf( greaterThanOrEqualTo( 720L ), lessThanOrEqualTo( 880L ) ) );
     }
 
     private static void retry( ExponentialBackoffRetryLogic retryLogic, final int times )

driver/src/test/java/org/neo4j/driver/internal/util/ImmediateSchedulingEventExecutor.java

@@ -180,7 +180,9 @@ public ScheduledFuture<?> schedule( Runnable command, long delay, TimeUnit unit
     @Override
     public <V> ScheduledFuture<V> schedule( Callable<V> callable, long delay, TimeUnit unit )
     {
-        throw new UnsupportedOperationException();
+        scheduleDelays.add( unit.toMillis( delay ) );
+        delegate.submit( callable );
+        return mock( ScheduledFuture.class );
     }
 
     @Override