[Java] Change role to leader after winning election so leadership can…

… be asserted during replay or replication.

(cherry picked from commit 796d050)

aeron-cluster/src/main/java/io/aeron/cluster/Election.java

@@ -157,6 +157,11 @@ boolean isLeaderStartup()
         return isLeaderStartup;
     }
 
+    int thisMemberId()
+    {
+        return thisMember.id();
+    }
+
     int doWork(final long nowNs)
     {
         int workCount = 0;
@@ -309,17 +314,9 @@ void onCanvassPosition(
 
             if (logLeadershipTermId < this.leadershipTermId)
             {
-                switch (state)
+                if (Cluster.Role.LEADER == consensusModuleAgent.role())
                 {
-                    case LEADER_LOG_REPLICATION:
-                    case LEADER_INIT:
-                    case LEADER_READY:
-                    case LEADER_REPLAY:
-                        publishNewLeadershipTerm(follower, logLeadershipTermId, ctx.clusterClock().time());
-                        break;
-
-                    default:
-                        break;
+                    publishNewLeadershipTerm(follower, logLeadershipTermId, ctx.clusterClock().time());
                 }
             }
             else if (logLeadershipTermId > this.leadershipTermId)
@@ -663,15 +660,6 @@ private int init(final long nowNs)
         return 1;
     }
 
-    private void prepareForNewLeadership(final long nowNs)
-    {
-        final long lastAppendPosition = consensusModuleAgent.prepareForNewLeadership(logPosition, nowNs);
-        if (NULL_POSITION != lastAppendPosition)
-        {
-            appendPosition = lastAppendPosition;
-        }
-    }
-
     private int canvass(final long nowNs)
     {
         int workCount = 0;
@@ -790,14 +778,6 @@ private int followerBallot(final long nowNs)
         return workCount;
     }
 
-    /**
-     * Leader log replication must wait until we have consensus on the leaders append position. However,
-     * we want to be careful about updating the commit position as this will cause the clustered service to progress
-     * forward to early.
-     *
-     * @param nowNs current time
-     * @return work done
-     */
     private int leaderLogReplication(final long nowNs)
     {
         int workCount = 0;
@@ -917,7 +897,7 @@ private int followerLogReplication(final long nowNs)
                 if (replicationCommitPosition >= appendPosition)
                 {
                     ConsensusModuleAgent.logReplicationEnded(
-                        thisMemberId(),
+                        thisMember.id(),
                         "ELECTION",
                         logReplication.srcArchiveChannel(),
                         logReplication.recordingId(),
@@ -1313,11 +1293,8 @@ private void state(final ElectionState newState, final long nowNs)
                     break;
 
                 case LEADER_LOG_REPLICATION:
-                    logSessionId = consensusModuleAgent.addLogPublication(appendPosition);
-                    break;
-
-                case LEADER_INIT:
                     consensusModuleAgent.role(Cluster.Role.LEADER);
+                    logSessionId = consensusModuleAgent.addLogPublication(appendPosition);
                     break;
 
                 case FOLLOWER_LOG_REPLICATION:
@@ -1486,9 +1463,13 @@ private void logStateChange(
          */
     }
 
-    int thisMemberId()
+    private void prepareForNewLeadership(final long nowNs)
     {
-        return thisMember.id();
+        final long lastAppendPosition = consensusModuleAgent.prepareForNewLeadership(logPosition, nowNs);
+        if (NULL_POSITION != lastAppendPosition)
+        {
+            appendPosition = lastAppendPosition;
+        }
     }
 
     public String toString()