Address review feedback

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

@@ -32,23 +32,23 @@ public class LeaderInfo implements Serializable
 
     private final MemberId memberId;
     private final long term;
-    private boolean isStepDown;
+    private boolean isSteppingDown;
 
     public LeaderInfo( MemberId memberId, long term )
     {
-        this.memberId = memberId;
-        this.term = term;
-        this.isStepDown = false;
+        this( memberId, term, false );
     }
 
-    public void stepDown()
+    public LeaderInfo( MemberId memberId, long term, boolean isSteppingDown )
     {
-        this.isStepDown = true;
+        this.memberId = memberId;
+        this.term = term;
+        this.isSteppingDown = isSteppingDown;
     }
 
-    public boolean isStepDown()
+    public boolean isSteppingDown()
     {
-        return isStepDown;
+        return isSteppingDown;
     }
 
     public MemberId memberId()

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

@@ -213,17 +213,11 @@ public synchronized ExposedRaftState state()
 
     private void notifyLeaderChanges( Outcome outcome )
     {
-        LeaderInfo leaderInfo = new LeaderInfo( outcome.getLeader(), outcome.getTerm() );
-        boolean isStepDown = outcome.isSteppingDown();
-
-        if ( isStepDown )
-        {
-            leaderInfo.stepDown();
-        }
+        LeaderInfo leaderInfo = new LeaderInfo( outcome.getLeader(), outcome.getTerm(), outcome.isSteppingDown() );
 
         for ( LeaderListener listener : leaderListeners )
         {
-            if ( isStepDown )
+            if ( outcome.isSteppingDown() )
             {
                 listener.onLeaderStepDown( leaderInfo );
             }
@@ -251,10 +245,7 @@ else if ( outcome.isSteppingDown() )
 
     private boolean leaderChanged( Outcome outcome, MemberId oldLeader )
     {
-        Optional<MemberId> oldLeaderOpt = Optional.ofNullable( oldLeader );
-        Optional<MemberId> outcomeLeaderOpt = Optional.ofNullable( outcome.getLeader() );
-
-        return !oldLeaderOpt.equals( outcomeLeaderOpt );
+        return !Objects.equals( oldLeader, outcome.getLeader() );
     }
 
     public synchronized ConsensusOutcome handle( RaftMessages.RaftMessage incomingMessage ) throws IOException

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

@@ -56,10 +56,10 @@ public interface CoreTopologyService extends TopologyService
      * Set the leader memberId to null for a given database (i.e. Raft consensus group).
      * This is intended to trigger state cleanup for informational procedures like {@link ClusterOverviewProcedure}
      *
-     * @param leaderInfo
-     * @param dbName
+     * @param stepDownLeaderInfo Information about the stepdown event, including term
+     * @param dbName The database for which this topology member should handle a stepdown
      */
-    void handleStepDown( LeaderInfo leaderInfo, String dbName );
+    void handleStepDown( LeaderInfo stepDownLeaderInfo, String dbName );
 
     interface Listener
     {

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

@@ -216,18 +216,25 @@ static void casLeaders( HazelcastInstance hazelcastInstance, LeaderInfo leaderIn
     {
         IAtomicReference<LeaderInfo> leaderRef = hazelcastInstance.getAtomicReference( DB_NAME_LEADER_TERM_PREFIX + dbName );
 
-        LeaderInfo expected = leaderRef.get();
+        LeaderInfo current = leaderRef.get();
+        Optional<LeaderInfo> currentOpt = Optional.ofNullable( current );
 
-        boolean noUpdate = Optional.ofNullable( expected ).map( LeaderInfo::memberId ).equals( Optional.ofNullable( leaderInfo.memberId() ) );
+        boolean noUpdate =  currentOpt.map( LeaderInfo::memberId ).equals( Optional.ofNullable( leaderInfo.memberId() ) );
 
-        boolean greaterOrEqualTermExists = Optional.ofNullable( expected ).map(l -> l.term() >= leaderInfo.term() ).orElse( false );
+        int termComparison =  currentOpt.map( l -> Long.compare( l.term(), leaderInfo.term() ) ).orElse( -1 );
 
-        if ( (greaterOrEqualTermExists || noUpdate) && !leaderInfo.isStepDown() )
+        boolean greaterTermExists = termComparison > 0;
+
+        boolean invalidTerm = greaterTermExists || ( termComparison == 0 && !leaderInfo.isSteppingDown() );
+
+        boolean success = !( invalidTerm || noUpdate);
+
+        if ( !success )
         {
             return;
         }
 
-        leaderRef.compareAndSet( expected, leaderInfo );
+        leaderRef.compareAndSet( current, leaderInfo );
     }
 
     private static Optional<LeaderInfo> getLeaderForDBName( HazelcastInstance hazelcastInstance, String dbName )

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

@@ -71,7 +71,6 @@ public class HazelcastCoreTopologyService extends AbstractTopologyService implem
     private static final long HAZELCAST_IS_HEALTHY_TIMEOUT_MS = TimeUnit.MINUTES.toMillis( 10 );
     private static final int HAZELCAST_MIN_CLUSTER = 2;
 
-    private final int minimumConsensusSize;
     private final Config config;
     private final MemberId myself;
     private final Log log;
@@ -101,7 +100,6 @@ public HazelcastCoreTopologyService( Config config, MemberId myself, JobSchedule
             TopologyServiceRetryStrategy topologyServiceRetryStrategy )
     {
         this.config = config;
-        this.minimumConsensusSize = config.get( CausalClusteringSettings.minimum_core_cluster_size_at_runtime );
         this.myself = myself;
         this.listenerService = new CoreTopologyListenerService();
         this.log = logProvider.getLog( getClass() );
@@ -143,12 +141,12 @@ public void setLeader( LeaderInfo leaderInfo, String dbName )
     }
 
     @Override
-    public void handleStepDown( LeaderInfo leaderInfo, String dbName )
+    public void handleStepDown( LeaderInfo stepDownLeaderInfo, String dbName )
     {
         boolean wasPreviousLeader = myself.equals( this.leaderInfo.memberId() );
         if ( wasPreviousLeader )
         {
-            HazelcastClusterTopology.casLeaders( hazelcastInstance, leaderInfo, dbName );
+            HazelcastClusterTopology.casLeaders( hazelcastInstance, stepDownLeaderInfo, dbName );
         }
     }
 

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

@@ -395,7 +395,9 @@ public List<CoreClusterMember> getAllMembersWithAnyRole( String dbName, Role...
         Set<Role> roleSet = Arrays.stream( roles ).collect( toSet() );
 
         return coreMembers.values().stream()
-                .filter( m -> m.database() != null && m.dbName().equals( dbName ) &&  roleSet.contains( m.database().getRole() ) )
+                .filter( m -> m.database() != null )
+                .filter( m -> m.dbName().equals( dbName ) )
+                .filter( m -> roleSet.contains( m.database().getRole() ) )
                 .collect( Collectors.toList() );
     }
 

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

@@ -91,7 +91,7 @@ public Map<MemberId,RoleInfo> allCoreRoles()
     @Override
     public void setLeader( LeaderInfo newLeader, String dbName )
     {
-        if ( this.leaderInfo.term() < newLeader.term() )
+        if ( this.leaderInfo.term() < newLeader.term() && newLeader.memberId() != null )
         {
             this.leaderInfo = newLeader;
             sharedDiscoveryService.casLeaders( newLeader, localDBName );
@@ -143,12 +143,12 @@ public CoreTopology allCoreServers()
     }
 
     @Override
-    public void handleStepDown( LeaderInfo leaderInfo, String dbName )
+    public void handleStepDown( LeaderInfo stepDownLeaderInfo, String dbName )
     {
         boolean wasPreviousLeader = myself.equals( this.leaderInfo.memberId() );
         if ( wasPreviousLeader )
         {
-            sharedDiscoveryService.casLeaders( leaderInfo, dbName );
+            sharedDiscoveryService.casLeaders( stepDownLeaderInfo, dbName );
         }
     }
 

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

@@ -134,9 +134,13 @@ void casLeaders( LeaderInfo leaderInfo, String dbName )
 
             boolean noUpdate = current.map( LeaderInfo::memberId ).equals( Optional.ofNullable( leaderInfo.memberId() ) );
 
-            boolean greaterOrEqualTermExists = current.map( l -> l.term() >= leaderInfo.term() ).orElse( false );
+            int termComparison = current.map( l -> Long.compare( l.term(), leaderInfo.term() ) ).orElse( -1 );
 
-            boolean success = !(greaterOrEqualTermExists || noUpdate) || leaderInfo.isStepDown();
+            boolean greaterTermExists = termComparison > 0;
+
+            boolean invalidTerm = greaterTermExists || ( termComparison == 0 && !leaderInfo.isSteppingDown() );
+
+            boolean success = !( invalidTerm || noUpdate);
 
             if ( success )
             {

enterprise/causal-clustering/src/test/java/org/neo4j/causalclustering/scenarios/ClusterOverviewIT.java

@@ -277,7 +277,6 @@ public void shouldHaveNoLeaderIfMajorityCoreMembersDead() throws Exception
         clusterRule.withNumberOfCoreMembers( 2 );
 
         Cluster cluster = clusterRule.startCluster();
-        //TODO: address proliferation of Role enums
         List<CoreClusterMember> followers = cluster.getAllMembersWithRole( Role.FOLLOWER );
         CoreClusterMember leader = cluster.getMemberWithRole( Role.LEADER );
         followers.forEach( CoreClusterMember::shutdown );