Merge branch 'master' into ci-coverage

negokaz · web-flow · commit 31fc4ccd2537 · 2022-03-22T10:49:47.000+09:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -18,7 +18,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   This feature is enabled only by using `typed.ClusterReplication`.
   It is highly recommended that you switch using the typed API since the classic API was deprecated.
 
-- Raft actors track the progress of the event sourcing [#136](https://github.com/lerna-stack/akka-entity-replication/issues/136).
+- Raft actors track the progress of the event sourcing
+  [#136](https://github.com/lerna-stack/akka-entity-replication/issues/136),
+  [PR#137](https://github.com/lerna-stack/akka-entity-replication/pull/137),
+  [PR#142](https://github.com/lerna-stack/akka-entity-replication/pull/142).
 
   This feature ensures that
   - Event Sourcing won't halt even if the event-sourcing store is unavailable for a long period.
@@ -36,8 +39,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
   It requires that
   `lerna.akka.entityreplication.raft.compaction.preserve-log-size` is less than
-  `lerna.akka.entityreplication.raft.compaction.log-size-threshold`. 
+  `lerna.akka.entityreplication.raft.compaction.log-size-threshold`.
 
+- Compaction warns if it might not delete enough entries [PR#142](https://github.com/lerna-stack/akka-entity-replication/pull/142)
 
 ### Changed
 - Bump up Akka version to 2.6.17 [PR#98](https://github.com/lerna-stack/akka-entity-replication/pull/98)
diff --git a/src/main/scala/lerna/akka/entityreplication/raft/RaftActor.scala b/src/main/scala/lerna/akka/entityreplication/raft/RaftActor.scala
@@ -465,31 +465,34 @@ private[raft] class RaftActor(
     }
     commitLogStore ! CommitLogStoreActor.AppendCommittedEntries(shardId, Seq.empty)
 
-    if (
-      currentData.replicatedLog.entries.size >= settings.compactionLogSizeThreshold
-      && currentData.hasLogEntriesThatCanBeCompacted
-    ) {
+    if (currentData.replicatedLog.entries.size >= settings.compactionLogSizeThreshold) {
       val estimatedCompactedLogSize: Int =
         currentData.estimatedReplicatedLogSizeAfterCompaction(settings.compactionPreserveLogSize)
       if (estimatedCompactedLogSize >= settings.compactionLogSizeThreshold) {
+        // This warning might also happen when the first SnapshotTick expires if
+        //   * there are already enough entries (>= compactionLogSizeThreshold), and eventSourcingIndex is unknown yet.
+        //   * there is no leader (e.g., split vote), and no entries are applied.
+        //   * there is the leader, but only a few entries are applied (the replication is too slow).
         if (log.isWarningEnabled) {
           log.warning(
-            "[{}] Skipping compaction since compaction might not delete enough entries " +
-            "(even if this compaction continues, the remaining entries will trigger new compaction at the next tick). " +
+            "[{}] Compaction might not delete enough entries, but will continue to reduce log size as possible " +
+            "(even if this compaction continues, the remaining entries might trigger new compaction at the next tick). " +
             s"Estimated compacted log size is [{}] entries (lastApplied [{}], eventSourcingIndex [{}], preserveLogSize [${settings.compactionPreserveLogSize}]), " +
             s"however compaction.log-size-threshold is [${settings.compactionLogSizeThreshold}] entries. " +
-            "This warning happens if event sourcing is too slow or compaction is too fast.",
+            "This warning might happen if event sourcing is too slow or compaction is too fast (or too slow). " +
+            "If this warning continues, please consult settings related to event sourcing and compaction.",
             currentState,
             estimatedCompactedLogSize,
             currentData.lastApplied,
             currentData.eventSourcingIndex,
           )
         }
-      } else if (snapshotSynchronizationIsInProgress) {
+      }
+      if (snapshotSynchronizationIsInProgress) {
         // Snapshot updates during synchronizing snapshot will break consistency
         if (log.isInfoEnabled)
           log.info("Skipping compaction because snapshot synchronization is in progress")
-      } else {
+      } else if (currentData.hasAppliedLogEntries) {
         val (term, logEntryIndex, entityIds) = currentData.resolveSnapshotTargets()
         applyDomainEvent(SnapshottingStarted(term, logEntryIndex, entityIds)) { _ =>
           if (log.isInfoEnabled)
diff --git a/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala b/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala
@@ -397,7 +397,8 @@ private[entityreplication] trait RaftMemberData
     prevLogIndex == lastSnapshotStatus.targetSnapshotLastLogIndex
   }
 
-  def hasLogEntriesThatCanBeCompacted: Boolean = {
+  /** Returns true if [[replicatedLog]] has entries that have been already applied */
+  def hasAppliedLogEntries: Boolean = {
     replicatedLog.sliceEntriesFromHead(lastApplied).nonEmpty
   }
 
diff --git a/src/test/scala/lerna/akka/entityreplication/raft/RaftActorSpec.scala b/src/test/scala/lerna/akka/entityreplication/raft/RaftActorSpec.scala
@@ -75,7 +75,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
     "ログが追加された後にログの長さがしきい値を超えている場合はスナップショットがとられる" in {
       val snapshotStore       = TestProbe()
       val replicationActor    = TestProbe()
-      val commitLogStore      = TestProbe()
       val shardId             = createUniqueShardId()
       val followerMemberIndex = createUniqueMemberIndex()
       val follower = createRaftActor(
@@ -84,7 +83,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
         shardSnapshotStore = snapshotStore.ref,
         replicationActor = replicationActor.ref,
         settings = RaftSettings(raftConfig),
-        commitLogStore = commitLogStore.ref,
       )
 
       val leaderMemberIndex = createUniqueMemberIndex()
@@ -105,10 +103,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
         leaderCommit = applicableIndex,
       )
 
-      // To ensure compaction starts, CommitLogStore should handle AppendCommittedEntries.
-      commitLogStore.expectMsg(CommitLogStoreActor.AppendCommittedEntries(shardId, Seq.empty))
-      commitLogStore.reply(CommitLogStoreActor.AppendCommittedEntriesResponse(applicableIndex))
-
       val command =
         replicationActor.fishForSpecificMessage() {
           case msg: TakeSnapshot => msg
@@ -375,7 +369,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
     "prevent to start snapshot synchronization during compaction" in {
       val snapshotStore       = TestProbe()
       val replicationActor    = TestProbe()
-      val commitLogStore      = TestProbe()
       val shardId             = createUniqueShardId()
       val followerMemberIndex = createUniqueMemberIndex()
       val follower = createRaftActor(
@@ -384,7 +377,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
         shardSnapshotStore = snapshotStore.ref,
         replicationActor = replicationActor.ref,
         settings = RaftSettings(raftConfig),
-        commitLogStore = commitLogStore.ref,
       )
 
       val leaderMemberIndex = createUniqueMemberIndex()
@@ -413,10 +405,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
         leaderCommit = applicableIndex,
       )
 
-      // To ensure compaction starts, CommitLogStore should handle AppendCommittedEntries.
-      commitLogStore.expectMsg(CommitLogStoreActor.AppendCommittedEntries(shardId, Seq.empty))
-      commitLogStore.reply(CommitLogStoreActor.AppendCommittedEntriesResponse(applicableIndex))
-
       // wait for starting compaction
       val takeSnapshot =
         replicationActor.fishForSpecificMessage() {
@@ -440,7 +428,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
     "not persist snapshots that have already been persisted in the next compaction" in {
       val snapshotStore       = TestProbe()
       val replicationActor    = TestProbe()
-      val commitLogStore      = TestProbe()
       val shardId             = createUniqueShardId()
       val followerMemberIndex = createUniqueMemberIndex()
       val follower = createRaftActor(
@@ -449,7 +436,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
         shardSnapshotStore = snapshotStore.ref,
         replicationActor = replicationActor.ref,
         settings = RaftSettings(raftConfig),
-        commitLogStore = commitLogStore.ref,
       )
 
       val leaderMemberIndex = createUniqueMemberIndex()
@@ -470,10 +456,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
         leaderCommit = LogEntryIndex(4),
       )
 
-      // To ensure compaction starts, CommitLogStore should handle AppendCommittedEntries.
-      commitLogStore.expectMsg(CommitLogStoreActor.AppendCommittedEntries(shardId, Seq.empty))
-      commitLogStore.reply(CommitLogStoreActor.AppendCommittedEntriesResponse(LogEntryIndex(4)))
-
       replicationActor.fishForSpecificMessage() {
         case msg: TakeSnapshot =>
           msg.metadata.entityId should be(entityId1)
@@ -499,10 +481,6 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
         leaderCommit = LogEntryIndex(7),
       )
 
-      // To ensure compaction starts, CommitLogStore should handle AppendCommittedEntries.
-      commitLogStore.expectMsg(CommitLogStoreActor.AppendCommittedEntries(shardId, Seq.empty))
-      commitLogStore.reply(CommitLogStoreActor.AppendCommittedEntriesResponse(LogEntryIndex(7)))
-
       // the snapshot should be only for entity2
       replicationActor.fishForSpecificMessage() {
         case msg: TakeSnapshot =>
@@ -516,16 +494,18 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
       }
     }
 
-    "not start if the compaction might not delete enough log entries" in {
+    "warn and continue if the compaction might not delete enough log entries" in {
       val commitLogStore      = TestProbe()
       val shardId             = createUniqueShardId()
       val followerMemberIndex = createUniqueMemberIndex()
       val raftSettings        = RaftSettings(raftConfig)
+      val replicationActor    = TestProbe()
       val follower = createRaftActor(
         shardId = shardId,
         selfMemberIndex = followerMemberIndex,
         commitLogStore = commitLogStore.ref,
         settings = raftSettings,
+        replicationActor = replicationActor.ref,
       )
 
       // The compaction cannot delete any entries by setting eventSourcingIndex to 0
@@ -536,11 +516,12 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
       assume(raftSettings.compactionLogSizeThreshold == 3)
       LoggingTestKit
         .warn(
-          "[Follower] Skipping compaction since compaction might not delete enough entries " +
-          "(even if this compaction continues, the remaining entries will trigger new compaction at the next tick). " +
+          "[Follower] Compaction might not delete enough entries, but will continue to reduce log size as possible " +
+          "(even if this compaction continues, the remaining entries might trigger new compaction at the next tick). " +
           "Estimated compacted log size is [3] entries (lastApplied [3], eventSourcingIndex [Some(0)], preserveLogSize [1]), " +
           "however compaction.log-size-threshold is [3] entries. " +
-          "This warning happens if event sourcing is too slow or compaction is too fast.",
+          "This warning might happen if event sourcing is too slow or compaction is too fast (or too slow). " +
+          "If this warning continues, please consult settings related to event sourcing and compaction.",
         ).expect {
           val leaderMemberIndex = createUniqueMemberIndex()
           val entityId          = NormalizedEntityId.from("entity1")
@@ -556,11 +537,12 @@ class RaftActorSpec extends TestKit(ActorSystem()) with RaftActorSpecBase {
             entries = logEntries,
             leaderCommit = LogEntryIndex(3),
           )
-          // A few seconds later (`compaction.log-size-check-interval`),
-          // the follower will output a warn log and not start the compaction
-          // since the estimated compacted entries size is larger than the threshold (`compaction.log-size-threshold`).
-        }
 
+          val takeSnapshot = replicationActor.fishForSpecificMessage() {
+            case msg: TakeSnapshot => msg
+          }
+          takeSnapshot.metadata shouldBe EntitySnapshotMetadata(entityId, LogEntryIndex(3))
+        }
     }
 
   }

Original file line number	Diff line number	Diff line change
`@@ -397,7 +397,8 @@ private[entityreplication] trait RaftMemberData`
`397`	`397`	`prevLogIndex == lastSnapshotStatus.targetSnapshotLastLogIndex`
`398`	`398`	`}`
`399`	`399`
`400`		`- def hasLogEntriesThatCanBeCompacted: Boolean = {`
	`400`	`+ /** Returns true if [[replicatedLog]] has entries that have been already applied */`
	`401`	`+ def hasAppliedLogEntries: Boolean = {`
`401`	`402`	`replicatedLog.sliceEntriesFromHead(lastApplied).nonEmpty`
`402`	`403`	`}`
`403`	`404`