Use leaderListener logic in RaftReplicator

This commit keeps the previous logic where RaftReplicator would send
to last known leader. But, instead, it uses the listener logic.

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

@@ -27,15 +27,14 @@
 import java.util.UUID;
 
 import org.neo4j.causalclustering.core.CausalClusteringSettings;
-import org.neo4j.causalclustering.core.consensus.ConsensusModule;
+import org.neo4j.causalclustering.core.consensus.RaftMachine;
 import org.neo4j.causalclustering.core.consensus.RaftMessages;
 import org.neo4j.causalclustering.core.replication.ProgressTrackerImpl;
 import org.neo4j.causalclustering.core.replication.RaftReplicator;
 import org.neo4j.causalclustering.core.replication.session.GlobalSession;
 import org.neo4j.causalclustering.core.replication.session.GlobalSessionTrackerState;
 import org.neo4j.causalclustering.core.replication.session.LocalSessionPool;
 import org.neo4j.causalclustering.core.state.storage.DurableStateStorage;
-import org.neo4j.causalclustering.helper.ConstantTimeTimeoutStrategy;
 import org.neo4j.causalclustering.helper.ExponentialBackoffStrategy;
 import org.neo4j.causalclustering.helper.TimeoutStrategy;
 import org.neo4j.causalclustering.identity.MemberId;
@@ -55,8 +54,8 @@ public class ReplicationModule
     private final ProgressTrackerImpl progressTracker;
     private final SessionTracker sessionTracker;
 
-    public ReplicationModule( MemberId myself, PlatformModule platformModule, Config config,
-            ConsensusModule consensusModule, Outbound<MemberId,RaftMessages.RaftMessage> outbound,
+    public ReplicationModule( RaftMachine raftMachine, MemberId myself, PlatformModule platformModule, Config config,
+            Outbound<MemberId,RaftMessages.RaftMessage> outbound,
             File clusterStateDirectory, FileSystemAbstraction fileSystem, LogProvider logProvider, AvailabilityGuard globalAvailabilityGuard )
     {
         LifeSupport life = platformModule.life;
@@ -74,20 +73,15 @@ public ReplicationModule( MemberId myself, PlatformModule platformModule, Config
 
         Duration initialBackoff = config.get( CausalClusteringSettings.replication_retry_timeout_base );
         Duration upperBoundBackoff = config.get( CausalClusteringSettings.replication_retry_timeout_limit );
-        Duration leaderBackoff = config.get( CausalClusteringSettings.replication_leader_retry_timeout );
 
         TimeoutStrategy progressRetryStrategy = new ExponentialBackoffStrategy( initialBackoff, upperBoundBackoff );
-        TimeoutStrategy leaderRetryStrategy = new ConstantTimeTimeoutStrategy( leaderBackoff );
         long availabilityTimeoutMillis = config.get( CausalClusteringSettings.replication_retry_timeout_base ).toMillis();
         replicator = new RaftReplicator(
-                consensusModule.raftMachine(),
+                raftMachine,
                 myself,
                 outbound,
                 sessionPool,
-                progressTracker,
-                progressRetryStrategy,
-                leaderRetryStrategy,
-                availabilityTimeoutMillis,
+                progressTracker, progressRetryStrategy, availabilityTimeoutMillis,
                 globalAvailabilityGuard, logProvider,
                 platformModule.monitors );
     }

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

@@ -350,10 +350,6 @@ public HostnameResolver getHostnameResolver( LogProvider logProvider, LogProvide
     public static final Setting<Duration> replication_retry_timeout_limit =
             setting( "causal_clustering.replication_retry_timeout_limit", DURATION, "60s" );
 
-    @Description( "The retry timeout for finding a leader for replication. Relevant during leader elections." )
-    public static final Setting<Duration> replication_leader_retry_timeout =
-            setting( "causal_clustering.replication_leader", DURATION, "500ms" );
-
     @Description( "The number of operations to be processed before the state machines flush to disk" )
     public static final Setting<Integer> state_machine_flush_window_size =
             setting( "causal_clustering.state_machine_flush_window_size", INTEGER, "4096" );

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

@@ -297,9 +297,8 @@ public EnterpriseCoreEditionModule( final PlatformModule platformModule,
 
         dependencies.satisfyDependency( consensusModule.raftMachine() );
 
-        replicationModule =
-                new ReplicationModule( identityModule.myself(), platformModule, config, consensusModule,
-                loggingOutbound, clusterStateDirectory.get(), fileSystem, logProvider, globalGuard );
+        replicationModule = new ReplicationModule( consensusModule.raftMachine(), identityModule.myself(), platformModule, config, loggingOutbound,
+                clusterStateDirectory.get(), fileSystem, logProvider, globalGuard );
 
         coreStateMachinesModule = new CoreStateMachinesModule( identityModule.myself(),
                 platformModule, clusterStateDirectory.get(), config, replicationModule.getReplicator(),

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/replication/RaftReplicator.java

@@ -28,7 +28,6 @@
 import org.neo4j.causalclustering.core.consensus.LeaderInfo;
 import org.neo4j.causalclustering.core.consensus.LeaderListener;
 import org.neo4j.causalclustering.core.consensus.LeaderLocator;
-import org.neo4j.causalclustering.core.consensus.NoLeaderFoundException;
 import org.neo4j.causalclustering.core.consensus.RaftMessages;
 import org.neo4j.causalclustering.core.replication.monitoring.ReplicationMonitor;
 import org.neo4j.causalclustering.core.replication.session.LocalSessionPool;
@@ -53,46 +52,38 @@ public class RaftReplicator implements Replicator, LeaderListener
     private final LocalSessionPool sessionPool;
     private final TimeoutStrategy progressTimeoutStrategy;
     private final AvailabilityGuard availabilityGuard;
-    private final LeaderLocator leaderLocator;
-    private final TimeoutStrategy leaderTimeoutStrategy;
     private final Log log;
     private final ReplicationMonitor replicationMonitor;
     private final long availabilityTimeoutMillis;
+    private volatile MemberId lastKnownLeader;
 
     public RaftReplicator( LeaderLocator leaderLocator, MemberId me, Outbound<MemberId,RaftMessages.RaftMessage> outbound, LocalSessionPool sessionPool,
-            ProgressTracker progressTracker, TimeoutStrategy progressTimeoutStrategy, TimeoutStrategy leaderTimeoutStrategy, long availabilityTimeoutMillis,
-            AvailabilityGuard availabilityGuard, LogProvider logProvider, Monitors monitors )
+            ProgressTracker progressTracker, TimeoutStrategy progressTimeoutStrategy, long availabilityTimeoutMillis, AvailabilityGuard availabilityGuard,
+            LogProvider logProvider, Monitors monitors )
     {
         this.me = me;
         this.outbound = outbound;
         this.progressTracker = progressTracker;
         this.sessionPool = sessionPool;
         this.progressTimeoutStrategy = progressTimeoutStrategy;
-        this.leaderTimeoutStrategy = leaderTimeoutStrategy;
         this.availabilityTimeoutMillis = availabilityTimeoutMillis;
         this.availabilityGuard = availabilityGuard;
-        this.leaderLocator = leaderLocator;
-        leaderLocator.registerListener( this );
-        log = logProvider.getLog( getClass() );
+        this.log = logProvider.getLog( getClass() );
         this.replicationMonitor = monitors.newMonitor( ReplicationMonitor.class );
+        leaderLocator.registerListener( this );
     }
 
     @Override
     public Future<Object> replicate( ReplicatedContent command, boolean trackResult ) throws ReplicationFailureException
     {
-        MemberId originalLeader;
-        try
-        {
-            originalLeader = leaderLocator.getLeader();
-        }
-        catch ( NoLeaderFoundException e )
+        if ( lastKnownLeader == null )
         {
-            throw new ReplicationFailureException( "Replication aborted since no leader was available", e );
+            throw new ReplicationFailureException( "Replication aborted since no leader was available" );
         }
-        return replicate0( command, trackResult, originalLeader );
+        return replicate0( command, trackResult );
     }
 
-    private Future<Object> replicate0( ReplicatedContent command, boolean trackResult, MemberId leader ) throws ReplicationFailureException
+    private Future<Object> replicate0( ReplicatedContent command, boolean trackResult ) throws ReplicationFailureException
     {
         replicationMonitor.startReplication();
         try
@@ -103,7 +94,6 @@ private Future<Object> replicate0( ReplicatedContent command, boolean trackResul
             Progress progress = progressTracker.start( operation );
 
             TimeoutStrategy.Timeout progressTimeout = progressTimeoutStrategy.newTimeout();
-            TimeoutStrategy.Timeout leaderTimeout = leaderTimeoutStrategy.newTimeout();
             int attempts = 0;
             try
             {
@@ -116,24 +106,14 @@ private Future<Object> replicate0( ReplicatedContent command, boolean trackResul
                     }
                     replicationMonitor.replicationAttempt();
                     assertDatabaseAvailable();
-                    try
+                    // blocking at least until the send has succeeded or failed before retrying
+                    outbound.send( lastKnownLeader, new RaftMessages.NewEntry.Request( me, operation ), true );
+                    progress.awaitReplication( progressTimeout.getMillis() );
+                    if ( progress.isReplicated() )
                     {
-                        // blocking at least until the send has succeeded or failed before retrying
-                        outbound.send( leader, new RaftMessages.NewEntry.Request( me, operation ), true );
-                        progress.awaitReplication( progressTimeout.getMillis() );
-                        if ( progress.isReplicated() )
-                        {
-                            break;
-                        }
-                        progressTimeout.increment();
-                        leader = leaderLocator.getLeader();
-                    }
-                    catch ( NoLeaderFoundException e )
-                    {
-                        log.debug( "Could not replicate operation " + operation + " because no leader was found. Retrying.", e );
-                        Thread.sleep( leaderTimeout.getMillis() );
-                        leaderTimeout.increment();
+                        break;
                     }
+                    progressTimeout.increment();
                 }
             }
             catch ( InterruptedException e )
@@ -167,6 +147,10 @@ private Future<Object> replicate0( ReplicatedContent command, boolean trackResul
     public void onLeaderSwitch( LeaderInfo leaderInfo )
     {
         progressTracker.triggerReplicationEvent();
+        if ( leaderInfo.memberId() != null )
+        {
+            lastKnownLeader = leaderInfo.memberId();
+        }
     }
 
     private void assertDatabaseAvailable() throws ReplicationFailureException