apache / pekko-persistence-cassandra Goto Github PK

View Code? Open in Web Editor NEW

9.0 15.0 9.0 2.27 MB

A replicated Apache Pekko Persistence journal backed by Apache Cassandra

Home Page: https://pekko.apache.org/

License: Apache License 2.0

Scala 100.00%

cassandra cqrs eventsourcing java pekko pekko-persistence scala

pekko-persistence-cassandra's People

Contributors

Stargazers

Watchers

Forkers

pjfanning mdedetrich nvollmar mehmetsalgar salgar-org laglangyue jxnu-liguobin roiocam he-pin

pekko-persistence-cassandra's Issues

Feedback: ScyllaDB support and compatibility considerations.

When attempting to use the tables-autocreate feature specified in an application.conf file, I encountered a failure due to a compatibility issue between Cassandra and ScyllaDB. The problem arises because the unchecked_tombstone_compaction option, used within the compaction strategy configuration, is not recognized by ScyllaDB.

The plugin expects this option during the table creation process. The relevant code can be found here.

I worked around this by manually creating the necessary tables without using the unchecked_tombstone_compaction option. This workaround suggests that broader compatibility might be achievable with some adjustments.
Given the growing popularity and adoption of ScyllaDB, it would be nice if future versions of the plugin could consider providing some support for ScyllaDB.

rethink `cassandra-launcher` approach

We're using sbt-assembly in an unusual way for cassandra-launcher - I'm not sure sbt-assembly really promised to do what we ask it to do here.

This causes trouble trying to update of sbt-multi-jvm or sbt-assembly.

uptake pekko jars

pekko-connectors, pekko core libs, etc. are all now available as snapshot jars in repository.apache.org

compile with Scala 3

Waiting for pekko-http/pekko-management Scala 3 artifacts

Move link validator to nightly/weekly job

like apache/pekko#1309

fix unstable tests

org.apache.pekko.persistence.cassandra.query.EventsByTagSpecBackTracking

sort delayed events by timeuuid

org.apache.pekko.persistence.cassandra.query.EventsByTagPersistenceIdCleanupSpec

drop state and trigger new persistence id lookup periodically

org.apache.pekko.persistence.cassandra.EventsByTagRecoverySpec

must recover if snapshot is for the latest sequence nr

investigate datastax link issues (docs)

## HTTP failure response
`[https://docs.datastax.com/en/developer/java-driver/4.17/`](https://docs.datastax.com/en/developer/java-driver/4.17/%60) status 404 Not Found
 - index.html
 - overview.html

`[https://docs.datastax.com/en/developer/java-driver/4.17/manual/core/configuration/`](https://docs.datastax.com/en/developer/java-driver/4.17/manual/core/configuration/%60) status 404 Not Found
 - configuration.html
 - journal.html

acknowledge use of code copied from Apache Cassandra

https://github.com/apache/incubator-pekko-persistence-cassandra/blob/5ca43b58c6fe74b29cce776797d250085446fc49/core/src/main/scala/org/apache/pekko/persistence/cassandra/compaction/LeveledCompactionStrategy.scala#L19

is it possible just to reference the Cassandra code directly (jar dependency)?
if not, we need entries in our LICENCE and NOTICE acknowledging the copying of this code
we should probably replace our Pekko ASF header with Apache Cassandra ASF header (which is the standard one, unmodified)

https://github.com/apache/cassandra/blob/cassandra-2.2/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java

publish snapshot jars (nightly)

after #29 is ready

revert cassandra driver to 4.6.1

last OSS akka release (1.0.6) depended on cassandra driver v4.6.1

https://mvnrepository.com/artifact/com.typesafe.akka/akka-persistence-cassandra_2.13/1.0.6
inherits cassandra dependency transitively from https://mvnrepository.com/artifact/com.lightbend.akka/akka-stream-alpakka-cassandra_2.13/2.0.2
the Pekko fork was made after some prep changes were made for the BSL release (including the upgrade of the cassandra driver)
the new cassandra driver depends on a newer version of Jackson - and Jackson is a dependency we have in a lot of places so this makes things complicated in pekko-projection
apache/pekko#7 (comment)
also needs a downgrade in https://github.com/apache/incubator-pekko-connectors/blob/main/project/Dependencies.scala

fyi @nvollmar @jrudolph @mdedetrich

I'm neutral enough - I can see some benefit to just simply updating Jackson to latest v2.12 or latest v2.14 - but this revert might be closer to our usual aim as to trying to closely match the last OSS Akka releases.

Spawning entities fails continually with JournalFailureException when PreparedStatement initialization fails

scala: 2.13.12
pekko: 1.0.1
pekko-persistence-cassandra: 0.0.0-1120-5b7555fe-SNAPSHOT

CassandraJournal caches Future[PreparedStatement] with lazy val.
It caches failed Future[PreparedStatement] when PreparedStatement initialization fails.
https://github.com/apache/incubator-pekko-persistence-cassandra/blob/7741ba0ab3be3ab0c132dcfc866d14c25e2c08de/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraJournal.scala#L117-L152
The failure can occur when the app starts where Cassandra or the network is unstable.

The broken cache makes it fail to read events from the Journal continually.

[2023-10-02 06:43:27,936] [ERROR] [com.example.Counter$] [system-pekko.actor.default-dispatcher-14] [] - Supervisor StopSupervisor saw failure: Exception during recovery. Last known sequence number [0]. PersistenceId [example2], due to: Exception during recovery. Last known sequence number [0]. PersistenceId [example2], due to: Query timed out after PT0.5S
org.apache.pekko.persistence.typed.internal.JournalFailureException: Exception during recovery. Last known sequence number [0]. PersistenceId [example2], due to: Exception during recovery. Last known sequence number [0]. PersistenceId [example2], due to: Query timed out after PT0.5S
        at org.apache.pekko.persistence.typed.internal.ReplayingEvents.onRecoveryFailure(ReplayingEvents.scala:278)
        at org.apache.pekko.persistence.typed.internal.ReplayingEvents.onJournalResponse(ReplayingEvents.scala:211)
        at org.apache.pekko.persistence.typed.internal.ReplayingEvents.onMessage(ReplayingEvents.scala:109)
        at org.apache.pekko.persistence.typed.internal.ReplayingEvents.onMessage(ReplayingEvents.scala:84)
        at org.apache.pekko.actor.typed.scaladsl.AbstractBehavior.receive(AbstractBehavior.scala:93)
        at org.apache.pekko.actor.typed.Behavior$.interpret(Behavior.scala:283)
        at org.apache.pekko.actor.typed.Behavior$.interpretMessage(Behavior.scala:239)
        at org.apache.pekko.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:67)
        at org.apache.pekko.persistence.typed.internal.EventSourcedBehaviorImpl$$anon$1.aroundReceive(EventSourcedBehaviorImpl.scala:222)
        at org.apache.pekko.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:95)
        at org.apache.pekko.actor.typed.Behavior$.interpret(Behavior.scala:283)
        at org.apache.pekko.actor.typed.Behavior$.interpretMessage(Behavior.scala:239)
        at org.apache.pekko.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:67)
        at org.apache.pekko.actor.typed.internal.SimpleSupervisor.aroundReceive(Supervision.scala:141)
        at org.apache.pekko.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:95)
        at org.apache.pekko.actor.typed.Behavior$.interpret(Behavior.scala:283)
        at org.apache.pekko.actor.typed.Behavior$.interpretMessage(Behavior.scala:239)
        at org.apache.pekko.actor.typed.internal.adapter.ActorAdapter.handleMessage(ActorAdapter.scala:141)
        at org.apache.pekko.actor.typed.internal.adapter.ActorAdapter.aroundReceive(ActorAdapter.scala:117)
        at org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590)
        at org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557)
        at org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:280)
        at org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:241)
        at org.apache.pekko.dispatch.Mailbox.exec(Mailbox.scala:253)
        at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
        at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
        at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
        at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
        at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)
Caused by: org.apache.pekko.persistence.typed.internal.JournalFailureException: Exception during recovery. Last known sequence number [0]. PersistenceId [example2], due to: Query timed out after PT0.5S
        at org.apache.pekko.persistence.typed.internal.ReplayingEvents.onRecoveryFailure(ReplayingEvents.scala:278)
        at org.apache.pekko.persistence.typed.internal.ReplayingEvents.onJournalResponse(ReplayingEvents.scala:201)
        ... 27 common frames omitted
Caused by: com.datastax.oss.driver.api.core.DriverTimeoutException: Query timed out after PT0.5S
        at com.datastax.oss.driver.internal.core.cql.CqlPrepareHandler.lambda$scheduleTimeout$1(CqlPrepareHandler.java:164)
        at io.netty.util.HashedWheelTimer$HashedWheelTimeout.run(HashedWheelTimer.java:715)
        at io.netty.util.concurrent.ImmediateExecutor.execute(ImmediateExecutor.java:34)
        at io.netty.util.HashedWheelTimer$HashedWheelTimeout.expire(HashedWheelTimer.java:703)
        at io.netty.util.HashedWheelTimer$HashedWheelBucket.expireTimeouts(HashedWheelTimer.java:790)
        at io.netty.util.HashedWheelTimer$Worker.run(HashedWheelTimer.java:503)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:829)

We can reproduce the problem with the following sample project.

pekko-persistence-cassandra-prepared-statement-cache-issue.zip.

migration doc - remove old Lightbend stuff

best to link to Lightbend doc and say that users will need to migrate to a particular akka-persistence-cassandra version before attempting to migrate to pekko version
we can have a link to pekko core migration
and promise to provide a more detailed migration doc later

full migration doc

obvious package / class renames
config scope renames (i.e. akka.persistence.cassandra -> pekko.persistence.cassandra)
keyspace rename (akka -> pekko)

option 1: overwrite config to keep using akka keyspace
option 2: manual to copy keyspace to pekko (and eventually drop the akka keyspace)

MultiNodeClusterSpec does not compile

https://github.com/apache/incubator-pekko-persistence-cassandra/blob/main/core/src/multi-jvm/scala/org/apache/pekko/cluster/persistence/cassandra/MultiNodeClusterSpec.scala#L20

The FlightRecordingSupport class is not in the pekko-multi-node-testkit jar or in akka-multi-node-testkit.

MultiNodeClusterSpec is in multi-jvm source so maybe it doesn't get compiled by sbt builds normally (eg in the CI runs).

MultiNodeClusterSpec fails to compile in my IDE.