Fixed MultiRetryStrategyTest

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/state/ClusteringModule.java

@@ -65,7 +65,7 @@ public ClusteringModule( DiscoveryServiceFactory discoveryServiceFactory, Member
 
         topologyService = discoveryServiceFactory
                 .coreTopologyService( config, sslPolicy, myself, platformModule.jobScheduler, logProvider,
-                        userLogProvider, hostnameResolver, resolveStrategy( config ) );
+                        userLogProvider, hostnameResolver, resolveStrategy( config, logProvider ) );
 
         life.add( topologyService );
 
@@ -82,13 +82,13 @@ public ClusteringModule( DiscoveryServiceFactory discoveryServiceFactory, Member
                 () -> sleep( 100 ), 300_000, coreBootstrapper );
     }
 
-    private static TopologyServiceRetryStrategy resolveStrategy( Config config )
+    private static TopologyServiceRetryStrategy resolveStrategy( Config config, LogProvider logProvider )
     {
         long refreshPeriodMillis = config.get( CausalClusteringSettings.cluster_topology_refresh ).toMillis();
         int pollingFrequencyWithinRefreshWindow = 2;
         int numberOfRetries =
                 pollingFrequencyWithinRefreshWindow + 1; // we want to have more retries at the given frequency than there is time in a refresh period
-        return new TopologyServiceMultiRetryStrategy( refreshPeriodMillis / pollingFrequencyWithinRefreshWindow, numberOfRetries );
+        return new TopologyServiceMultiRetryStrategy( refreshPeriodMillis / pollingFrequencyWithinRefreshWindow, numberOfRetries, logProvider );
     }
 
     public CoreTopologyService topologyService()

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/discovery/HazelcastClient.java

@@ -83,15 +83,15 @@ class HazelcastClient extends LifecycleAdapter implements TopologyService
         this.refreshPeriod = config.get( CausalClusteringSettings.cluster_topology_refresh ).toMillis();
         this.myself = myself;
         this.groups = config.get( CausalClusteringSettings.server_groups );
-        this.topologyServiceRetryStrategy = resolveStrategy( refreshPeriod );
+        this.topologyServiceRetryStrategy = resolveStrategy( refreshPeriod, logProvider );
     }
 
-    private static TopologyServiceRetryStrategy resolveStrategy( long refreshPeriodMillis )
+    private static TopologyServiceRetryStrategy resolveStrategy( long refreshPeriodMillis, LogProvider logProvider )
     {
         int pollingFrequencyWithinRefreshWindow = 2;
         int numberOfRetries =
                 pollingFrequencyWithinRefreshWindow + 1; // we want to have more retries at the given frequency than there is time in a refresh period
-        return new TopologyServiceMultiRetryStrategy( refreshPeriodMillis / pollingFrequencyWithinRefreshWindow, numberOfRetries );
+        return new TopologyServiceMultiRetryStrategy( refreshPeriodMillis / pollingFrequencyWithinRefreshWindow, numberOfRetries, logProvider );
     }
 
     @Override

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/discovery/MultiRetryStrategy.java

@@ -19,69 +19,55 @@
  */
 package org.neo4j.causalclustering.discovery;
 
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
 import org.neo4j.logging.Log;
 import org.neo4j.logging.LogProvider;
-import org.neo4j.logging.NullLogProvider;
 
 /**
  * Implementation of the RetryStrategy that repeats the retriable function until the correct result has been retrieved or the limit of retries has been
  * encountered.
  * There is a fixed delay between each retry.
  *
- * @param <I> the type of input of the retriable function
- * @param <E> the type of output of the retriable function
+ * @param <INPUT> the type of input of the retriable function
+ * @param <OUTPUT> the type of output of the retriable function
  */
-public class MultiRetryStrategy<I, E> implements RetryStrategy<I,E>
+public class MultiRetryStrategy<INPUT, OUTPUT> implements RetryStrategy<INPUT,OUTPUT>
 {
     private final long delayInMillis;
     private final long retries;
     private final LogProvider logProvider;
-
-    /**
-     * @param delayInMillis number of milliseconds between each attempt at getting the desired result
-     * @param retries the number of attempts to perform before giving up
-     */
-    public MultiRetryStrategy( long delayInMillis, long retries )
-    {
-        this( delayInMillis, retries, NullLogProvider.getInstance() );
-    }
+    private final Consumer<Long> sleeper;
 
     /**
      * @param delayInMillis number of milliseconds between each attempt at getting the desired result
      * @param retries the number of attempts to perform before giving up
      * @param logProvider {@see LogProvider}
      */
-    public MultiRetryStrategy( long delayInMillis, long retries, LogProvider logProvider )
+    public MultiRetryStrategy( long delayInMillis, long retries, LogProvider logProvider, Consumer<Long> sleeper )
     {
         this.delayInMillis = delayInMillis;
         this.retries = retries;
         this.logProvider = logProvider;
+        this.sleeper = sleeper;
     }
 
     /**
      * {@inheritDoc}
      */
     @Override
-    public E apply( I input, Function<I,E> retriable, Predicate<E> wasRetrySuccessful )
+    public OUTPUT apply( INPUT retriableInput, Function<INPUT,OUTPUT> retriable, Predicate<OUTPUT> wasRetrySuccessful )
     {
         Log log = logProvider.getLog( MultiRetryStrategy.class );
-        E result = retriable.apply( input );
+        OUTPUT result = retriable.apply( retriableInput );
         int currentIteration = 0;
         while ( !wasRetrySuccessful.test( result ) && currentIteration++ < retries )
         {
-            log.debug( "Try attempt was unsuccessful for input: %s\n", input );
-            try
-            {
-                Thread.sleep( delayInMillis );
-            }
-            catch ( InterruptedException e )
-            {
-                throw new RuntimeException( e );
-            }
-            result = retriable.apply( input );
+            log.debug( "Try attempt was unsuccessful for input: %s\n", retriableInput );
+            sleeper.accept( delayInMillis );
+            result = retriable.apply( retriableInput );
         }
         return result;
     }

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/discovery/TopologyServiceMultiRetryStrategy.java

@@ -23,11 +23,22 @@
 
 import org.neo4j.causalclustering.identity.MemberId;
 import org.neo4j.helpers.AdvertisedSocketAddress;
+import org.neo4j.logging.LogProvider;
 
 public class TopologyServiceMultiRetryStrategy extends MultiRetryStrategy<MemberId,Optional<AdvertisedSocketAddress>> implements TopologyServiceRetryStrategy
 {
-    public TopologyServiceMultiRetryStrategy( long delayInMillis, long retries )
+    public TopologyServiceMultiRetryStrategy( long delayInMillis, long retries, LogProvider logProvider )
     {
-        super( delayInMillis, retries );
+        super( delayInMillis, retries, logProvider, millis ->
+        {
+            try
+            {
+                Thread.sleep( millis );
+            }
+            catch ( InterruptedException e )
+            {
+                throw new RuntimeException( e );
+            }
+        } );
     }
 }

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/readreplica/EnterpriseReadReplicaEditionModule.java

@@ -192,7 +192,7 @@ public class EnterpriseReadReplicaEditionModule extends EditionModule
 
         TopologyService topologyService = discoveryServiceFactory.topologyService( config, clusterSslPolicy,
                 logProvider, platformModule.jobScheduler, myself,
-                        hostnameResolver, resolveStrategy( config ) );
+                        hostnameResolver, resolveStrategy( config, logProvider ) );
 
         life.add( dependencies.satisfyDependency( topologyService ) );
 
@@ -383,12 +383,12 @@ public UpstreamDatabaseSelectionStrategy next()
         }
     }
 
-    private static TopologyServiceRetryStrategy resolveStrategy( Config config )
+    private static TopologyServiceRetryStrategy resolveStrategy( Config config, LogProvider logProvider )
     {
         long refreshPeriodMillis = config.get( CausalClusteringSettings.cluster_topology_refresh ).toMillis();
         int pollingFrequencyWithinRefreshWindow = 2;
         int numberOfRetries =
                 pollingFrequencyWithinRefreshWindow + 1; // we want to have more retries at the given frequency than there is time in a refresh period
-        return new TopologyServiceMultiRetryStrategy( refreshPeriodMillis / pollingFrequencyWithinRefreshWindow, numberOfRetries );
+        return new TopologyServiceMultiRetryStrategy( refreshPeriodMillis / pollingFrequencyWithinRefreshWindow, numberOfRetries, logProvider );
     }
 }

enterprise/causal-clustering/src/test/java/org/neo4j/causalclustering/discovery/MultiRetryStrategyTest.java

@@ -19,16 +19,17 @@
  */
 package org.neo4j.causalclustering.discovery;
 
-import org.junit.Ignore;
 import org.junit.Test;
 
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
+import org.neo4j.logging.NullLogProvider;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
-@Ignore
 public class MultiRetryStrategyTest
 {
     private static final Predicate<Integer> ALWAYS_VALID = i -> true;
@@ -52,9 +53,9 @@ public boolean test( Integer integer )
     public void successOnRetryCausesNoDelay()
     {
         // given
-        int delay = 1000;
+        CountingSleeper countingSleeper = new CountingSleeper();
         int retries = 10;
-        MultiRetryStrategy<Integer,Integer> subject = new MultiRetryStrategy<>( delay, retries );
+        MultiRetryStrategy<Integer,Integer> subject = new MultiRetryStrategy<>( 0, retries, NullLogProvider.getInstance(), countingSleeper );
 
         // when
         long startTime = System.currentTimeMillis();
@@ -63,49 +64,52 @@ public void successOnRetryCausesNoDelay()
 
         // then
         long duration = endTime - startTime;
-        assertTrue( "First execution should not be called after delay", duration < delay );
+        assertEquals( 0, countingSleeper.invocationCount() );
         assertEquals( "Function identity should be used to retrieve the expected value", 3, result.intValue() );
     }
 
     @Test
     public void numberOfIterationsDoesNotExceedMaximum()
     {
         // given
-        int delay = 200;
+        CountingSleeper countingSleeper = new CountingSleeper();
         int retries = 5;
-        MultiRetryStrategy<Integer,Integer> subject = new MultiRetryStrategy<>( delay, retries );
+        MultiRetryStrategy<Integer,Integer> subject = new MultiRetryStrategy<>( 0, retries, NullLogProvider.getInstance(), countingSleeper );
 
         // when
-        long startTime = System.currentTimeMillis();
-        Integer result = subject.apply( 3, Function.identity(), NEVER_VALID );
-        long endTime = System.currentTimeMillis();
+        subject.apply( 3, Function.identity(), NEVER_VALID );
 
         // then
-        long duration = endTime - startTime;
-        double durationInSeconds = duration / 1000.0;
-        double expectedDurationInSeconds = (delay * retries) / 1000.0;
-        double marginOfErrorInSeconds = (delay / 1000.0) / 2;
-        assertEquals( expectedDurationInSeconds, durationInSeconds, marginOfErrorInSeconds );
+        assertEquals( retries, countingSleeper.invocationCount() );
     }
 
     @Test
     public void successfulRetriesBreakTheRetryLoop()
     {
-        // given
-        int delay = 200;
-        int retries = 10;
-        MultiRetryStrategy<Integer,Integer> subject = new MultiRetryStrategy<>( delay, retries );
+        CountingSleeper countingSleeper = new CountingSleeper();
+        int retries = 5;
+        MultiRetryStrategy<Integer,Integer> subject = new MultiRetryStrategy<>( 0, retries, NullLogProvider.getInstance(), countingSleeper );
 
         // when
-        long startTime = System.currentTimeMillis();
-        Integer result = subject.apply( 3, Function.identity(), VALID_ON_SECOND_TIME );
-        long endTime = System.currentTimeMillis();
+        subject.apply( 3, Function.identity(), VALID_ON_SECOND_TIME );
 
         // then
-        long duration = endTime - startTime;
-        double durationInSeconds = duration / 1000.0;
-        double expectedDurationInSeconds = delay / 1000.0;
-        double marginOfErrorInSeconds = (delay / 1000.0) / 4;
-        assertEquals( expectedDurationInSeconds, durationInSeconds, marginOfErrorInSeconds );
+        assertEquals( 1, countingSleeper.invocationCount() );
+    }
+
+    private class CountingSleeper implements Consumer<Long>
+    {
+        private int counter;
+
+        @Override
+        public void accept( Long duration )
+        {
+            counter++;
+        }
+
+        public int invocationCount()
+        {
+            return counter;
+        }
     }
 }