raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

=per #15457 Correlate persistAsync handlers with journal messages

Browse source 
We have assumed that the handlers can be popped when replies come back from journal, but if messages to journal are in flight when the actor is restarted the handlers does not match up with journal replies.

This solution ignores journal replies that were emitted by an old PersistentActor instance
by passing an uid with the journal messages. This means that the handler will not be
invoked for such messages.

(cherry picked from commit 7ebaaab669c9e467a1ffb4d9ed8b6500e1801a7c)

Conflicts:
	akka-persistence/src/main/scala/akka/persistence/JournalProtocol.scala
	akka-persistence/src/main/scala/akka/persistence/Processor.scala
	akka-persistence/src/main/scala/akka/persistence/journal/AsyncWriteJournal.scala
This commit is contained in:
Patrik Nordwall 2014-06-26 17:35:46 +02:00
parent 33c7f6bb4f
commit 8eec3f92d3
11 changed files with 126 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

10
akka-persistence/src/main/scala/akka/persistence/JournalProtocol.scala
View file

@ -58,7 +58,7 @@ private[persistence] object JournalProtocol {
* @param messages messages to be written.
* @param persistentActor write requestor.
*/
final case class WriteMessages(messages: immutable.Seq[Resequenceable], persistentActor: ActorRef)
final case class WriteMessages(messages: immutable.Seq[Resequenceable], persistentActor: ActorRef, actorInstanceId: Int)
/**
* Reply message to a successful [[WriteMessages]] request. This reply is sent to the requestor
@ -80,7 +80,7 @@ private[persistence] object JournalProtocol {
*
* @param persistent successfully written message.
*/
final case class WriteMessageSuccess(persistent: PersistentRepr)
final case class WriteMessageSuccess(persistent: PersistentRepr, actorInstanceId: Int)
/**
* Reply message to a failed [[WriteMessages]] request. For each contained [[PersistentRepr]] message
@ -89,7 +89,7 @@ private[persistence] object JournalProtocol {
* @param message message failed to be written.
* @param cause failure cause.
*/
final case class WriteMessageFailure(message: PersistentRepr, cause: Throwable)
final case class WriteMessageFailure(message: PersistentRepr, cause: Throwable, actorInstanceId: Int)
/**
* Request to loop a `message` back to `persistent actor`, without persisting the message. Looping of messages
@ -98,14 +98,14 @@ private[persistence] object JournalProtocol {
* @param message message to be looped through the journal.
* @param persistentActor loop requestor.
*/
final case class LoopMessage(message: Any, persistentActor: ActorRef)
final case class LoopMessage(message: Any, persistentActor: ActorRef, actorInstanceId: Int)
/**
* Reply message to a [[LoopMessage]] request.
*
* @param message looped message.
*/
final case class LoopMessageSuccess(message: Any)
final case class LoopMessageSuccess(message: Any, actorInstanceId: Int)
/**
* Request to replay messages to `persistentActor`.