Merge #4531 #4582

4531: Add tests for snapshotting and compaction r=Zelldon a=Zelldon

## Description

Improved further the `RaftRule` such that we are able to trigger snapshotting, which will create snapshots on all nodes. After the snapshot is taken the snapshot listeners are called, which trigger compaction. This setup reflects now how our system works, but it is much easier to test I think.

Normally we do a snapshot on the Leader and replicate that. The snapshot replication call at the end also `SnapshotStore#newSnapshot`. With the current test approach we were able to simplify that approach and test the same behavior. 

I added a test for:

 * normal snapshot taking and verify that all nodes have this snapshot
 * snapshotting triggers compaction verify log is compacted
 * snapshot is send on rejoin cluster

I plan to add further tests, more complex ones probably.

I checked #4467 and the problem was that the node was not able to truncate his log because it was at zero index and the leader was not able to send a snapshot, because it had none. The test was wrongly written I would say.

<!-- Please explain the changes you made here. -->

## Related issues

<!-- Which issues are closed by this PR or are related -->

closes #4467

#

4582: chore(deps): update version.sbe to v1.18.1 r=npepinpe a=renovate[bot]

This PR contains the following updates:

| Package | Update | Change |
|---|---|---|
| [uk.co.real-logic:sbe-tool](https://togithub.com/real-logic/simple-binary-encoding) | minor | `1.17.0` -> `1.18.1` |
| [uk.co.real-logic:sbe-all](https://togithub.com/real-logic/simple-binary-encoding) | minor | `1.17.0` -> `1.18.1` |

---

### Release Notes

<details>
<summary>real-logic/simple-binary-encoding</summary>

### [`v1.18.1`](https://togithub.com/real-logic/simple-binary-encoding/releases/1.18.1)

[Compare Source](https://togithub.com/real-logic/simple-binary-encoding/compare/1.18.0...1.18.1)

-   Fix case of importing buffers when var data is used in nested groups for Java codecs.

Java binaries can be found [here...](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22uk.co.real-logic%22%20sbe)

### [`v1.18.0`](https://togithub.com/real-logic/simple-binary-encoding/releases/1.18.0)

[Compare Source](https://togithub.com/real-logic/simple-binary-encoding/compare/1.17.0...1.18.0)

-   Only generate imports for Java codecs when required to address warnings.
-   Access fixed length arrays as Spans in C# codecs. [PR #&#8203;780](https://togithub.com/real-logic/simple-binary-encoding/pull/780).
-   Add `SbeSchemaId` and `SbeSchemaVersion` as constants in fixed flyweights for C# codecs.
-   Generate source docs from schema descriptions for C# codecs. [PR #&#8203;778](https://togithub.com/real-logic/simple-binary-encoding/pull/778).
-   Add offset and wrap methods to C# codecs for ease of use. [PR #&#8203;777](https://togithub.com/real-logic/simple-binary-encoding/pull/777).
-   Support non-standard message headers form the C++ codecs. [PR #&#8203;775](https://togithub.com/real-logic/simple-binary-encoding/pull/775).
-   Fix version support for enums in C codecs. [Issue #&#8203;773](https://togithub.com/real-logic/simple-binary-encoding/issues/773).
-   Improve formatting of generated C codecs.
-   Require a strict dependency on Agrona.
-   Upgrade to Agrona 1.5.0.
-   Upgrade to javadoc-links 5.1.0.
-   Upgrade to JUnit 5.6.2.
-   Upgrade to Gradle 6.4.1.

Java binaries can be found [here...](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22uk.co.real-logic%22%20sbe)

</details>

---

### Renovate configuration

:date: **Schedule**: At any time (no schedule defined).

:vertical_traffic_light: **Automerge**: Disabled by config. Please merge this manually once you are satisfied.

:recycle: **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox.

:no_bell: **Ignore**: Close this PR and you won't be reminded about these updates again.

---

 - [ ] <!-- rebase-check -->If you want to rebase/retry this PR, check this box

---

This PR has been generated by [WhiteSource Renovate](https://renovate.whitesourcesoftware.com). View repository job log [here](https://app.renovatebot.com/dashboard#zeebe-io/zeebe).

Co-authored-by: Christopher Zell <zelldon91@googlemail.com>
Co-authored-by: Renovate Bot <bot@renovateapp.com>

atomix/cluster/src/test/java/io/atomix/raft/RaftFailOverTest.java

@@ -16,12 +16,12 @@
 package io.atomix.raft;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import io.atomix.storage.journal.Indexed;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -118,19 +118,187 @@ public void shouldRecoverLeaderRestart() throws Throwable {
   }
 
   @Test
-  @Ignore("https://github.com/zeebe-io/zeebe/issues/4467")
-  public void testNodeCatchUpAfterCompaction() throws Exception {
+  public void shouldTakeSnapshot() throws Exception {
     // given
-    raftRule.shutdownServer("1");
+    raftRule.appendEntries(128);
+
+    // when
+    raftRule.doSnapshot(100);
+
+    // then
+    assertThat(raftRule.allNodesHaveSnapshotWithIndex(100)).isTrue();
+  }
+
+  @Test
+  public void shouldCompactLogOnSnapshot() throws Exception {
+    // given
+    raftRule.appendEntries(128);
+    final var memberLogs = raftRule.getMemberLogs();
+
+    // when
+    raftRule.doSnapshot(100);
+
+    // then
+    final var compactedLogs = raftRule.getMemberLogs();
+
+    assertThat(compactedLogs.isEmpty()).isFalse();
+    for (final String raftMember : compactedLogs.keySet()) {
+      final var compactedLog = compactedLogs.get(raftMember);
+      final var previousLog = memberLogs.get(raftMember);
+      assertThat(compactedLog.size()).isLessThan(previousLog.size());
+      assertThat(compactedLog).isSubsetOf(previousLog);
+    }
+  }
+
+  @Test
+  public void shouldReplicateSnapshotOnJoin() throws Exception {
+    // given
+    final var follower = raftRule.shutdownFollower();
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(100);
+    final var leaderSnapshot = raftRule.getSnapshotFromLeader();
+
+    // when
+    raftRule.joinCluster(follower);
+
+    // then
+    assertThat(raftRule.allNodesHaveSnapshotWithIndex(100)).isTrue();
+    final var snapshot = raftRule.getSnapshotOnNode(follower);
+
+    assertThat(snapshot.index()).isEqualTo(leaderSnapshot.index()).isEqualTo(100);
+    assertThat(snapshot.term()).isEqualTo(snapshot.term());
+  }
+
+  @Test
+  public void shouldReplicateEntriesAfterSnapshotOnJoin() throws Exception {
+    // given
+    final var follower = raftRule.shutdownFollower();
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(100);
+
+    // when
+    raftRule.joinCluster(follower);
+
+    // then
+    assertThat(raftRule.allNodesHaveSnapshotWithIndex(100)).isTrue();
+
+    final var memberLogs = raftRule.getMemberLogs();
+    final var entries = memberLogs.get(follower);
+    // entries after snapshot should be replicated
+    assertThat(entries.get(0).index()).isEqualTo(100 + 1);
+
+    for (final String member : memberLogs.keySet()) {
+      if (!follower.equals(member)) {
+        final var memberEntries = memberLogs.get(member);
+        assertThat(memberEntries).endsWith(entries.toArray(new Indexed[0]));
+      }
+    }
+  }
+
+  @Test
+  public void shouldNotJoinAfterDataLoss() throws Exception {
+    // given
+    final var follower = raftRule.shutdownFollower();
+
+    // when
+    raftRule.triggerDataLossOnNode(follower);
+
+    // then
+    // follower is not allowed to join the cluster, he needs to bootstrap
+    assertThatThrownBy(() -> raftRule.joinCluster(follower))
+        .hasCauseInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("not a member of the cluster");
+  }
+
+  @Test
+  public void shouldReplicateSnapshotAfterDataLoss() throws Exception {
+    // given
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(100);
+    final var follower = raftRule.shutdownFollower();
+    final var leaderSnapshot = raftRule.getSnapshotFromLeader();
+
+    // when
+    raftRule.triggerDataLossOnNode(follower);
+    raftRule.bootstrapNode(follower);
+
+    // then
+    assertThat(raftRule.allNodesHaveSnapshotWithIndex(100)).isTrue();
+    final var snapshot = raftRule.getSnapshotOnNode(follower);
+
+    assertThat(snapshot.index()).isEqualTo(leaderSnapshot.index()).isEqualTo(100);
+    assertThat(snapshot.term()).isEqualTo(snapshot.term());
+  }
+
+  @Test
+  public void shouldReplicateEntriesAfterSnapshotAfterDataLoss() throws Exception {
+    // given
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(100);
+    final var follower = raftRule.shutdownFollower();
+
+    // when
+    raftRule.triggerDataLossOnNode(follower);
+    raftRule.bootstrapNode(follower);
+
+    // then
+    assertThat(raftRule.allNodesHaveSnapshotWithIndex(100)).isTrue();
+    final var memberLogs = raftRule.getMemberLogs();
+    final var entries = memberLogs.get(follower);
+    // entries after snapshot should be replicated
+    assertThat(entries.get(0).index()).isEqualTo(100 + 1);
+
+    for (final String member : memberLogs.keySet()) {
+      if (!follower.equals(member)) {
+        final var memberEntries = memberLogs.get(member);
+        assertThat(memberEntries).endsWith(entries.toArray(new Indexed[0]));
+      }
+    }
+  }
+
+  @Test
+  public void shouldTakeMultipleSnapshotsAndReplicateSnapshotAfterRestart() throws Exception {
+    // given
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(100);
+    final var follower = raftRule.shutdownFollower();
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(200);
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(300);
+    final var leaderSnapshot = raftRule.getSnapshotFromLeader();
+
+    // when
+    raftRule.joinCluster(follower);
+
+    // then
+    assertThat(raftRule.allNodesHaveSnapshotWithIndex(300)).isTrue();
+    final var snapshot = raftRule.getSnapshotOnNode(follower);
+
+    assertThat(snapshot.index()).isEqualTo(leaderSnapshot.index()).isEqualTo(300);
+    assertThat(snapshot.term()).isEqualTo(snapshot.term());
+  }
+
+  @Test
+  public void shouldReplicateSnapshotToOldLeaderAfterRestart() throws Exception {
+    // given
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(100);
+    final var leader = raftRule.shutdownLeader();
     raftRule.awaitNewLeader();
-    raftRule.appendEntries(100);
-    raftRule.tryToCompactLogsOnServersExcept("1", 100).join();
+    raftRule.appendEntries(128);
+    raftRule.doSnapshot(200);
+    final var leaderSnapshot = raftRule.getSnapshotFromLeader();
 
     // when
-    final var future = raftRule.startServer("1");
+    raftRule.joinCluster(leader);
 
     // then
-    future.join();
+    assertThat(raftRule.allNodesHaveSnapshotWithIndex(200)).isTrue();
+    final var snapshot = raftRule.getSnapshotOnNode(leader);
+
+    assertThat(snapshot.index()).isEqualTo(leaderSnapshot.index()).isEqualTo(200);
+    assertThat(snapshot.term()).isEqualTo(snapshot.term());
   }
 
   private void assertMemberLogs(final Map<String, List<Indexed<?>>> memberLog) {