This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 72e663f9d2 add akka PRs from v2.8.4 (#3432)
72e663f9d2 is described below
commit 72e663f9d2764c1a19613e915e20b6c3c11d9d9b
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Aug 15 10:32:38 2026 +0100
add akka PRs from v2.8.4 (#3432)
compile
Update EventWriter.scala
Update EventWriterSpec.scala
scalafmt
doc fixes
javafmt
Delete diff.txt
---
docs/src/main/paradox/io-dns.md | 2 +-
docs/src/main/paradox/split-brain-resolver.md | 12 +-
.../typed/internal/EventWriterSpec.scala | 214 +++++++++++
.../typed/scaladsl/EventSourcedBehaviorSpec.scala | 69 +++-
.../src/main/resources/reference.conf | 8 +
.../persistence/typed/internal/EventWriter.scala | 391 +++++++++++++++++++++
.../typed/internal/ReplayingEvents.scala | 5 +-
.../persistence/typed/state/javadsl/Effect.scala | 7 +-
.../typed/DurableStatePersistentBehaviorTest.java | 5 +
persistence/src/main/resources/reference.conf | 3 +
.../apache/pekko/persistence/Eventsourced.scala | 1 +
.../persistence/journal/inmem/InmemJournal.scala | 14 +-
.../pekko/persistence/PersistentActorSpec.scala | 34 ++
13 files changed, 748 insertions(+), 17 deletions(-)
diff --git a/docs/src/main/paradox/io-dns.md b/docs/src/main/paradox/io-dns.md
index 7f0095449b..adb48d6c7b 100644
--- a/docs/src/main/paradox/io-dns.md
+++ b/docs/src/main/paradox/io-dns.md
@@ -27,7 +27,7 @@ Users should pick one of the built in extensions.
@@@
-Pekko DNS is a pluggable way to interact with DNS. Implementations much
implement `org.apache.pekko.io.DnsProvider` and provide a configuration
+Pekko DNS is a pluggable way to interact with DNS. Implementations must
implement `org.apache.pekko.io.DnsProvider` and provide a configuration
block that specifies the implementation via `provider-object`.
@@@ note { title="DNS via Pekko Discovery" }
diff --git a/docs/src/main/paradox/split-brain-resolver.md
b/docs/src/main/paradox/split-brain-resolver.md
index a92fa7218b..ac52324fd5 100644
--- a/docs/src/main/paradox/split-brain-resolver.md
+++ b/docs/src/main/paradox/split-brain-resolver.md
@@ -227,22 +227,22 @@ of 4 and 5 nodes the side with 5 nodes will survive and
the other 4 nodes will b
in the 5 node cluster, no more failures can be handled, because the remaining
cluster size would be
less than 5. In the case of another failure in that 5 node cluster all nodes
will be downed.
-Therefore it is important that you join new nodes when old nodes have been
removed.
+Therefore, it is important that you join new nodes when old nodes have been
removed.
Another consequence of this is that if there are unreachable nodes when
starting up the cluster,
before reaching this limit, the cluster may shut itself down immediately. This
is not an issue
if you start all nodes at approximately the same time or use the
`pekko.cluster.min-nr-of-members`
-to define required number of members before the leader changes member status
of 'Joining' members to 'Up'
+to define required number of members before the leader changes member status
of 'Joining' members to 'Up'.
You can tune the timeout after which downing decisions are made using the
`stable-after` setting.
You should not add more members to the cluster than **quorum-size * 2 - 1**. A
warning is logged
-if this recommendation is violated. If the exceeded cluster size remains when
a SBR decision is
+if this recommendation is violated. If the exceeded cluster size remains when
an SBR decision is
needed it will down all nodes because otherwise there is a risk that both
sides may down each
other and thereby form two separate clusters.
-For rolling updates it's best to leave the cluster gracefully via
+For rolling updates, it's best to leave the cluster gracefully via
@ref:[Coordinated Shutdown](coordinated-shutdown.md) (SIGTERM).
-For successful leaving SBR will not be used (no downing) but if there is an
unreachability problem
+For successful leaving, SBR will not be used (no downing) but if there is an
unreachability problem
at the same time as the rolling update is in progress there could be an SBR
decision. To avoid that
the total number of members limit is not exceeded during the rolling update
it's recommended to
leave and fully remove one node before adding a new one, when using
`static-quorum`.
@@ -380,7 +380,7 @@ See also configuration and additional dependency in
[Kubernetes Lease]($pekko.do
## Indirectly connected nodes
-In a malfunctional network there can be situations where nodes are observed as
unreachable via some network
+In a malfunctioning network there can be situations where nodes are observed
as unreachable via some network
links but they are still indirectly connected via other nodes, i.e. it's not a
clean network partition (or node crash).
When this situation is detected the Split Brain Resolvers will keep fully
connected nodes and down all the indirectly
diff --git
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterSpec.scala
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterSpec.scala
new file mode 100644
index 0000000000..918fb4b4b4
--- /dev/null
+++
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterSpec.scala
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.typed.internal
+
+import scala.concurrent.ExecutionContext
+import scala.concurrent.Future
+import scala.concurrent.duration.DurationInt
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.pattern.StatusReply
+import pekko.persistence.AtomicWrite
+import pekko.persistence.JournalProtocol
+
+import com.typesafe.config.ConfigFactory
+import org.scalatest.wordspec.AnyWordSpecLike
+
+object EventWriterSpec {
+ def config =
+ ConfigFactory
+ .parseString("""
+ pekko.persistence.journal.inmem.delay-writes=10ms
+ """)
+ .withFallback(ConfigFactory.load())
+ .resolve()
+}
+
+class EventWriterSpec
+ extends ScalaTestWithActorTestKit(EventWriterSpec.config)
+ with AnyWordSpecLike
+ with LogCapturing {
+
+ private val settings = EventWriter.EventWriterSettings(10, 5.seconds)
+ implicit val ec: ExecutionContext = testKit.system.executionContext
+
+ "The event writer" should {
+
+ "handle duplicates" in new TestSetup {
+ sendWrite(1)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ // should also be ack:ed
+ sendWrite(1)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "handle batched duplicates" in new TestSetup {
+ // first write
+ sendWrite(1)
+ // first batch
+ for (n <- 2L to 10L) {
+ sendWrite(n)
+ }
+ // 0 will be written directly
+ journalAckWrite() should ===(1)
+ clientExpectSuccess(1)
+
+ // completing 1 triggers write of batch with 0-9
+ // second
+ for (n <- 1L to 10L) {
+ sendWrite(n)
+ }
+ // batch 0-9 in flight, writes in the meanwhile go in a new batch
+ journalAckWrite() should ===(9)
+ journalFailWrite("duplicate") should ===(10)
+ journalHighestSeqNr(10L)
+
+ clientExpectSuccess(19)
+ }
+
+ "handle batches with half duplicates" in new TestSetup {
+ for (n <- 1L to 10L) {
+ sendWrite(n)
+ }
+ journalAckWrite() should ===(1)
+ journalAckWrite() should ===(9)
+ clientExpectSuccess(10)
+
+ for (n <- 5L until 15L) {
+ sendWrite(n)
+ }
+ journalFailWrite("duplicate") should ===(1) // seq nr 5
+ journalHighestSeqNr(10L)
+ journalFailWrite("duplicate") should ===(9) // batch of 6-15
+ journalHighestSeqNr(10L)
+ journalAckWrite() should ===(4) // new write of 11-15 (non duplicates)
+
+ // all writes succeeded
+ clientExpectSuccess(10)
+ }
+
+ "pass real errors from journal back" in new TestSetup {
+ sendWrite(1L)
+ journalFailWrite("error error")
+ // duplicate handling will ask for highest seq nr, can't know it is an
actual error
+ journalHighestSeqNr(0L)
+ val response = clientProbe.receiveMessage()
+ response.isError should ===(true)
+ response.getError.getMessage should ===("Journal write failed")
+ }
+
+ "ignores old failures when replay triggered" in new TestSetup {
+ sendWrite(1L) // triggers write
+
+ sendWrite(1L) // goes into batch
+ sendWrite(2L)
+ journalAckWrite()
+
+ val firstWrite =
fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ val payloads = firstWrite.messages.head.asInstanceOf[AtomicWrite].payload
+ // signal first failure, triggers HighestSeq
+ firstWrite.persistentActor ! JournalProtocol.WriteMessageFailure(
+ payloads.head,
+ new RuntimeException("duplicate"),
+ firstWrite.actorInstanceId)
+
+ // replay response triggers partial rewrite, seq nr 2
+ val replayRequest =
fakeJournal.expectMessageType[JournalProtocol.ReplayMessages]
+ replayRequest.persistentActor ! JournalProtocol.RecoverySuccess(1L)
+
+ // but then original write failure arrives for seq nr 2 after sequence
number lookup succeeded
+ firstWrite.persistentActor ! JournalProtocol.WriteMessageFailure(
+ payloads.tail.head,
+ new RuntimeException("duplicate"),
+ firstWrite.actorInstanceId)
+ firstWrite.persistentActor ! JournalProtocol.WriteMessagesFailed(
+ new RuntimeException("duplicate"),
+ payloads.size)
+
+ // ack the partial retry
+ journalAckWrite() should ===(1)
+
+ clientExpectSuccess(3)
+ }
+
+ "handle writes to many pids" in {
+ val writer = spawn(EventWriter("pekko.persistence.journal.inmem",
settings))
+ val probe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
+ (0 to 1000).map { pidN =>
+ Future {
+ for (n <- 0 until 20) {
+ writer ! EventWriter.Write(s"pid$pidN", n.toLong, n.toString,
None, Set.empty, probe.ref)
+ }
+ }
+ }
+ probe.receiveMessages(20 * 1000, 20.seconds)
+ }
+ }
+
+ trait TestSetup {
+ def pid1 = "pid1"
+ val fakeJournal = createTestProbe[JournalProtocol.Message]()
+ val writer = spawn(EventWriter(fakeJournal.ref, settings))
+ val clientProbe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
+
+ def sendWrite(seqNr: Long, pid: String = pid1): Unit = {
+ writer ! EventWriter.Write(pid, seqNr, seqNr.toString, None, Set.empty,
clientProbe.ref)
+ }
+
+ def journalAckWrite(pid: String = pid1): Int = {
+ val write = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ write.messages should have size 1
+ val atomicWrite = write.messages.head.asInstanceOf[AtomicWrite]
+ atomicWrite.payload.foreach { repr =>
+ repr.persistenceId should ===(pid)
+ write.persistentActor ! JournalProtocol.WriteMessageSuccess(repr,
write.actorInstanceId)
+ }
+ write.persistentActor ! JournalProtocol.WriteMessagesSuccessful
+ atomicWrite.payload.size
+ }
+
+ def journalFailWrite(reason: String, pid: String = pid1): Int = {
+ val write = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ write.messages should have size 1
+ val atomicWrite = write.messages.head.asInstanceOf[AtomicWrite]
+ atomicWrite.payload.foreach { repr =>
+ repr.persistenceId should ===(pid)
+ write.persistentActor ! JournalProtocol.WriteMessageFailure(
+ repr,
+ new RuntimeException(reason),
+ write.actorInstanceId)
+ }
+ write.persistentActor ! JournalProtocol.WriteMessagesFailed(
+ new RuntimeException(reason),
+ atomicWrite.payload.size)
+ atomicWrite.payload.size
+ }
+
+ def journalHighestSeqNr(highestSeqNr: Long): Unit = {
+ val replay =
fakeJournal.expectMessageType[JournalProtocol.ReplayMessages]
+ replay.persistentActor ! JournalProtocol.RecoverySuccess(highestSeqNr)
+ }
+
+ def clientExpectSuccess(n: Int): Unit = {
+ clientProbe.receiveMessages(n).foreach { reply =>
+ reply.isSuccess should be(true)
+ }
+ }
+ }
+}
diff --git
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorSpec.scala
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorSpec.scala
index 0a65ea9645..efc9d43093 100644
---
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorSpec.scala
+++
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorSpec.scala
@@ -37,8 +37,11 @@ import pekko.actor.typed.scaladsl.ActorContext
import pekko.actor.typed.scaladsl.Behaviors
import pekko.persistence.{ SnapshotMetadata => ClassicSnapshotMetadata }
import pekko.persistence.{ SnapshotSelectionCriteria =>
ClassicSnapshotSelectionCriteria }
+import pekko.persistence.FilteredPayload
import pekko.persistence.SelectedSnapshot
import pekko.persistence.journal.inmem.InmemJournal
+import pekko.persistence.typed.EventAdapter
+import pekko.persistence.typed.EventSeq
import pekko.persistence.query.EventEnvelope
import pekko.persistence.query.Offset
import pekko.persistence.query.PersistenceQuery
@@ -120,9 +123,12 @@ object EventSourcedBehaviorSpec {
case object LogThenStop extends Command
case object Fail extends Command
case object StopIt extends Command
+ case object PersistFilteredEvent extends Command
+ final case class GetLastSequenceNumber(replyTo: ActorRef[Long]) extends
Command
sealed trait Event extends CborSerializable
final case class Incremented(delta: Int) extends Event
+ case object FilteredEvent extends Event
final case class State(value: Int, history: Vector[Int]) extends
CborSerializable
@@ -274,19 +280,48 @@ object EventSourcedBehaviorSpec {
case StopIt =>
Effect.none.thenStop()
+ case PersistFilteredEvent =>
+ // FilteredEvent will be converted FilteredPayload in eventAdapter
+ Effect.persist(FilteredEvent)
+
+ case GetLastSequenceNumber(replyTo) =>
+ replyTo ! EventSourcedBehavior.lastSequenceNumber(ctx)
+ Effect.none
+
},
eventHandler = (state, evt) =>
evt match {
case Incremented(delta) =>
probe ! ((state, evt))
State(state.value + delta, state.history :+ state.value)
- }).receiveSignal {
- case (_, RecoveryCompleted) => ()
- case (_, SnapshotCompleted(metadata)) =>
- snapshotProbe ! Success(metadata)
- case (_, SnapshotFailed(_, failure)) =>
- snapshotProbe ! Failure(failure)
- }
+ case FilteredEvent =>
+ state
+ })
+ .eventAdapter(new EventAdapter[Event, Any] {
+ override def toJournal(e: Event): Any =
+ e match {
+ case FilteredEvent => FilteredPayload
+ case _ => e
+ }
+
+ override def manifest(event: Event): String = ""
+
+ override def fromJournal(p: Any, manifest: String): EventSeq[Event] =
+ p match {
+ case FilteredPayload =>
+ throw new IllegalStateException("Unexpected FilteredPayload")
+ case e: Event => EventSeq.single(e)
+ case _ =>
+ throw new IllegalStateException(s"Unexpected event type $p")
+ }
+ })
+ .receiveSignal {
+ case (_, RecoveryCompleted) => ()
+ case (_, SnapshotCompleted(metadata)) =>
+ snapshotProbe ! Success(metadata)
+ case (_, SnapshotFailed(_, failure)) =>
+ snapshotProbe ! Failure(failure)
+ }
}
}
@@ -332,6 +367,26 @@ class EventSourcedBehaviorSpec
probe.expectMessage(State(4, Vector(0, 1, 2, 3)))
}
+ "exclude FilteredEvent in replay of persisted events" in {
+ val pid = nextPid()
+ val c = spawn(counter(pid))
+
+ val probe = TestProbe[State]()
+ val seqNrProbe = TestProbe[Long]()
+ c ! Increment
+ c ! PersistFilteredEvent
+ c ! Increment
+ c ! GetValue(probe.ref)
+ probe.expectMessage(10.seconds, State(2, Vector(0, 1)))
+
+ val c2 = spawn(counter(pid))
+ c2 ! Increment
+ c2 ! GetValue(probe.ref)
+ probe.expectMessage(State(3, Vector(0, 1, 2)))
+ c2 ! GetLastSequenceNumber(seqNrProbe.ref)
+ seqNrProbe.expectMessage(4L) // seqNr 2 was used for FilteredPayload
+ }
+
"handle Terminated signal" in {
val c = spawn(counter(nextPid()))
val probe = TestProbe[State]()
diff --git a/persistence-typed/src/main/resources/reference.conf
b/persistence-typed/src/main/resources/reference.conf
index 2513dac159..c810d71c01 100644
--- a/persistence-typed/src/main/resources/reference.conf
+++ b/persistence-typed/src/main/resources/reference.conf
@@ -54,6 +54,14 @@ pekko.persistence.typed {
# recursively (that was the default before this setting was introduced).
That might cause a stack
# overflow in case there are many messages to unstash.
recurse-when-unstashing-read-only-commands = false
+
+ event-writer {
+ # The maximum number of events to batch together when writing to the
journal through the event writer
+ max-batch-size = 10
+ # The event-writer occasionally needs to ask the journal about highest
sequence number to handle duplicate
+ # writes, this timeout is for that interaction
+ ask-timeout = 20s
+ }
}
pekko.reliable-delivery {
diff --git
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala
new file mode 100644
index 0000000000..4bb5abd34d
--- /dev/null
+++
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala
@@ -0,0 +1,391 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.typed.internal
+
+import java.net.URLEncoder
+import java.util.UUID
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.concurrent.duration.FiniteDuration
+import scala.jdk.DurationConverters._
+import scala.util.Failure
+import scala.util.Success
+
+import org.apache.pekko
+import pekko.actor.typed.ActorRef
+import pekko.actor.typed.ActorSystem
+import pekko.actor.typed.Behavior
+import pekko.actor.typed.Extension
+import pekko.actor.typed.ExtensionId
+import pekko.actor.typed.SupervisorStrategy
+import pekko.actor.typed.scaladsl.Behaviors
+import pekko.actor.typed.scaladsl.LoggerOps
+import pekko.actor.typed.scaladsl.adapter.ClassicActorRefOps
+import pekko.actor.typed.scaladsl.adapter.TypedActorRefOps
+import pekko.annotation.InternalStableApi
+import pekko.pattern.StatusReply
+import pekko.persistence.AtomicWrite
+import pekko.persistence.JournalProtocol
+import pekko.persistence.Persistence
+import pekko.persistence.PersistentRepr
+import pekko.persistence.journal.Tagged
+import pekko.util.Timeout
+
+/**
+ * INTERNAL API
+ */
+@InternalStableApi
+private[pekko] object EventWriterExtension extends
ExtensionId[EventWriterExtension] {
+ def createExtension(system: ActorSystem[?]): EventWriterExtension = new
EventWriterExtension(system)
+
+ def get(system: ActorSystem[?]): EventWriterExtension = apply(system)
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalStableApi
+private[pekko] class EventWriterExtension(system: ActorSystem[?]) extends
Extension {
+
+ private val settings = EventWriter.EventWriterSettings(system)
+ private val writersPerJournalId = new ConcurrentHashMap[String,
ActorRef[EventWriter.Command]]()
+
+ def writerForJournal(journalId: Option[String]):
ActorRef[EventWriter.Command] =
+ writersPerJournalId.computeIfAbsent(
+ journalId.getOrElse(""),
+ { _ =>
+ system.systemActorOf(
+ EventWriter(journalId.getOrElse(""), settings),
+ s"EventWriter-${URLEncoder.encode(journalId.getOrElse("default"),
"UTF-8")}")
+ })
+
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalStableApi
+private[pekko] object EventWriter {
+
+ type SeqNr = Long
+ type Pid = String
+
+ object EventWriterSettings {
+ def apply(system: ActorSystem[?]): EventWriterSettings =
+ EventWriterSettings(
+ maxBatchSize =
system.settings.config.getInt("pekko.persistence.typed.event-writer.max-batch-size"),
+ askTimeout =
system.settings.config.getDuration("pekko.persistence.typed.event-writer.ask-timeout").toScala)
+
+ }
+ final case class EventWriterSettings(maxBatchSize: Int, askTimeout:
FiniteDuration)
+
+ sealed trait Command
+ final case class Write(
+ persistenceId: Pid,
+ sequenceNumber: SeqNr,
+ event: Any,
+ metadata: Option[Any],
+ tags: Set[String],
+ replyTo: ActorRef[StatusReply[WriteAck]])
+ extends Command
+ final case class WriteAck(persistenceId: Pid, sequenceNumber: SeqNr)
+
+ private case class MaxSeqNrForPid(persistenceId: Pid, sequenceNumber: SeqNr,
originalErrorDesc: String)
+ extends Command
+
+ private def emptyWaitingForWrite = Vector.empty[(PersistentRepr,
ActorRef[StatusReply[WriteAck]])]
+ private case class StateForPid(
+ waitingForReply: Map[SeqNr, (PersistentRepr,
ActorRef[StatusReply[WriteAck]])],
+ waitingForWrite: Vector[(PersistentRepr,
ActorRef[StatusReply[WriteAck]])] = emptyWaitingForWrite,
+ writeErrorHandlingInProgress: Boolean = false,
+ currentTransactionId: Int = 0)
+
+ def apply(journalPluginId: String, eventWriterSettings:
EventWriterSettings): Behavior[Command] =
+ Behaviors
+ .supervise(Behaviors.setup[Command] { context =>
+ val journal =
Persistence(context.system.classicSystem).journalFor(journalPluginId)
+ context.log.debug("Event writer for journal [{}] starting up",
journalPluginId)
+ apply(journal.toTyped, eventWriterSettings)
+ })
+ .onFailure[Exception](SupervisorStrategy.restart)
+
+ def apply(journal: ActorRef[JournalProtocol.Message], eventWriterSettings:
EventWriterSettings): Behavior[Command] =
+ Behaviors
+ .setup[AnyRef] { context =>
+ val writerUuid = UUID.randomUUID().toString
+
+ var perPidWriteState = Map.empty[Pid, StateForPid]
+ implicit val askTimeout: Timeout = eventWriterSettings.askTimeout
+
+ def sendToJournal(transactionId: Int, reprs: Vector[PersistentRepr]) =
{
+ journal ! JournalProtocol.WriteMessages(
+ AtomicWrite(reprs) :: Nil,
+ context.self.toClassic,
+ // Note: we use actorInstanceId to correlate replies from one
write request
+ actorInstanceId = transactionId)
+ }
+
+ def handleUpdatedStateForPid(pid: Pid, newStateForPid: StateForPid):
Unit = {
+ if (newStateForPid.waitingForReply.nonEmpty) {
+ // more waiting replyTo before we could batch it or scrap the entry
+ perPidWriteState = perPidWriteState.updated(pid, newStateForPid)
+ } else {
+ if (newStateForPid.waitingForWrite.isEmpty) {
+ perPidWriteState = perPidWriteState - pid
+ } else {
+ // batch waiting for pid
+ val newReplyTo = newStateForPid.waitingForWrite.map {
+ case (repr, replyTo) => (repr.sequenceNr, (repr, replyTo))
+ }.toMap
+ val updatedState =
+ newStateForPid.copy(
+ newReplyTo,
+ emptyWaitingForWrite,
+ currentTransactionId = newStateForPid.currentTransactionId +
1)
+
+ if (context.log.isTraceEnabled())
+ context.log.traceN(
+ "Writing batch of {} events for pid [{}], seq nrs [{}-{}],
tx id [{}]",
+ newStateForPid.waitingForWrite.size,
+ pid,
+ newStateForPid.waitingForWrite.head._1.sequenceNr,
+ newStateForPid.waitingForWrite.last._1.sequenceNr,
+ updatedState.currentTransactionId)
+
+ val batch = newStateForPid.waitingForWrite.map { case (repr, _)
=> repr }
+ sendToJournal(updatedState.currentTransactionId, batch)
+ perPidWriteState = perPidWriteState.updated(pid, updatedState)
+
+ }
+ }
+ }
+
+ def handleJournalResponse(response: JournalProtocol.Response):
Behavior[AnyRef] =
+ response match {
+ case JournalProtocol.WriteMessageSuccess(message, transactionId) =>
+ val pid = message.persistenceId
+ val sequenceNr = message.sequenceNr
+ perPidWriteState.get(pid) match {
+ case None =>
+ context.log.debugN(
+ "Got write success reply for event with no previous state
for pid, ignoring (pid [{}], seq nr [{}], tx id [{}])",
+ pid,
+ sequenceNr,
+ transactionId)
+ case Some(stateForPid) =>
+ if (transactionId == stateForPid.currentTransactionId) {
+ stateForPid.waitingForReply.get(sequenceNr) match {
+ case None =>
+ context.log.debugN(
+ "Got write error reply for event with no waiting
request for seq nr, ignoring (pid [{}], seq nr [{}], tx id [{}])",
+ pid,
+ sequenceNr,
+ transactionId)
+ case Some((_, waiting)) =>
+ context.log.trace2(
+ "Successfully wrote event persistence id [{}],
sequence nr [{}]",
+ pid,
+ message.sequenceNr)
+ waiting ! StatusReply.success(WriteAck(pid,
sequenceNr))
+ val newState = stateForPid.copy(waitingForReply =
stateForPid.waitingForReply - sequenceNr)
+ handleUpdatedStateForPid(pid, newState)
+ }
+ } else {
+ if (context.log.isTraceEnabled) {
+ context.log.traceN(
+ "Got reply for old tx id [{}] (current [{}]) for pid
[{}], ignoring",
+ transactionId,
+ stateForPid.currentTransactionId,
+ pid)
+ }
+ }
+ }
+ Behaviors.same
+
+ case JournalProtocol.WriteMessageFailure(message, error,
transactionId) =>
+ val pid = message.persistenceId
+ val sequenceNr = message.sequenceNr
+ perPidWriteState.get(pid) match {
+ case None =>
+ context.log.debugN(
+ "Got write error reply for event with no previous state
for pid, ignoring (pid [{}], seq nr [{}], tx id [{}])",
+ pid,
+ sequenceNr,
+ transactionId)
+ case Some(state) =>
+ // write failure could be re-delivery, we need to check
+ state.waitingForReply.get(sequenceNr) match {
+ case None =>
+ context.log.debugN(
+ "Got write error reply for event with no waiting
request for seq nr, ignoring (pid [{}], seq nr [{}], tx id [{}])",
+ pid,
+ sequenceNr,
+ transactionId)
+ case Some(_) =>
+ // quite likely a re-delivery of already persisted
events, the whole batch will fail
+ // check highest seqnr and see if we can ack events
+ if (!state.writeErrorHandlingInProgress &&
state.currentTransactionId == transactionId) {
+ perPidWriteState =
+ perPidWriteState.updated(pid,
state.copy(writeErrorHandlingInProgress = true))
+ context.ask(
+ journal,
+ (replyTo: ActorRef[JournalProtocol.Response]) =>
+ JournalProtocol.ReplayMessages(0L, 0L, 1L, pid,
replyTo.toClassic)) {
+ case
Success(JournalProtocol.RecoverySuccess(highestSequenceNr)) =>
+ MaxSeqNrForPid(pid, highestSequenceNr,
error.getMessage)
+ case Success(unexpected) =>
+ throw new IllegalArgumentException(
+ s"Got unexpected reply from journal
${unexpected.getClass} for pid $pid")
+ case Failure(exception) =>
+ throw new RuntimeException(
+ s"Error finding highest sequence number in
journal for pid $pid",
+ exception)
+ }
+ } else {
+ context.log.traceN(
+ "Ignoring failure for pid [{}], seq nr [{}], tx id
[{}], since write error handling already in progress or old tx id (current tx
id [{}])",
+ pid,
+ sequenceNr,
+ transactionId,
+ state.currentTransactionId)
+ }
+ }
+ }
+ Behaviors.same
+
+ case _ =>
+ // ignore all other journal protocol messages
+ Behaviors.same
+ }
+
+ Behaviors.receiveMessage {
+ case Write(persistenceId, sequenceNumber, event, metadata, tags,
replyTo) =>
+ val payload = if (tags.isEmpty) event else Tagged(event, tags)
+ val repr = PersistentRepr(
+ payload,
+ persistenceId = persistenceId,
+ sequenceNr = sequenceNumber,
+ manifest = "", // adapters would be on the producing side,
already applied
+ writerUuid = writerUuid,
+ sender = pekko.actor.ActorRef.noSender)
+
+ val reprWithMeta = metadata match {
+ case Some(meta) => repr.withMetadata(meta)
+ case _ => repr
+ }
+
+ val newStateForPid =
+ perPidWriteState.get(persistenceId) match {
+ case None =>
+ if (context.log.isTraceEnabled)
+ context.log.traceN(
+ "Writing event persistence id [{}], sequence nr [{}],
payload {}",
+ persistenceId,
+ sequenceNumber,
+ event)
+ sendToJournal(1, Vector(reprWithMeta))
+ StateForPid(
+ Map((reprWithMeta.sequenceNr, (reprWithMeta, replyTo))),
+ emptyWaitingForWrite,
+ currentTransactionId = 1)
+ case Some(state) =>
+ // write in progress for pid, add write to batch and perform
once current write completes
+ if (state.waitingForWrite.size ==
eventWriterSettings.maxBatchSize) {
+ replyTo ! StatusReply.error(
+ s"Max batch reached for pid $persistenceId, at most
${eventWriterSettings.maxBatchSize} writes for " +
+ "the same pid may be in flight at the same time")
+ state
+ } else {
+ if (context.log.isTraceEnabled)
+ context.log.traceN(
+ "Writing event in progress for persistence id [{}],
adding sequence nr [{}], payload {} to batch",
+ persistenceId,
+ sequenceNumber,
+ event)
+ state.copy(waitingForWrite = state.waitingForWrite :+
((reprWithMeta, replyTo)))
+ }
+ }
+ perPidWriteState = perPidWriteState.updated(persistenceId,
newStateForPid)
+ Behaviors.same
+
+ case MaxSeqNrForPid(pid, maxSeqNr, originalErrorDesc) =>
+ // write failed, so we looked up the maxSeqNr to detect if it was
duplicate events, already in journal
+ perPidWriteState.get(pid) match {
+ case None =>
+ context.log.debug2(
+ "Got max seq nr with no waiting previous state for pid,
ignoring (pid [{}], original error desc: {})",
+ pid,
+ originalErrorDesc)
+ case Some(state) =>
+ val sortedSeqs = state.waitingForReply.keys.toSeq.sorted
+ val (alreadyInJournal, needsWrite) =
sortedSeqs.partition(seqNr => seqNr <= maxSeqNr)
+ if (alreadyInJournal.isEmpty) {
+ // error was not about duplicates
+ state.waitingForReply.values.foreach {
+ case (_, replyTo) =>
+ replyTo ! StatusReply.error("Journal write failed")
+ }
+ context.log.warnN(
+ "Failed writing event batch persistence id [{}], sequence
nr [{}-{}]: {}",
+ pid,
+ sortedSeqs.head,
+ sortedSeqs.last,
+ originalErrorDesc)
+ val newState = state.copy(waitingForReply = Map.empty,
writeErrorHandlingInProgress = false)
+ handleUpdatedStateForPid(pid, newState)
+ } else {
+ // ack all already written
+ val stateAfterWritten = alreadyInJournal
+ .foldLeft(state) { (state, seqNr) =>
+ val (_, replyTo) = state.waitingForReply(seqNr)
+ replyTo ! StatusReply.success(WriteAck(pid, seqNr))
+ state.copy(waitingForReply = state.waitingForReply -
seqNr)
+ }
+ .copy(writeErrorHandlingInProgress = false)
+ if (needsWrite.isEmpty) {
+ handleUpdatedStateForPid(pid, stateAfterWritten)
+ } else {
+ // retrigger write for those left if any, note that we do a
+ val reprsToRewrite =
+ stateAfterWritten.waitingForReply.values.map { case
(repr, _) => repr }.toVector
+ if (context.log.isDebugEnabled())
+ context.log.debugN(
+ "Partial batch was duplicates, re-triggering write of
persistence id [{}], sequence nr [{}-{}]",
+ pid,
+ reprsToRewrite.head.sequenceNr,
+ reprsToRewrite.last.sequenceNr)
+ // Not going via batch/handleUpdatedStateForPid here
because adding partial
+ // failure to current waiting could go over batch size
limit
+ val partialRewriteState =
+ stateAfterWritten.copy(currentTransactionId =
stateAfterWritten.currentTransactionId + 1)
+ sendToJournal(partialRewriteState.currentTransactionId,
reprsToRewrite)
+ handleUpdatedStateForPid(pid, partialRewriteState)
+ }
+
+ }
+
+ }
+ Behaviors.same
+
+ case response: JournalProtocol.Response =>
handleJournalResponse(response)
+
+ case unexpected =>
+ context.log.warn("Unexpected message sent to EventWriter [{}],
ignored", unexpected.getClass)
+ Behaviors.same
+
+ }
+ }
+ .narrow[Command]
+
+}
diff --git
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ReplayingEvents.scala
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ReplayingEvents.scala
index 50abf44347..3cf16769fe 100644
---
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ReplayingEvents.scala
+++
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ReplayingEvents.scala
@@ -28,6 +28,7 @@ import pekko.event.Logging
import pekko.persistence._
import pekko.persistence.JournalProtocol._
import pekko.persistence.typed.EmptyEventSeq
+import pekko.persistence.typed.EventSeq
import pekko.persistence.typed.EventsSeq
import pekko.persistence.typed.RecoveryCompleted
import pekko.persistence.typed.RecoveryFailed
@@ -138,7 +139,9 @@ private[pekko] final class ReplayingEvents[C, E, S](
case ReplayedMessage(repr) =>
var eventForErrorReporting: OptionVal[Any] = OptionVal.None
try {
- val eventSeq = setup.eventAdapter.fromJournal(repr.payload,
repr.manifest)
+ val eventSeq =
+ if (repr.payload == FilteredPayload) EventSeq.empty // ignore
FilteredPayload
+ else setup.eventAdapter.fromJournal(repr.payload, repr.manifest)
def handleEvent(event: E): Unit = {
eventForErrorReporting = OptionVal.Some(event)
state = state.copy(seqNr = repr.sequenceNr, eventsReplayed =
state.eventsReplayed + 1)
diff --git
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/Effect.scala
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/Effect.scala
index 95369627ac..29f5442e41 100644
---
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/Effect.scala
+++
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/Effect.scala
@@ -44,7 +44,12 @@ import pekko.persistence.typed.state.internal.SideEffect
*/
final def persist(state: State): EffectBuilder[State] = Persist(state)
- // FIXME add delete effect
+ /**
+ * Delete the persisted state.
+ *
+ * Side effects can be chained with `thenRun`.
+ */
+ def delete(): EffectBuilder[State] =
Delete().asInstanceOf[EffectBuilder[State]]
/**
* Do not persist anything
diff --git
a/persistence-typed/src/test/java/jdocs/org/apache/pekko/persistence/typed/DurableStatePersistentBehaviorTest.java
b/persistence-typed/src/test/java/jdocs/org/apache/pekko/persistence/typed/DurableStatePersistentBehaviorTest.java
index d30deaf6c0..278971ceca 100644
---
a/persistence-typed/src/test/java/jdocs/org/apache/pekko/persistence/typed/DurableStatePersistentBehaviorTest.java
+++
b/persistence-typed/src/test/java/jdocs/org/apache/pekko/persistence/typed/DurableStatePersistentBehaviorTest.java
@@ -105,6 +105,10 @@ public class DurableStatePersistentBehaviorTest {
}
}
+ public enum Delete implements Command<Void> {
+ INSTANCE
+ }
+
// #command
// #state
@@ -150,6 +154,7 @@ public class DurableStatePersistentBehaviorTest {
(state, command) -> Effect().persist(new State(state.get() +
command.value)))
.onCommand(
GetValue.class, (state, command) ->
Effect().reply(command.replyTo, state.get()))
+ .onCommand(Delete.class, (state, command) -> Effect().delete())
.build();
}
// #command-handler
diff --git a/persistence/src/main/resources/reference.conf
b/persistence/src/main/resources/reference.conf
index c00b1f5763..6e13d82a8c 100644
--- a/persistence/src/main/resources/reference.conf
+++ b/persistence/src/main/resources/reference.conf
@@ -266,6 +266,9 @@ pekko.persistence.journal.inmem {
# Turn this on to test serialization of the events
test-serialization = off
+
+ # Useful for tests, increase to make writes take time like an actual
persistent journal
+ delay-writes = 0s
}
# Local file system snapshot store plugin.
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala
index ccdb90c9eb..65d7ec5fe2 100644
--- a/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala
+++ b/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala
@@ -653,6 +653,7 @@ private[persistence] trait Eventsourced
}
{
+ case PersistentRepr(FilteredPayload, _)
=> // ignore
case PersistentRepr(payload, _) if recoveryRunning &&
_receiveRecover.isDefinedAt(payload) =>
_receiveRecover(payload)
case s: SnapshotOffer if _receiveRecover.isDefinedAt(s) =>
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/inmem/InmemJournal.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/inmem/InmemJournal.scala
index 235a003010..09ddde75da 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/inmem/InmemJournal.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/inmem/InmemJournal.scala
@@ -15,6 +15,8 @@ package org.apache.pekko.persistence.journal.inmem
import scala.collection.immutable
import scala.concurrent.Future
+import scala.concurrent.duration.Duration
+import scala.jdk.DurationConverters._
import scala.util.Try
import scala.util.control.NonFatal
@@ -23,6 +25,7 @@ import pekko.actor.ActorRef
import pekko.annotation.ApiMayChange
import pekko.annotation.InternalApi
import pekko.event.Logging
+import pekko.pattern.after
import pekko.persistence.AtomicWrite
import pekko.persistence.JournalProtocol.RecoverySuccess
import pekko.persistence.PersistentRepr
@@ -77,6 +80,12 @@ object InmemJournal {
else false
}
+ private val delayWrites = {
+ val key = "delay-writes"
+ if (cfg.hasPath(key)) cfg.getDuration(key).toScala
+ else Duration.Zero
+ }
+
private val serialization = SerializationExtension(context.system)
private val eventStream = context.system.eventStream
@@ -92,7 +101,10 @@ object InmemJournal {
add(p)
eventStream.publish(InmemJournal.Write(p.payload, p.persistenceId,
p.sequenceNr))
}
- Future.successful(Nil) // all good
+ if (delayWrites.length > 0)
+ after(delayWrites, context.system.scheduler) { Future.successful(Nil)
}(context.dispatcher)
+ else
+ Future.successful(Nil) // all good
} catch {
case NonFatal(e) =>
// serialization problem
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorSpec.scala
index 31f5a2238d..34eae20bfc 100644
---
a/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorSpec.scala
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorSpec.scala
@@ -127,6 +127,24 @@ object PersistentActorSpec {
extends Behavior3PersistentActor(name)
with InmemRuntimePluginConfig
+ class Behavior4PersistentActor(name: String) extends
ExamplePersistentActor(name) {
+ val receiveCommand: Receive = commonBehavior.orElse {
+ case FilteredPayload =>
+ persist(FilteredPayload)(_ => ())
+ case Cmd(data) =>
+ persist(Evt(s"$data-${lastSequenceNr + 1}"))(updateState)
+ }
+
+ override def receiveRecover: Receive = super.receiveRecover.orElse {
+ case FilteredPayload =>
+ throw new IllegalStateException("Unexpected FilteredPayload")
+ }
+ }
+
+ class Behavior4PersistentActorWithInmemRuntimePluginConfig(name: String, val
providedConfig: Config)
+ extends Behavior4PersistentActor(name)
+ with InmemRuntimePluginConfig
+
class ChangeBehaviorInLastEventHandlerPersistentActor(name: String) extends
ExamplePersistentActor(name) {
val newBehavior: Receive = {
case Cmd(data) =>
@@ -975,6 +993,8 @@ abstract class PersistentActorSpec(config: Config) extends
PersistenceSpec(confi
protected def behavior3PersistentActor: ActorRef =
namedPersistentActor[Behavior3PersistentActor]
+ protected def behavior4PersistentActor: ActorRef =
namedPersistentActor[Behavior4PersistentActor]
+
protected def changeBehaviorInFirstEventHandlerPersistentActor: ActorRef =
namedPersistentActor[ChangeBehaviorInFirstEventHandlerPersistentActor]
@@ -1137,6 +1157,17 @@ abstract class PersistentActorSpec(config: Config)
extends PersistenceSpec(confi
persistentActor ! GetState
expectMsg(List("a-1", "a-2", "b-10", "b-11", "b-12", "c-10", "c-11",
"c-12"))
}
+ "exclude FilteredEvent in replay of persisted events" in {
+ val persistentActor = behavior4PersistentActor
+ persistentActor ! GetState
+ expectMsg(List("a-1", "a-2"))
+ persistentActor ! FilteredPayload
+ persistentActor ! Cmd("b")
+ persistentActor ! "boom"
+ persistentActor ! Cmd("c")
+ persistentActor ! GetState
+ expectMsg(List("a-1", "a-2", "b-4", "c-5")) // seqNr 3 was for
FilteredPayload
+ }
"recover on command failure" in {
val persistentActor = behavior3PersistentActor
persistentActor ! Cmd("b")
@@ -1693,6 +1724,9 @@ class InmemPersistentActorWithRuntimePluginConfigSpec
override protected def behavior3PersistentActor: ActorRef =
namedPersistentActorWithProvidedConfig[Behavior3PersistentActorWithInmemRuntimePluginConfig](providedActorConfig)
+ override protected def behavior4PersistentActor: ActorRef =
+
namedPersistentActorWithProvidedConfig[Behavior4PersistentActorWithInmemRuntimePluginConfig](providedActorConfig)
+
override protected def changeBehaviorInFirstEventHandlerPersistentActor:
ActorRef =
namedPersistentActorWithProvidedConfig[
ChangeBehaviorInFirstEventHandlerPersistentActorWithInmemRuntimePluginConfig](providedActorConfig)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
