Rewrite Snapshotting

[Zelldon]

Hey Team,

sorry for the big PR but was saw no better way to do this. I think most of the new lines are actually unit tests.

@pihme I think it is just for you to learn how it works don't need to be to check every details.

@npepinpe or @deepthidevaki who ever has time would be nice if you could take a look, especially for raft and zeebe endpoints where we use it. Hope it makes sense to you.

Happy to get your feedback.

Description

Introduces new interfaces for Snapshotting and rewrites and merge existing implementation.

PersistedSnapshotStore can be used to take snapshots and receive
snapshot chunks. After all chunks are received or the snapshot is taken
the volatile snapshot can be persisted. It is always possible
to abort an snapshot, before persisting. The store can also be used
to get the latest snapshot which was persisted. The store allows
to register listeners, which are called on new persisted snapshots.

On receiving snapshot chunks and consuming them, checksums of the chunks
are verified. On persisting an received snapshot the complete snapshot
checksum is verified.

An persisted snapshot can be read from its persisted storage via a
reader and will be chunked in several snapshot chunks, which
then can be transmitted and consumed again by the store.

The StateSnapshotController use by the AsyncSnapshotDirector and is the
only one which takes a snapshot. The ReplicationController is
registered at the SnapshotStore as listener to replicate new persisted
snapshots. The replication will send all chunks over the network.

The ReplicationController will receive the chunks on the follower side and will use
the newReceviedSnapshot method of the store to consume the SnapshotChunks.

On the LeaderRole (in the AbstractAppender) the last snapshot can also
be replicated. This is done via the SnapshotChunkReader, which
reads the chunks from the persisted store.

The Follower (in PassiveRole) will consume the replicated chunks and
will store it similar to the ReplicationController.

On both ReplicationController and PassiveRole the checksum of the
snapshot chunks but also of the complete snapshot is verified before
the received snapshot is persisted.

Related work:

• Most snapshot implementation details are now in Atomix
• Zeebe related snapshoting like replication and taking of snapshot is
  part of Broker
• Use SBE for snapshot chunk replication/consumption
• StreamProcessorTests moved again to engine module - removed snapshot
  related tests, dont need to be part of that
• clean up management schema file in broker
• Wrote a bunch of new unit tests for snapshot implementation

Class Diagram

Generate Class diagram via Intellij:

Related issues

closes #4424
closes #4604
closes #4345
closes #4122
closes #4067
closes #3214
closes #2721
closes #4425

Pull Request Checklist

• All commit messages match our commit message guidelines
• The submitting code follows our code style
• If submitting code, please run mvn clean install -DskipTests locally before committing

[pihme]

Lots of good stuff. I like the unit tests. Several smaller comments.

The important feedback is for the state controller, where I think there is a memory leak and the metrics aren't updated correctly in some cases.

[deepthidevaki]

Here are some initial comments.

[deepthidevaki]

Here are some more minor comments. 🙂

[Zelldon]

@pihme @deepthidevaki thanks for reviewing. I applied almost all review hints, except the interface changes. It felt like it make it more complex then necessary and I would need to do more changes. We can discuss them if necessary. Happy to hear your feedback.

[deepthidevaki]

Thanks for the changes. I have just 2 more comments/questions.

[deepthidevaki]

👍

[Zelldon]

bors r+

[zeebe-bors]

Build succeeded

• continuous-integration/jenkins/branch

[Zelldon]

Forget to run the benchmark before merge sorry. I run overnight the benchmarks and they look good.

Benchmark with PR changes:

Benchmark with current Develop (incl. changes):

Benchmark with Develop (excl. changes):