Skip to content

inmem journal must not reset highestSequenceNr after journal cleanup, #27994 #27995

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
johanandren merged 1 commit into from
Oct 15, 2019
Merged

inmem journal must not reset highestSequenceNr after journal cleanup, #27994 #27995

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions akka-persistence-tck/src/test/scala/akka/persistence/journal/inmem/InmemJournalSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
/*
* Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
*/

package akka.persistence.journal.inmem

import akka.persistence.CapabilityFlag
import akka.persistence.PersistenceSpec
import akka.persistence.journal.JournalSpec

class InmemJournalSpec extends JournalSpec(config = PersistenceSpec.config("inmem", "InmemJournalSpec")) {
override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = CapabilityFlag.off()
}
5 changes: 5 additions & 0 deletions ...istence/src/main/mima-filters/2.5.x.backwards.excludes/issue-27994-inmem-journal.excludes
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
# #27994 changes to internal inmem journal
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.persistence.journal.inmem.InmemMessages.update")
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.persistence.journal.inmem.InmemMessages.akka$persistence$journal$inmem$InmemMessages$$highestSequenceNumbers")
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.persistence.journal.inmem.InmemMessages.akka$persistence$journal$inmem$InmemMessages$$highestSequenceNumbers_=")
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.persistence.journal.inmem.InmemJournal.update")
16 changes: 6 additions & 10 deletions akka-persistence/src/main/scala/akka/persistence/journal/inmem/InmemJournal.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,16 +87,16 @@ private[persistence] class InmemJournal(cfg: Config) extends AsyncWriteJournal w
private[persistence] trait InmemMessages {
// persistenceId -> persistent message
var messages = Map.empty[String, Vector[PersistentRepr]]
// persistenceId -> highest used sequence number
private var highestSequenceNumbers = Map.empty[String, Long]

@helena helena Oct 15, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Makes me wonder if we'd need a faster data structure or if this isn't quite on the hot path then no.

@patriknw patriknw Oct 15, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

inmem journal is only intended for testing so shouldn't be a problem


def add(p: PersistentRepr): Unit =
def add(p: PersistentRepr): Unit = {
messages = messages + (messages.get(p.persistenceId) match {
case Some(ms) => p.persistenceId -> (ms :+ p)
case None => p.persistenceId -> Vector(p)
})

def update(pid: String, snr: Long)(f: PersistentRepr => PersistentRepr): Unit = messages = messages.get(pid) match {
case Some(ms) => messages + (pid -> ms.map(sp => if (sp.sequenceNr == snr) f(sp) else sp))
case None => messages
highestSequenceNumbers =
highestSequenceNumbers.updated(p.persistenceId, math.max(highestSequenceNr(p.persistenceId), p.sequenceNr))
}

def delete(pid: String, snr: Long): Unit = messages = messages.get(pid) match {
Expand All @@ -111,11 +111,7 @@ private[persistence] trait InmemMessages {
}

def highestSequenceNr(pid: String): Long = {
val snro = for {
ms <- messages.get(pid)
m <- ms.lastOption
} yield m.sequenceNr
snro.getOrElse(0L)
highestSequenceNumbers.getOrElse(pid, 0L)
}

private def safeLongToInt(l: Long): Int =
Expand Down