fix compile and delete condition

apache · Apr 19, 2024 · 2682cbd · 2682cbd
1 parent b30654b
commit 2682cbd
Show file tree

Hide file tree

Showing 4 changed files with 6 additions and 6 deletions.
diff --git a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/DefaultJournalDao.scala b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/DefaultJournalDao.scala
@@ -59,7 +59,7 @@ class DefaultJournalDao(
     new JournalQueries(profile, journalConfig.eventJournalTableConfiguration, journalConfig.eventTagTableConfiguration)
 
   override def delete(persistenceId: String, maxSequenceNr: Long): Future[Unit] = {
-    db.run(queries.delete(persistenceId, maxSequenceNr - 1)).map(_ => ())
+    db.run(queries.delete(persistenceId, maxSequenceNr)).map(_ => ())
   }
 
   override def highestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] = {

diff --git a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/JournalQueries.scala b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/JournalQueries.scala
@@ -51,7 +51,7 @@ class JournalQueries(
   }
 
   def delete(persistenceId: String, toSequenceNr: Long) = {
-    selectByPersistenceIdAndSequenceNr(persistenceId, toSequenceNr).delete
+    selectByPersistenceIdAndMaxSequenceNr(persistenceId, toSequenceNr).delete
   }
 
   private def _selectAllJournalForPersistenceId(persistenceId: Rep[String]) =
@@ -62,7 +62,7 @@ class JournalQueries(
 
   private def _selectByPersistenceIdAndMaxSequenceNr(persistenceId: Rep[String], maxSeqNr: Rep[Long]) =
     _selectByPersistenceId(persistenceId).filter(_.sequenceNumber <= maxSeqNr)
-  val selectByPersistenceIdAndSequenceNr = Compiled(_selectByPersistenceIdAndMaxSequenceNr _)
+  val selectByPersistenceIdAndMaxSequenceNr = Compiled(_selectByPersistenceIdAndMaxSequenceNr _)
 
   private def _highestSequenceNrForPersistenceId(persistenceId: Rep[String]): Rep[Option[Long]] =
     _selectAllJournalForPersistenceId(persistenceId).take(1).map(_.sequenceNumber).max

diff --git a/...main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/ByteArrayJournalDao.scala b/...main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/ByteArrayJournalDao.scala
@@ -87,7 +87,7 @@ trait BaseByteArrayJournalDao
   override def delete(persistenceId: String, maxSequenceNr: Long): Future[Unit] = {
     // We should keep journal record with highest sequence number in order to be compliant
     // with @see [[pekko.persistence.journal.JournalSpec]]
-    db.run(queries.delete(persistenceId, maxSequenceNr - 1)).map(_ => ())
+    db.run(queries.delete(persistenceId, maxSequenceNr)).map(_ => ())
   }
 
   def update(persistenceId: String, sequenceNr: Long, payload: AnyRef): Future[Done] = {

diff --git a/.../src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/JournalQueries.scala b/.../src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/legacy/JournalQueries.scala
@@ -31,7 +31,7 @@ class JournalQueries(val profile: JdbcProfile, override val journalTableCfg: Leg
     _selectByPersistenceId(persistenceId).sortBy(_.sequenceNumber.desc)
 
   def delete(persistenceId: String, toSequenceNr: Long) = {
-    selectByPersistenceIdAndSequenceNr(persistenceId, toSequenceNr).delete
+    selectByPersistenceIdAndMaxSequenceNr(persistenceId, toSequenceNr).delete
   }
 
   /**
@@ -49,7 +49,7 @@ class JournalQueries(val profile: JdbcProfile, override val journalTableCfg: Leg
 
   private def _selectByPersistenceIdAndMaxSequenceNr(persistenceId: Rep[String], maxSeqNr: Rep[Long]) =
     _selectByPersistenceId(persistenceId).filter(_.sequenceNumber <= maxSeqNr)
-  val selectByPersistenceIdAndSequenceNr = Compiled(_selectByPersistenceIdAndMaxSequenceNr _)
+  val selectByPersistenceIdAndMaxSequenceNr = Compiled(_selectByPersistenceIdAndMaxSequenceNr _)
 
   private def _highestSequenceNrForPersistenceId(persistenceId: Rep[String]): Rep[Option[Long]] =
     selectAllJournalForPersistenceId(persistenceId).take(1).map(_.sequenceNumber).max