Exception during recovery. Last known sequence number [0]. PersistenceId [abc], due to: null

Akka Persistence / Event Sourcing
1

Hi,
I’m getting the following error when I use Akka persistence with Cassandra plugin:

Caused by: akka.persistence.typed.internal.JournalFailureException: Exception during recovery. Last known sequence number [0]. PersistenceId [abc], due to: null
	at akka.persistence.typed.internal.ReplayingEvents.onRecoveryFailure(ReplayingEvents.scala:221)
	at akka.persistence.typed.internal.ReplayingEvents.onJournalResponse(ReplayingEvents.scala:143)
	... 31 more

my build.sbt is :

libraryDependencies += "com.typesafe.akka" %% "akka-persistence-typed" % "2.6.4"
libraryDependencies += "org.slf4j" % "slf4j-simple" % "1.8.0-beta4"
val AkkaVersion = "2.6.4"
libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-persistence-cassandra" % "1.0.0",
  "com.typesafe.akka" %% "akka-persistence" % AkkaVersion,
  "com.typesafe.akka" %% "akka-persistence-query" % AkkaVersion,
  "com.typesafe.akka" %% "akka-cluster-tools" % AkkaVersion
)

my code is:

object MyPersistentBehavior {

  sealed trait Command
  final case class Add(data: String) extends Command
  case object Clear extends Command
  case class Read(id: String) extends Command

  sealed trait Event
  final case class Added(data: String) extends Event
  case object Cleared extends Event

  final case class State(history: List[String] = Nil)

  val commandHandler: (State, Command) => Effect[Event, State] = { (state, command) =>
    command match {
      case Add(data) =>
          println("接收到Add命令")
          Effect.persist(Added(data))
      case Clear     => Effect.persist(Cleared)
      case Read(id) =>
        println("接收到Read命令")
          println(state)
          Effect.none
    }
  }

  val eventHandler: (State, Event) => State = { (state, event) =>
    event match {
      case Added(data) =>
        println("持久化成功")
        state.copy((data :: state.history).take(5))
      case Cleared     => State(Nil)

    }
  }

  def apply(id:String): Behavior[Command] =
    EventSourcedBehavior[Command, Event, State](
      persistenceId = PersistenceId.ofUniqueId(id),
      emptyState = State(Nil),
      commandHandler = commandHandler,
      eventHandler = eventHandler)

}

and:

def main(args: Array[String]): Unit = {
    val theSystem = ActorSystem(MyPersistentBehavior("abc"),"cook")   
    theSystem ! Add("lklk")
}

Am I doing something wrong here?

2

[cook-akka.actor.default-dispatcher-3] INFO akka.event.slf4j.Slf4jLogger - Slf4jLogger started
SLF4J: A number (1) of logging calls during the initialization phase have been intercepted and are
SLF4J: now being replayed. These are subject to the filtering rules of the underlying logging system.
SLF4J: See also http://www.slf4j.org/codes.html#replay
[cook-akka.persistence.cassandra.default-dispatcher-8] INFO com.datastax.oss.driver.internal.core.context.InternalDriverContext - Could not register Geo codecs; this is normal if ESRI was explicitly excluded from classpath
[cook-akka.persistence.cassandra.default-dispatcher-8] INFO com.datastax.oss.driver.internal.core.DefaultMavenCoordinates - DataStax Java driver for Apache Cassandra® (com.datastax.oss:java-driver-core) version 4.5.1
[cook-akka.persistence.cassandra.default-dispatcher-8] INFO com.datastax.oss.driver.internal.core.context.InternalDriverContext - Could not register Graph extensions; this is normal if Tinkerpop was explicitly excluded from classpath
[s0-admin-0] INFO com.datastax.oss.driver.internal.core.time.Clock - Could not access native clock (see debug logs for details), falling back to Java system clock
[s0-io-0] INFO com.datastax.oss.driver.internal.core.channel.ChannelFactory - [s0] Failed to connect with protocol DSE_V2, retrying with DSE_V1
[s0-io-1] INFO com.datastax.oss.driver.internal.core.channel.ChannelFactory - [s0] Failed to connect with protocol DSE_V1, retrying with V4
[cook-akka.persistence.cassandra.default-dispatcher-7] INFO com.datastax.oss.driver.api.core.uuid.Uuids - PID obtained through native call to getpid(): 8292
[s0-admin-0] WARN com.datastax.oss.driver.internal.core.metadata.token.NetworkTopologyReplicationStrategy - [s0] Error while computing token map for replication settings {class=org.apache.cassandra.locator.NetworkTopologyStrategy, datacenter1=3}: could not achieve replication factor 3 for datacenter datacenter1 (found only 1 replicas).
[cook-akka.actor.default-dispatcher-3] ERROR com.MyPersistentBehavior$ - Supervisor StopSupervisor saw failure: Exception during recovery. Last known sequence number [0]. PersistenceId [abc], due to: Exception during recovery. Last known sequence number [0]. PersistenceId [abc], due to: null
akka.persistence.typed.internal.JournalFailureException: Exception during recovery. Last known sequence number [0]. PersistenceId [abc], due to: Exception during recovery. Last known sequence number [0]. PersistenceId [abc], due to: null
at akka.persistence.typed.internal.ReplayingEvents.onRecoveryFailure(ReplayingEvents.scala:221)
at akka.persistence.typed.internal.ReplayingEvents.onJournalResponse(ReplayingEvents.scala:153)
at akka.persistence.typed.internal.ReplayingEvents.onMessage(ReplayingEvents.scala:90)
at akka.persistence.typed.internal.ReplayingEvents.onMessage(ReplayingEvents.scala:65)
at akka.actor.typed.scaladsl.AbstractBehavior.receive(AbstractBehavior.scala:83)
at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:55)
at akka.persistence.typed.internal.EventSourcedBehaviorImpl$$anon$1.aroundReceive(EventSourcedBehaviorImpl.scala:157)
at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:83)
at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:55)
at akka.actor.typed.internal.SimpleSupervisor.aroundReceive(Supervision.scala:123)
at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:83)
at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:55)
at akka.actor.typed.internal.adapter.GuardianStopInterceptor.aroundReceive(GuardianStartupBehavior.scala:59)
at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:83)
at akka.actor.typed.Behavior$.interpret(Behavior.scala:274)
at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:230)
at akka.actor.typed.internal.adapter.ActorAdapter.handleMessage(ActorAdapter.scala:125)
at akka.actor.typed.internal.adapter.ActorAdapter.aroundReceive(ActorAdapter.scala:105)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:573)
at akka.actor.ActorCell.invoke(ActorCell.scala:543)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:269)
at akka.dispatch.Mailbox.run(Mailbox.scala:230)
at akka.dispatch.Mailbox.exec(Mailbox.scala:242)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: akka.persistence.typed.internal.JournalFailureException: Exception during recovery. Last known sequence number [0]. PersistenceId [abc], due to: null
at akka.persistence.typed.internal.ReplayingEvents.onRecoveryFailure(ReplayingEvents.scala:221)
at akka.persistence.typed.internal.ReplayingEvents.onJournalResponse(ReplayingEvents.scala:143)
… 31 more
Caused by: java.lang.NullPointerException
at com.datastax.oss.driver.internal.core.DefaultConsistencyLevelRegistry.nameToCode(DefaultConsistencyLevelRegistry.java:46)
at com.datastax.oss.driver.internal.core.cql.Conversions.toMessage(Conversions.java:124)
at com.datastax.oss.driver.internal.core.cql.CqlRequestHandler.(CqlRequestHandler.java:173)
at com.datastax.oss.driver.internal.core.cql.CqlRequestAsyncProcessor.process(CqlRequestAsyncProcessor.java:44)
at com.datastax.oss.driver.internal.core.cql.CqlRequestAsyncProcessor.process(CqlRequestAsyncProcessor.java:29)
at com.datastax.oss.driver.internal.core.session.DefaultSession.execute(DefaultSession.java:210)
at com.datastax.oss.driver.api.core.cql.AsyncCqlSession.executeAsync(AsyncCqlSession.java:41)
at akka.stream.alpakka.cassandra.scaladsl.CassandraSession.$anonfun$selectResultSet$1(CassandraSession.scala:197)
at scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:302)
at scala.concurrent.impl.Promise.$anonfun$transformWith$1(Promise.scala:37)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60)
at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:92)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12)
at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:81)
at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:92)
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:47)
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:47)
… 4 more

3

The NPE is coming from the DataStax Java Driver. Are you overriding any of its configuration regarding consistencies?

5

yes,you are right.I find my error. Thank you!!!