Merge branch 'master' into ci-coverage

Author: negokaz

src/multi-jvm/scala/lerna/akka/entityreplication/typed/MultiSnapshotSyncSpec.scala

@@ -154,6 +154,9 @@ class MultiSnapshotSyncSpec extends MultiNodeSpec(MultiSnapshotSyncSpecConfig) w
   /** RaftActor should compact it's log within this timeout */
   private val compactionTimeout: FiniteDuration = 10.seconds
 
+  /** Delayed Followers should synchronize snapshot within this timeout */
+  private val snapshotSynchronizationTimeout: FiniteDuration = 10.seconds
+
   /** None if this node doesn't shut it down. */
   private var maybeNewSystem: Option[classic.ActorSystem] = None
   private var runningNodes: Seq[RoleName]                 = roles
@@ -241,6 +244,14 @@ class MultiSnapshotSyncSpec extends MultiNodeSpec(MultiSnapshotSyncSpecConfig) w
     Thread.sleep(compactionTimeout.toMillis)
   }
 
+  /** Verifies snapshot synchronization completed by inspecting logging
+    */
+  private def expectSnapshotSynchronizationCompleted(): Unit = {
+    LoggingTestKit.info("Snapshot synchronization completed").expect {
+      Thread.sleep(snapshotSynchronizationTimeout.toMillis)
+    }
+  }
+
   private def setValue(id: String, value: Int)(implicit timeout: Timeout): Int = {
     // NOTE:
     // Too short retryInterval (e.g.200ms) will cause premature compaction and make this test unstable
@@ -286,6 +297,13 @@ class MultiSnapshotSyncSpec extends MultiNodeSpec(MultiSnapshotSyncSpecConfig) w
     enterBarrier("The cluster has nodes: [4,_,5].")
   }
 
+  "The leader (which is on node4) installs snapshots to the delayed follower (node5)" in {
+    runOn(node5) {
+      expectSnapshotSynchronizationCompleted()
+    }
+    enterBarrier("The snapshots were installed")
+  }
+
   "The leader (which is on node4) replicates some log entries. Nodes ([4,5]) compacts their log entries" in {
     object BarrierNames {
       val ExpectingCompactionCompleted = "Expecting compaction completed"
@@ -325,6 +343,13 @@ class MultiSnapshotSyncSpec extends MultiNodeSpec(MultiSnapshotSyncSpecConfig) w
     enterBarrier("The cluster has nodes: [6,7,8].")
   }
 
+  "The leader (which is on node8) installs snapshots to the delayed follower (node7)" in {
+    runOn(node7) {
+      expectSnapshotSynchronizationCompleted()
+    }
+    enterBarrier("The snapshots were installed")
+  }
+
   "The leader (which is on node8) replicates some log entries" in {
     object BarrierNames {
       val ExpectingCompactionCompleted = "Expecting compaction completed"

src/test/scala/lerna/akka/entityreplication/raft/RaftActorCandidateSpec.scala

@@ -309,7 +309,10 @@ class RaftActorCandidateSpec extends TestKit(ActorSystem()) with RaftActorSpecBa
       candidate ! createAppendEntries(shardId, term2, anotherMemberIndex)
       expectMsg(AppendEntriesSucceeded(term2, LogEntryIndex(0), candidateMemberIndex))
 
-      getState(candidate).stateName should be(Follower)
+      val state = getState(candidate)
+      state.stateName should be(Follower)
+      state.stateData.currentTerm should be(term2)
+      state.stateData.leaderMember should contain(anotherMemberIndex)
     }
 
     "AppendEntries が同じ Term を持っているときは（先に別のリーダーが当選したということなので）Follower に降格" in {
@@ -324,7 +327,9 @@ class RaftActorCandidateSpec extends TestKit(ActorSystem()) with RaftActorSpecBa
       candidate ! createAppendEntries(shardId, term, anotherMemberIndex)
       expectMsg(AppendEntriesSucceeded(term, LogEntryIndex(0), candidateMemberIndex))
 
-      getState(candidate).stateName should be(Follower)
+      val state = getState(candidate)
+      state.stateName should be(Follower)
+      state.stateData.leaderMember should contain(anotherMemberIndex)
     }
 
     "コマンドは保留し、フォロワーに降格した場合はリーダーに転送する" in {
@@ -440,6 +445,10 @@ class RaftActorCandidateSpec extends TestKit(ActorSystem()) with RaftActorSpecBa
       setState(candidate, Candidate, candidateData)
       candidate ! createAppendEntries(shardId, term1, leaderMemberIndex, index3, term1)
       expectMsg(AppendEntriesFailed(term1, candidateMemberIndex))
+
+      val state = getState(candidate)
+      state.stateName should be(Follower)
+      state.stateData.leaderMember should contain(leaderMemberIndex)
     }
 
     "prevLogIndex の Term が prevLogTerm に一致するログエントリでない場合は AppendEntriesFailed を返す" in {
@@ -497,6 +506,7 @@ class RaftActorCandidateSpec extends TestKit(ActorSystem()) with RaftActorSpecBa
       val state = getState(candidate)
       state.stateName should be(Follower)
       state.stateData.currentTerm should be(leaderTerm)
+      state.stateData.leaderMember should contain(leaderMemberIndex)
     }
 
     "leaderCommit > commitIndex となる場合、 commitIndex に min(leaderCommit, 新規エントリの最後のインデックス) を設定" in {

src/test/scala/lerna/akka/entityreplication/raft/RaftActorFollowerSpec.scala

@@ -207,6 +207,10 @@ class RaftActorFollowerSpec extends TestKit(ActorSystem()) with RaftActorSpecBas
       val leaderMemberIndex = candidateMemberIndex
       follower ! createAppendEntries(shardId, term, leaderMemberIndex)
       expectMsg(AppendEntriesSucceeded(term, LogEntryIndex(0), followerMemberIndex))
+
+      val state = getState(follower)
+      state.stateName should be(Follower)
+      state.stateData.leaderMember should contain(leaderMemberIndex)
     }
 
     "自分が持っている Term より新しい場合は AppendEntries を成功させる" in {
@@ -227,6 +231,11 @@ class RaftActorFollowerSpec extends TestKit(ActorSystem()) with RaftActorSpecBas
       val leaderMemberIndex = candidateMemberIndex
       follower ! createAppendEntries(shardId, term2, leaderMemberIndex)
       expectMsg(AppendEntriesSucceeded(term2, LogEntryIndex(0), followerMemberIndex))
+
+      val state = getState(follower)
+      state.stateName should be(Follower)
+      state.stateData.currentTerm should be(term2)
+      state.stateData.leaderMember should contain(leaderMemberIndex)
     }
 
     "自分が持っている Term より古い場合は AppendEntries を失敗させる" in {
@@ -494,6 +503,10 @@ class RaftActorFollowerSpec extends TestKit(ActorSystem()) with RaftActorSpecBas
         logEntries,
       )
       expectMsg(AppendEntriesFailed(term1, followerMemberIndex))
+
+      val state = getState(follower)
+      state.stateName should be(Follower)
+      state.stateData.leaderMember should contain(leaderMemberIndex)
     }
 
     "agree to a term if it receives AppendEntries which includes log entries that cannot be merged and newer Term" in {
@@ -598,6 +611,10 @@ class RaftActorFollowerSpec extends TestKit(ActorSystem()) with RaftActorSpecBas
         followerLogEntries.last.term.next(),
       )
       expectMsg(AppendEntriesFailed(term1, followerMemberIndex))
+
+      val state = getState(follower)
+      state.stateName should be(Follower)
+      state.stateData.leaderMember should contain(leaderMemberIndex)
     }
 
   }