Cassandra snapshot store by rtar · Pull Request #532 · evolution-gaming/kafka-journal

rtar · 2023-11-01T14:29:40Z

It is kind of a draft for now, but any feedback is welcome.

The Idea

The big idea is to get rid of sequence_nr column from a primary key. The presence of this column in Akka Persistence Cassandra causes creation of a tombstone each time the snapshot is deleted and the tombstone stays there until gc_grace_seconds is reached. This causes slower reads and may limit the ability to scan the table whenever the incident happen.

The new implementation has a fixed maximum number of rows per one pesistenceId and updates them in-place using a ring buffer approach.

Why part of Kafka Journal, not a separate library?

The goal is to reuse Kafka Journal infrastructure:

Converting from persistenceId to (id, topic) pair and back,
Reusing serialization approach,
Reusing Cassandra specific utility classes from eventual-cassandra.

How to review?

In this specific PR there are only two useful / added-value classes:
SnapshotStatements - defines the new table used to store snapshots, and CQL statements to get/update/delete them,
SnapshotCassandra - defines the actual ring buffer logic to insert/update/delete snapshots from the new table.

You might also find SnapshotCassandraTest useful as it defines uses cases I was thinking about when writing SnapshotCassandra, and SnapshotPerfSpec which demonstrates that, indeed, a new plugin is faster when reading, and slower when writing snapshots.

Everything else are just wrappers around Akka SnapshotStore API without much added-value code.

Why it is a draft and not ready for merge?

I do not want to break the existing kafka-journal, obviously, so I tried to change as little as possible in a journal plugin, but I am not sure I fully succeeded. I.e. I had to squeeze in the code into CreateSchema and SchemaConfig classes, which creates a new snapshot table if it does not exist.

I am not sure it is a right thing to do, because most people won't use a new snapshotter for a long time.

Besides that, I was thinking that to sell something like this to the users, I should not increase complexity of kafka-journal, hence separating the code from kafka-journal into separate subproject / documenting everything well could be essential.

And the last, but not the least, I kind of hacked serialization part in, and it still needs to be checked / worked on.

The biggest obstacle is how to achieve these goals without making the changelist huge and unreviewable. I am playing with separating the code here (it is PR to PR), but I am not completely sure I am going to a right direction: https://github.com/evolution-gaming/kafka-journal/pull/537/files

coveralls · 2023-11-01T14:37:30Z

Pull Request Test Coverage Report for Build 9712199194

Details

295 of 323 (91.33%) changed or added relevant lines in 12 files are covered.
12 unchanged lines in 10 files lost coverage.
Overall coverage increased (+1.1%) to 82.287%

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
snapshot/src/main/scala/com/evolutiongaming/kafka/journal/SnapshotStoreError.scala	0	1	0.0%
snapshot-cassandra/src/main/scala/com/evolutiongaming/kafka/journal/snapshot/cassandra/SnapshotCassandra.scala	55	58	94.83%
persistence/src/main/scala/akka/persistence/kafka/journal/SnapshotSerializer.scala	28	32	87.5%
snapshot-cassandra/src/main/scala/com/evolutiongaming/kafka/journal/snapshot/cassandra/SnapshotStatements.scala	79	84	94.05%
persistence/src/main/scala/akka/persistence/kafka/journal/CassandraSnapshotStore.scala	57	72	79.17%

Files with Coverage Reduction	New Missed Lines	%
persistence/src/main/scala/akka/persistence/kafka/journal/KafkaJournal.scala	1	84.76%
core/src/main/scala/com/evolutiongaming/kafka/journal/LogFromAkka.scala	1	28.57%
journal/src/main/scala/com/evolutiongaming/kafka/journal/HeadCacheConsumption.scala	1	96.43%
journal/src/main/scala/com/evolutiongaming/kafka/journal/Events.scala	1	90.0%
core/src/main/scala/com/evolutiongaming/kafka/journal/util/ActorSystemOf.scala	1	83.33%
core/src/main/scala/com/evolutiongaming/kafka/journal/util/GracefulFiber.scala	1	75.0%
core/src/main/scala/com/evolutiongaming/kafka/journal/SeqNr.scala	1	89.74%
journal/src/main/scala/com/evolutiongaming/kafka/journal/Journals.scala	1	84.43%
core/src/main/scala/com/evolutiongaming/kafka/journal/Key.scala	1	73.33%
journal/src/main/scala/com/evolutiongaming/kafka/journal/HeadCache.scala	3	86.67%

Totals
Change from base Build 9680733046:	1.1%
Covered Lines:	3670
Relevant Lines:	4460

💛 - Coveralls

t3hnar · 2023-12-01T13:28:53Z

+  def listOf(size: Int): List[BufferNr] =
+    (0 until size).toList.map(fromIntUnsafe)
+
+  def fromIntUnsafe(value: Int): BufferNr =


some validation?

non negative?

where is safe version? :)

t3hnar · 2023-12-01T13:32:29Z


+  implicit class RowOps(val self: Row) extends AnyVal {
+
+    def wasApplied: Boolean = self.getBool("[applied]")


isApplied ?

I named it wasApplied, because it is called like that in Cassandra driver, so I thought it might ease discoverability a bit. But isApplied sounds fine too.

https://docs.datastax.com/en/drivers/java/3.0/com/datastax/driver/core/ResultSet.html#wasApplied--

t3hnar · 2023-12-01T13:35:28Z

  metaJournalTable: String = "metajournal",
  pointerTable: String = "pointer",
  pointer2Table: String = "pointer2",
+  snapshotTable: String = "snapshot",


just a note, akka currently uses snapshots table. We need to be 100% to not conflict with it :)

I did not think about it, to be honest. Other name is required here, indeed.

How about "snapshot_buffer"?

t3hnar · 2023-12-01T13:41:40Z

+
+import java.time.Instant
+
+final case class SnapshotSelectionCriteria(


do you think we need this?

It provides a bit of additional typesafety on top of the original structure.

I can move these parameters to the method arguments though.

Here is the only place it is used: https://github.com/evolution-gaming/kafka-journal/pull/532/files#diff-9ccd7e724e3622d863ea66a0e6c83a3d5bf96a6d0d089ad1235680ca6266556aR114

t3hnar · 2023-12-01T13:44:00Z

+            val (bufferNr, (deleteSnapshot, _)) = oldestSnapshot
+            val wasApplied = statements.updateRecord(key, segmentNr, bufferNr, insertSnapshot, deleteSnapshot)
+            wasApplied.flatMap { wasApplied =>
+              // TODO: consider adding circuit breaker here


could you please explain this comment?

If there is a bug or high contention (with the writer itself?), this call may lead to an infinite loop, which feels like a thing to consider protecting against.

t3hnar · 2023-12-01T13:46:03Z

+      val segmentNr = SegmentNr.min
+
+      def save(key: Key, snapshot: SnapshotRecord[EventualPayloadAndType]): F[Unit] = {
+        statements.selectMetadata(key, segmentNr).flatMap {


on the hot path given that buffer is full, do you need to query ALL snapshots?

I need this information to find the oldest snapshot to be replaced. There are might be the ways to optimize it, though. 🤔

t3hnar · 2023-12-01T13:46:58Z

+          (bufferNr, (seqNr, timestamp))
+        }
+
+        rows.toList.map(_.toMap)


minor, not sure toMap should be part of statement :)

t3hnar · 2023-12-01T13:48:38Z

+        savedSnapshots: Map[BufferNr, (SeqNr, Instant)],
+        insertSnapshot: SnapshotRecord[EventualPayloadAndType]
+      ): F[Unit] = {
+        val sortedSnapshots = savedSnapshots.toList.sortBy { case (_, (seqNr, timestamp)) => (seqNr, timestamp) }


no need to sort collection, just minBy

is it correct to include timestamp here?

Technically, it should never happen, i.e. there should not be two snapshots with the same seqNr. 🤔

dfakhritdinov · 2024-02-01T17:09:24Z

+        savedSnapshots: Map[BufferNr, (SeqNr, Instant)],
+        insertSnapshot: SnapshotRecord[EventualPayloadAndType]
+      ): F[Unit] = {
+        val oldestSnapshot = savedSnapshots.toList.minByOption { case (_, (seqNr, timestamp)) => (seqNr, timestamp) }


Are you expecting to have few snapshots with same seqNr? I do not understand use-case for it, could you please share more details

It should never have several snapshots with the same seqNr, but it seemed like a good idea to be a little bit defensive here.

I.e., if there is a bug or something, it will still filter by timestamp and delete an oldest one, not a random one.

Would it be possible to log (or report any other way) the case of duplicated seqNr? Please resolve the conversation if its not easily possible

be small.

# Conflicts: # eventual-cassandra/src/test/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/CreateSchemaSpec.scala

# Conflicts: # eventual-cassandra/src/main/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/CreateSchema.scala # eventual-cassandra/src/test/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/CreateSchemaSpec.scala # Conflicts: # build.sbt # Conflicts: # build.sbt

It was inherited from original `Event` code, but it seems that it was only needed there to ensure backwards compatibility when introducing JSON payloads and no longer necessary. See c6d0543 for more details.

The reason behind this is that there won't be a flat counterpart of the store. It will be the only one.

sonarqubecloud · 2026-02-05T15:58:05Z

Quality Gate failed

Failed conditions
5.4% Duplication on New Code (required ≤ 3%)

See analysis details on SonarQube Cloud

rtar changed the title ~~Snapshot statements~~ Cassandra snapshot store Nov 21, 2023

t3hnar suggested changes Dec 1, 2023

View reviewed changes

rtar force-pushed the snapshot branch from 48bdbe5 to cdd7458 Compare December 5, 2023 09:12

rtar force-pushed the snapshot branch 3 times, most recently from 7af15e7 to 6a6da8e Compare December 18, 2023 14:07

rtar force-pushed the snapshot branch 2 times, most recently from 611ff72 to 05e761a Compare January 17, 2024 12:05

rtar force-pushed the snapshot branch 2 times, most recently from 40f0178 to e46f1ed Compare January 25, 2024 12:46

rtar added this to the snapshotter milestone Jan 25, 2024

rtar force-pushed the snapshot branch 2 times, most recently from 64b3b10 to 70adbd5 Compare February 1, 2024 15:14

dfakhritdinov reviewed Feb 1, 2024

View reviewed changes

rtar force-pushed the snapshot branch 2 times, most recently from f14d9ac to 09c9161 Compare February 2, 2024 17:30

rtar mentioned this pull request Feb 6, 2024

Add a class to pass SeqNr and snapshot payload in #563

Merged

rtar force-pushed the snapshot branch 5 times, most recently from 81de6f3 to 3dcd233 Compare February 15, 2024 09:33

rtar force-pushed the snapshot branch 2 times, most recently from ae02bdc to 4337c0a Compare February 21, 2024 13:17

rtar force-pushed the snapshot branch 2 times, most recently from d9579f6 to 6c4f376 Compare April 19, 2024 09:39

dfakhritdinov reviewed May 3, 2024

View reviewed changes

Comment thread ...src/main/scala/com/evolutiongaming/kafka/journal/snapshot/cassandra/SnapshotStatements.scala Outdated

rtar force-pushed the snapshot branch from 6c4f376 to eee311e Compare May 7, 2024 11:32

rtar force-pushed the snapshot branch from a51004f to 9936b96 Compare June 14, 2024 10:01

rtar added 28 commits June 28, 2024 14:35

Fix broken test.

590b371

Address review comments.

d9a4753

Use a new table name to avoid conflict with Akka Persistence Cassandra.

6ebea73

Remove segmentNr, because each set of snapshots per key is expected to

965f2e7

be small.

Fixed broken test.

f10b5cc

Remove question mark for segmentNr.

55a0493

Use correct table name.

c0b4f88

# Conflicts: # eventual-cassandra/src/test/scala/com/evolutiongaming/kafka/journal/eventual/cassandra/CreateSchemaSpec.scala

Use correct column index for bufferNr

d5a946b

Separate CassandraSnapshotStoreConfig from KafkaJournalConfig.

9a6d994

Fix compilation issues after rebase

5e05686

Remove unnecessary changes.

db90f9e

Minimize number of changes.

e9babc9

Fixed file names.

37ecc8c

Simplified method signatures.

74345c6

Removed unnecessary copy-paste.

26fe488

Fixed issues caused by merge conflicts.

8d2e676

Simplify CreateSnapshotSchema.

d76d18e

Remove obsolete test.

9cce3d5

Minimize number of changes.

4eb4bf8

Remove option wrapper around a payload.

d9089a8

It was inherited from original `Event` code, but it seems that it was only needed there to ensure backwards compatibility when introducing JSON payloads and no longer necessary. See c6d0543 for more details.

Add scaladoc for SnapshotStoreFlat.

a99b355

Replace drop with delete to adhere to Akka naming convention.

65697d9

Rename SnapshotStoreFlat to SnapshotStore.

17807a6

The reason behind this is that there won't be a flat counterpart of the store. It will be the only one.

Add scaladoc for SnapshotSerializer.

8fecd15

Fix compliation issues after rebase.

1a425e7

Disable LWTs by default.

adb39a9

Print out fractional part of the duration.

c04b508

rtar force-pushed the snapshot branch from 0b78cc4 to c04b508 Compare June 28, 2024 11:41


		implicit class RowOps(val self: Row) extends AnyVal {

		def wasApplied: Boolean = self.getBool("[applied]")


		import java.time.Instant

		final case class SnapshotSelectionCriteria(

Conversation

rtar commented Nov 1, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

The Idea

Why part of Kafka Journal, not a separate library?

How to review?

Why it is a draft and not ready for merge?

Uh oh!

coveralls commented Nov 1, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Pull Request Test Coverage Report for Build 9712199194

Details

💛 - Coveralls

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

sonarqubecloud Bot commented Feb 5, 2026

Quality Gate failed

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

rtar commented Nov 1, 2023 •

edited

Loading

coveralls commented Nov 1, 2023 •

edited

Loading