You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2023/11/07 10:56:39 UTC
(incubator-pekko-persistence-cassandra) branch main updated: try not to use stored failed future results and therefore allow recovery (#106)
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-cassandra.git
The following commit(s) were added to refs/heads/main by this push:
new 87f7d1a try not to use stored failed future results and therefore allow recovery (#106)
87f7d1a is described below
commit 87f7d1acb81f8a32a3853c6f5dfbf804435ddd68
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Tue Nov 7 11:56:33 2023 +0100
try not to use stored failed future results and therefore allow recovery (#106)
* try not to stored failed future results and therefore allow recovery
* refactor to try to limit calls to eval function at startup
* Update LazyFutureEval.scala
* rename as RetryableFutureEvalSpec
---
.../persistence/cassandra/cleanup/Cleanup.scala | 11 +--
.../cassandra/journal/CassandraEventUpdate.scala | 22 +++---
.../cassandra/journal/CassandraJournal.scala | 79 ++++++++++++----------
.../cassandra/journal/CassandraTagRecovery.scala | 4 +-
.../persistence/cassandra/journal/TagWriters.scala | 8 +--
.../journal/TaggedPreparedStatements.scala | 40 ++++++-----
.../query/scaladsl/CassandraReadJournal.scala | 65 +++++++++---------
.../cassandra/util/RetryableFutureEval.scala | 56 +++++++++++++++
.../cassandra/util/RetryableFutureEvalSpec.scala | 62 +++++++++++++++++
9 files changed, 244 insertions(+), 103 deletions(-)
diff --git a/core/src/main/scala/org/apache/pekko/persistence/cassandra/cleanup/Cleanup.scala b/core/src/main/scala/org/apache/pekko/persistence/cassandra/cleanup/Cleanup.scala
index 158bd5d..e15e049 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/cassandra/cleanup/Cleanup.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/cassandra/cleanup/Cleanup.scala
@@ -31,6 +31,7 @@ import pekko.persistence.cassandra.journal.CassandraJournal
import pekko.persistence.cassandra.reconciler.Reconciliation
import pekko.persistence.cassandra.reconciler.ReconciliationSettings
import pekko.persistence.cassandra.snapshot.{ CassandraSnapshotStatements, CassandraSnapshotStore }
+import pekko.persistence.cassandra.util.RetryableFutureEval
import pekko.stream.connectors.cassandra.scaladsl.{ CassandraSession, CassandraSessionRegistry }
import pekko.stream.scaladsl.{ Sink, Source }
import pekko.util.Timeout
@@ -78,8 +79,10 @@ final class Cleanup(systemProvider: ClassicActorSystemProvider, settings: Cleanu
private lazy val pluginSettings = PluginSettings(system, system.settings.config.getConfig(pluginLocation))
private lazy val statements = new CassandraSnapshotStatements(pluginSettings.snapshotSettings)
- private lazy val selectLatestSnapshotsPs = session.prepare(statements.selectLatestSnapshotMeta)
- private lazy val selectAllSnapshotMetaPs = session.prepare(statements.selectAllSnapshotMeta)
+ private val selectLatestSnapshotsPs = RetryableFutureEval(() =>
+ session.prepare(statements.selectLatestSnapshotMeta))
+ private val selectAllSnapshotMetaPs = RetryableFutureEval(() =>
+ session.prepare(statements.selectAllSnapshotMeta))
if (dryRun) {
log.info("Cleanup running in dry run mode. No operations will be executed against the database, only logged")
@@ -148,7 +151,7 @@ final class Cleanup(systemProvider: ClassicActorSystemProvider, settings: Cleanu
keepAfterUnixTimestamp: Long): Future[Option[SnapshotMetadata]] = {
require(snapshotsToKeep >= 1, "must keep at least one snapshot")
require(keepAfterUnixTimestamp >= 0, "keepAfter must be greater than 0")
- selectAllSnapshotMetaPs
+ selectAllSnapshotMetaPs.futureResult()
.flatMap { ps =>
val allRows: Source[Row, NotUsed] = session.select(ps.bind(persistenceId))
allRows.zipWithIndex
@@ -179,7 +182,7 @@ final class Cleanup(systemProvider: ClassicActorSystemProvider, settings: Cleanu
*/
def deleteBeforeSnapshot(persistenceId: String, maxSnapshotsToKeep: Int): Future[Option[SnapshotMetadata]] = {
require(maxSnapshotsToKeep >= 1, "Must keep at least one snapshot")
- val snapshots: Future[immutable.Seq[Row]] = selectLatestSnapshotsPs.flatMap { ps =>
+ val snapshots: Future[immutable.Seq[Row]] = selectLatestSnapshotsPs.futureResult().flatMap { ps =>
session.select(ps.bind(persistenceId, maxSnapshotsToKeep: JInt)).runWith(Sink.seq)
}
snapshots.flatMap(rows => issueSnapshotDelete(persistenceId, maxSnapshotsToKeep, rows))
diff --git a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraEventUpdate.scala b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraEventUpdate.scala
index 9e5867c..baf7ed0 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraEventUpdate.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraEventUpdate.scala
@@ -19,6 +19,7 @@ import pekko.annotation.InternalApi
import pekko.event.LoggingAdapter
import pekko.persistence.cassandra.PluginSettings
import pekko.persistence.cassandra.journal.CassandraJournal.{ Serialized, TagPidSequenceNr }
+import pekko.persistence.cassandra.util.RetryableFutureEval
import pekko.stream.connectors.cassandra.scaladsl.CassandraSession
import pekko.util.ccompat.JavaConverters._
import com.datastax.oss.driver.api.core.cql.{ PreparedStatement, Row, Statement }
@@ -36,11 +37,14 @@ import java.lang.{ Long => JLong }
private def journalSettings = settings.journalSettings
private lazy val journalStatements = new CassandraJournalStatements(settings)
- lazy val psUpdateMessage: Future[PreparedStatement] = session.prepare(journalStatements.updateMessagePayloadAndTags)
- lazy val psSelectTagPidSequenceNr: Future[PreparedStatement] =
- session.prepare(journalStatements.selectTagPidSequenceNr)
- lazy val psUpdateTagView: Future[PreparedStatement] = session.prepare(journalStatements.updateMessagePayloadInTagView)
- lazy val psSelectMessages: Future[PreparedStatement] = session.prepare(journalStatements.selectMessages)
+ private val psUpdateMessage: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(journalStatements.updateMessagePayloadAndTags))
+ private val psSelectTagPidSequenceNr: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(journalStatements.selectTagPidSequenceNr))
+ private val psUpdateTagView: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(journalStatements.updateMessagePayloadInTagView))
+ private val psSelectMessages: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(journalStatements.selectMessages))
/**
* Update the given event in the messages table and the tag_views table.
@@ -50,7 +54,7 @@ import java.lang.{ Long => JLong }
def updateEvent(event: Serialized): Future[Done] =
for {
(partitionNr, existingTags) <- findEvent(event)
- psUM <- psUpdateMessage
+ psUM <- psUpdateMessage.futureResult()
e = event.copy(tags = existingTags) // do not allow updating of tags
_ <- session.executeWrite(prepareUpdate(psUM, e, partitionNr))
_ <- Future.traverse(existingTags) { tag =>
@@ -61,7 +65,7 @@ import java.lang.{ Long => JLong }
private def findEvent(s: Serialized): Future[(Long, Set[String])] = {
val firstPartition = partitionNr(s.sequenceNr, journalSettings.targetPartitionSize)
for {
- ps <- psSelectMessages
+ ps <- psSelectMessages.futureResult()
row <- findEvent(ps, s.persistenceId, s.sequenceNr, firstPartition)
} yield (row.getLong("partition_nr"), row.getSet[String]("tags", classOf[String]).asScala.toSet)
}
@@ -86,7 +90,7 @@ import java.lang.{ Long => JLong }
}
private def updateEventInTagViews(event: Serialized, tag: String): Future[Done] =
- psSelectTagPidSequenceNr
+ psSelectTagPidSequenceNr.futureResult()
.flatMap { ps =>
val bound = ps
.bind()
@@ -107,7 +111,7 @@ import java.lang.{ Long => JLong }
}
private def updateEventInTagViews(event: Serialized, tag: String, tagPidSequenceNr: TagPidSequenceNr): Future[Done] =
- psUpdateTagView.flatMap { ps =>
+ psUpdateTagView.futureResult().flatMap { ps =>
// primary key
val bound = ps
.bind()
diff --git a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraJournal.scala b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraJournal.scala
index 0fa4028..3891ef2 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraJournal.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraJournal.scala
@@ -29,6 +29,7 @@ import pekko.persistence._
import pekko.persistence.cassandra._
import pekko.persistence.cassandra.Extractors
import pekko.persistence.cassandra.query.scaladsl.CassandraReadJournal
+import pekko.persistence.cassandra.util.RetryableFutureEval
import pekko.persistence.journal.{ AsyncWriteJournal, Tagged }
import pekko.persistence.query.PersistenceQuery
import pekko.persistence.cassandra.journal.TagWriters.{ BulkTagWrite, TagWrite, TagWritersSession }
@@ -114,42 +115,45 @@ import scala.util.{ Failure, Success, Try }
private val tagRecovery: Option[CassandraTagRecovery] =
tagWrites.map(ref => new CassandraTagRecovery(context.system, session, settings, taggedPreparedStatements, ref))
- private lazy val preparedWriteMessage =
- session.prepare(statements.journalStatements.writeMessage(withMeta = false))
- private lazy val preparedSelectDeletedTo: Option[Future[PreparedStatement]] = {
+ private val preparedWriteMessage: RetryableFutureEval[PreparedStatement] =
+ RetryableFutureEval(() => session.prepare(statements.journalStatements.writeMessage(withMeta = false)))
+ private val preparedSelectDeletedTo: Option[RetryableFutureEval[PreparedStatement]] = {
if (settings.journalSettings.supportDeletes)
- Some(session.prepare(statements.journalStatements.selectDeletedTo))
+ Some(RetryableFutureEval(() => session.prepare(statements.journalStatements.selectDeletedTo)))
else
None
}
- private lazy val preparedSelectHighestSequenceNr: Future[PreparedStatement] =
- session.prepare(statements.journalStatements.selectHighestSequenceNr)
+ private val preparedSelectHighestSequenceNr: RetryableFutureEval[PreparedStatement] =
+ RetryableFutureEval(() => session.prepare(statements.journalStatements.selectHighestSequenceNr))
- private def deletesNotSupportedException: Future[PreparedStatement] =
- Future.failed(new IllegalArgumentException(s"Deletes not supported because config support-deletes=off"))
+ private val deletesNotSupportedException: RetryableFutureEval[PreparedStatement] =
+ RetryableFutureEval(() =>
+ Future.failed(new IllegalArgumentException(s"Deletes not supported because config support-deletes=off")))
- private lazy val preparedInsertDeletedTo: Future[PreparedStatement] = {
+ private val preparedInsertDeletedTo: RetryableFutureEval[PreparedStatement] = {
if (settings.journalSettings.supportDeletes)
- session.prepare(statements.journalStatements.insertDeletedTo)
+ RetryableFutureEval(() => session.prepare(statements.journalStatements.insertDeletedTo))
else
deletesNotSupportedException
}
- private lazy val preparedDeleteMessages: Future[PreparedStatement] = {
+ private val preparedDeleteMessages: RetryableFutureEval[PreparedStatement] = {
if (settings.journalSettings.supportDeletes) {
- session.serverMetaData.flatMap { meta =>
- session.prepare(statements.journalStatements.deleteMessages(meta.isVersion2 || settings.cosmosDb))
+ RetryableFutureEval { () =>
+ session.serverMetaData.flatMap { meta =>
+ session.prepare(statements.journalStatements.deleteMessages(meta.isVersion2 || settings.cosmosDb))
+ }
}
} else
deletesNotSupportedException
}
- private lazy val preparedInsertIntoAllPersistenceIds: Future[PreparedStatement] = {
- session.prepare(statements.journalStatements.insertIntoAllPersistenceIds)
+ private val preparedInsertIntoAllPersistenceIds: RetryableFutureEval[PreparedStatement] = {
+ RetryableFutureEval(() => session.prepare(statements.journalStatements.insertIntoAllPersistenceIds))
}
- private lazy val preparedWriteMessageWithMeta =
- session.prepare(statements.journalStatements.writeMessage(withMeta = true))
- private lazy val preparedSelectMessages =
- session.prepare(statements.journalStatements.selectMessages)
+ private val preparedWriteMessageWithMeta: RetryableFutureEval[PreparedStatement] =
+ RetryableFutureEval(() => session.prepare(statements.journalStatements.writeMessage(withMeta = true)))
+ private val preparedSelectMessages: RetryableFutureEval[PreparedStatement] =
+ RetryableFutureEval(() => session.prepare(statements.journalStatements.selectMessages))
private lazy val queries: CassandraReadJournal =
PersistenceQuery(context.system.asInstanceOf[ExtendedActorSystem])
@@ -199,16 +203,16 @@ import scala.util.{ Failure, Success, Try }
case CassandraJournal.Init =>
// try initialize early, to be prepared for first real request
- preparedWriteMessage
- preparedWriteMessageWithMeta
- preparedSelectMessages
- preparedSelectHighestSequenceNr
+ preparedWriteMessage.futureResult()
+ preparedWriteMessageWithMeta.futureResult()
+ preparedSelectMessages.futureResult()
+ preparedSelectHighestSequenceNr.futureResult()
if (settings.journalSettings.supportAllPersistenceIds)
- preparedInsertIntoAllPersistenceIds
+ preparedInsertIntoAllPersistenceIds.futureResult()
if (settings.journalSettings.supportDeletes) {
- preparedDeleteMessages
- preparedSelectDeletedTo
- preparedInsertDeletedTo
+ preparedDeleteMessages.futureResult()
+ preparedSelectDeletedTo.map(_.futureResult())
+ preparedInsertDeletedTo.futureResult()
}
queries.initialize()
@@ -365,7 +369,8 @@ import scala.util.{ Failure, Success, Try }
require(atomicWrites.head.payload.nonEmpty)
val allPersistenceId =
if (settings.journalSettings.supportAllPersistenceIds && atomicWrites.head.payload.head.sequenceNr == 1L)
- preparedInsertIntoAllPersistenceIds.map(_.bind(atomicWrites.head.persistenceId)).flatMap(execute(_))
+ preparedInsertIntoAllPersistenceIds.futureResult().map(_.bind(atomicWrites.head.persistenceId)).flatMap(
+ execute(_))
else
FutureUnit
@@ -402,8 +407,8 @@ import scala.util.{ Failure, Success, Try }
// then users doesn't have to alter table and add the new columns if they don't use
// the meta data feature
val stmt =
- if (m.meta.isDefined) preparedWriteMessageWithMeta
- else preparedWriteMessage
+ if (m.meta.isDefined) preparedWriteMessageWithMeta.futureResult()
+ else preparedWriteMessage.futureResult()
stmt.map { stmt =>
val bs = stmt
@@ -540,7 +545,7 @@ import scala.util.{ Failure, Success, Try }
val deleteResult =
Future.sequence((lowestPartition to highestPartition).map { partitionNr =>
val boundDeleteMessages =
- preparedDeleteMessages.map(_.bind(persistenceId, partitionNr: JLong, toSeqNr: JLong))
+ preparedDeleteMessages.futureResult().map(_.bind(persistenceId, partitionNr: JLong, toSeqNr: JLong))
boundDeleteMessages.flatMap(execute(_))
})
deleteResult.failed.foreach { e =>
@@ -564,7 +569,8 @@ import scala.util.{ Failure, Success, Try }
toSeqNr: TagPidSequenceNr): Future[Done] = {
def asyncDeleteMessages(partitionNr: TagPidSequenceNr, messageIds: Seq[MessageId]): Future[Unit] = {
val boundStatements = messageIds.map(mid =>
- preparedDeleteMessages.map(_.bind(mid.persistenceId, partitionNr: JLong, mid.sequenceNr: JLong)))
+ preparedDeleteMessages.futureResult().map(_.bind(mid.persistenceId, partitionNr: JLong,
+ mid.sequenceNr: JLong)))
Future.sequence(boundStatements).flatMap { stmts =>
executeBatch(batch => stmts.foldLeft(batch) { case (acc, next) => acc.add(next) })
}
@@ -608,7 +614,7 @@ import scala.util.{ Failure, Success, Try }
FutureUnit
} else {
val boundInsertDeletedTo =
- preparedInsertDeletedTo.map(_.bind(persistenceId, toSeqNr: JLong))
+ preparedInsertDeletedTo.futureResult().map(_.bind(persistenceId, toSeqNr: JLong))
boundInsertDeletedTo.flatMap(execute)
}
logicalDelete.flatMap(_ => physicalDelete(lowestPartition, highestPartition, toSeqNr))
@@ -643,7 +649,8 @@ import scala.util.{ Failure, Success, Try }
}
private def partitionInfo(persistenceId: String, partitionNr: Long, maxSequenceNr: Long): Future[PartitionInfo] = {
- val boundSelectHighestSequenceNr = preparedSelectHighestSequenceNr.map(_.bind(persistenceId, partitionNr: JLong))
+ val boundSelectHighestSequenceNr =
+ preparedSelectHighestSequenceNr.futureResult().map(_.bind(persistenceId, partitionNr: JLong))
boundSelectHighestSequenceNr
.flatMap(selectOne)
.map(row =>
@@ -656,7 +663,7 @@ import scala.util.{ Failure, Success, Try }
private def asyncHighestDeletedSequenceNumber(persistenceId: String): Future[Long] = {
preparedSelectDeletedTo match {
case Some(pstmt) =>
- val boundSelectDeletedTo = pstmt.map(_.bind(persistenceId))
+ val boundSelectDeletedTo = pstmt.futureResult().map(_.bind(persistenceId))
boundSelectDeletedTo.flatMap(selectOne).map(rowOption => rowOption.map(_.getLong("deleted_to")).getOrElse(0))
case None =>
Future.successful(0L)
@@ -669,7 +676,7 @@ import scala.util.{ Failure, Success, Try }
partitionSize: Long): Future[Long] = {
def find(currentPnr: Long, currentSnr: Long, foundEmptyPartition: Boolean): Future[Long] = {
// if every message has been deleted and thus no sequence_nr the driver gives us back 0 for "null" :(
- val boundSelectHighestSequenceNr = preparedSelectHighestSequenceNr.map(ps => {
+ val boundSelectHighestSequenceNr = preparedSelectHighestSequenceNr.futureResult().map(ps => {
val bound = ps.bind(persistenceId, currentPnr: JLong)
bound
diff --git a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraTagRecovery.scala b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraTagRecovery.scala
index 5252bbd..9898a59 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraTagRecovery.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/CassandraTagRecovery.scala
@@ -65,7 +65,7 @@ import pekko.stream.connectors.cassandra.scaladsl.CassandraSession
// The result set size will be the number of distinct tags that this pid has used, expecting
// that to be small (<10) so call to all should be safe
def lookupTagProgress(persistenceId: String)(implicit ec: ExecutionContext): Future[Map[Tag, TagProgress]] =
- SelectTagProgressForPersistenceId
+ SelectTagProgressForPersistenceId.futureResult()
.map(_.bind(persistenceId).setExecutionProfileName(settings.journalSettings.readProfile))
.flatMap(stmt => {
session.select(stmt).runWith(Sink.seq)
@@ -82,7 +82,7 @@ import pekko.stream.connectors.cassandra.scaladsl.CassandraSession
// or min tag scanning sequence number, and fix any tags. This recovers any tag writes that
// happened before the latest snapshot
def tagScanningStartingSequenceNr(persistenceId: String): Future[SequenceNr] =
- SelectTagScanningForPersistenceId
+ SelectTagScanningForPersistenceId.futureResult()
.map(_.bind(persistenceId).setExecutionProfileName(settings.journalSettings.readProfile))
.flatMap(session.selectOne)
.map {
diff --git a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TagWriters.scala b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TagWriters.scala
index 71534a1..e0b8cd1 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TagWriters.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TagWriters.scala
@@ -68,8 +68,8 @@ import scala.util.Try
val batch = new BatchStatementBuilder(BatchType.UNLOGGED)
batch.setExecutionProfileName(writeProfile)
val tagWritePSs = for {
- withMeta <- taggedPreparedStatements.WriteTagViewWithMeta
- withoutMeta <- taggedPreparedStatements.WriteTagViewWithoutMeta
+ withMeta <- taggedPreparedStatements.WriteTagViewWithMeta.futureResult()
+ withoutMeta <- taggedPreparedStatements.WriteTagViewWithoutMeta.futureResult()
} yield (withMeta, withoutMeta)
tagWritePSs
@@ -113,7 +113,7 @@ import scala.util.Try
def writeProgress(tag: Tag, persistenceId: String, seqNr: Long, tagPidSequenceNr: Long, offset: UUID)(
implicit ec: ExecutionContext): Future[Done] = {
- WriteTagProgress
+ WriteTagProgress.futureResult()
.map(ps =>
ps.bind(persistenceId, tag, seqNr: JLong, tagPidSequenceNr: JLong, offset)
.setExecutionProfileName(writeProfile))
@@ -394,7 +394,7 @@ import scala.util.Try
updates.take(maxPrint).mkString(",") + s" ...and ${updates.size - 20} more")
}
- tagWriterSession.taggedPreparedStatements.WriteTagScanning.foreach { ps =>
+ tagWriterSession.taggedPreparedStatements.WriteTagScanning.futureResult().foreach { ps =>
val startTime = System.nanoTime()
def writeTagScanningBatch(group: Seq[(String, Long)]): Future[Done] = {
diff --git a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TaggedPreparedStatements.scala b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TaggedPreparedStatements.scala
index 3c411f8..af13991 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TaggedPreparedStatements.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TaggedPreparedStatements.scala
@@ -14,9 +14,10 @@
package org.apache.pekko.persistence.cassandra.journal
import com.datastax.oss.driver.api.core.cql.PreparedStatement
-import scala.concurrent.{ ExecutionContext, Future }
+import scala.concurrent.{ ExecutionContext, Future }
import org.apache.pekko.annotation.InternalApi
+import org.apache.pekko.persistence.cassandra.util.RetryableFutureEval
/**
* INTERNAL API
@@ -26,22 +27,27 @@ import org.apache.pekko.annotation.InternalApi
prepare: String => Future[PreparedStatement])(implicit val ec: ExecutionContext) {
def init(): Unit = {
- WriteTagViewWithoutMeta
- WriteTagViewWithMeta
- WriteTagProgress
- SelectTagProgress
- SelectTagProgressForPersistenceId
- WriteTagScanning
- SelectTagScanningForPersistenceId
+ WriteTagViewWithoutMeta.futureResult()
+ WriteTagViewWithMeta.futureResult()
+ WriteTagProgress.futureResult()
+ SelectTagProgress.futureResult()
+ SelectTagProgressForPersistenceId.futureResult()
+ WriteTagScanning.futureResult()
+ SelectTagScanningForPersistenceId.futureResult()
}
- lazy val WriteTagViewWithoutMeta: Future[PreparedStatement] = prepare(statements.writeTags(false))
- lazy val WriteTagViewWithMeta: Future[PreparedStatement] = prepare(statements.writeTags(true))
- lazy val WriteTagProgress: Future[PreparedStatement] = prepare(statements.writeTagProgress)
- lazy val SelectTagProgress: Future[PreparedStatement] = prepare(statements.selectTagProgress)
- lazy val SelectTagProgressForPersistenceId: Future[PreparedStatement] =
- prepare(statements.selectTagProgressForPersistenceId)
- lazy val WriteTagScanning: Future[PreparedStatement] = prepare(statements.writeTagScanning)
- lazy val SelectTagScanningForPersistenceId: Future[PreparedStatement] =
- prepare(statements.selectTagScanningForPersistenceId)
+ val WriteTagViewWithoutMeta: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ prepare(statements.writeTags(false)))
+ val WriteTagViewWithMeta: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ prepare(statements.writeTags(true)))
+ val WriteTagProgress: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ prepare(statements.writeTagProgress))
+ val SelectTagProgress: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ prepare(statements.selectTagProgress))
+ val SelectTagProgressForPersistenceId: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ prepare(statements.selectTagProgressForPersistenceId))
+ val WriteTagScanning: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ prepare(statements.writeTagScanning))
+ val SelectTagScanningForPersistenceId: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ prepare(statements.selectTagScanningForPersistenceId))
}
diff --git a/core/src/main/scala/org/apache/pekko/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala b/core/src/main/scala/org/apache/pekko/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala
index 97c25fe..f413dea 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala
@@ -29,6 +29,7 @@ import pekko.persistence.cassandra.journal._
import pekko.persistence.cassandra.query.EventsByTagStage.TagStageSession
import pekko.persistence.cassandra.query._
import pekko.persistence.cassandra.query.scaladsl.CassandraReadJournal.EventByTagStatements
+import pekko.persistence.cassandra.util.RetryableFutureEval
import pekko.persistence.query._
import pekko.persistence.query.scaladsl._
import pekko.persistence.{ Persistence, PersistentRepr }
@@ -146,7 +147,7 @@ class CassandraReadJournal protected (
private val queryStatements: CassandraReadStatements =
new CassandraReadStatements {
- override def settings = CassandraReadJournal.this.settings
+ override def settings: PluginSettings = CassandraReadJournal.this.settings
}
/**
@@ -168,48 +169,49 @@ class CassandraReadJournal protected (
Future
.sequence(
List(
- preparedSelectDeletedTo,
- preparedSelectAllPersistenceIds,
- preparedSelectEventsByPersistenceId,
- preparedSelectFromTagViewWithUpperBound,
- preparedSelectTagSequenceNrs))
+ preparedSelectDeletedTo.futureResult(),
+ preparedSelectAllPersistenceIds.futureResult(),
+ preparedSelectEventsByPersistenceId.futureResult(),
+ preparedSelectFromTagViewWithUpperBound.futureResult(),
+ preparedSelectTagSequenceNrs.futureResult()))
.map(_ => Done)
- private lazy val preparedSelectEventsByPersistenceId: Future[PreparedStatement] =
- session.prepare(statements.journalStatements.selectMessages)
+ private val preparedSelectEventsByPersistenceId: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(statements.journalStatements.selectMessages))
- private lazy val preparedSelectDeletedTo: Future[PreparedStatement] =
- session.prepare(statements.journalStatements.selectDeletedTo)
+ private val preparedSelectDeletedTo: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(statements.journalStatements.selectDeletedTo))
- private lazy val preparedSelectAllPersistenceIds: Future[PreparedStatement] =
- session.prepare(queryStatements.selectAllPersistenceIds)
+ private val preparedSelectAllPersistenceIds: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(queryStatements.selectAllPersistenceIds))
- private lazy val preparedSelectDistinctPersistenceIds: Future[PreparedStatement] =
- session.prepare(queryStatements.selectDistinctPersistenceIds)
+ private val preparedSelectDistinctPersistenceIds: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(queryStatements.selectDistinctPersistenceIds))
- private lazy val preparedSelectFromTagViewWithUpperBound: Future[PreparedStatement] =
- session.prepare(queryStatements.selectEventsFromTagViewWithUpperBound)
+ private val preparedSelectFromTagViewWithUpperBound: RetryableFutureEval[PreparedStatement] =
+ RetryableFutureEval(() =>
+ session.prepare(queryStatements.selectEventsFromTagViewWithUpperBound))
- private lazy val preparedSelectTagSequenceNrs: Future[PreparedStatement] =
- session.prepare(queryStatements.selectTagSequenceNrs)
+ private val preparedSelectTagSequenceNrs: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(queryStatements.selectTagSequenceNrs))
- private lazy val preparedSelectHighestSequenceNr: Future[PreparedStatement] =
- session.prepare(statements.journalStatements.selectHighestSequenceNr)
+ private val preparedSelectHighestSequenceNr: RetryableFutureEval[PreparedStatement] = RetryableFutureEval(() =>
+ session.prepare(statements.journalStatements.selectHighestSequenceNr))
/**
* INTERNAL API
*/
@InternalApi private[pekko] def combinedEventsByPersistenceIdStmts: Future[CombinedEventsByPersistenceIdStmts] =
for {
- ps1 <- preparedSelectEventsByPersistenceId
- ps2 <- preparedSelectHighestSequenceNr
- ps3 <- preparedSelectDeletedTo
+ ps1 <- preparedSelectEventsByPersistenceId.futureResult()
+ ps2 <- preparedSelectHighestSequenceNr.futureResult()
+ ps3 <- preparedSelectDeletedTo.futureResult()
} yield CombinedEventsByPersistenceIdStmts(ps1, ps2, ps3)
/** INTERNAL API */
@InternalApi private[pekko] def combinedEventsByTagStmts: Future[EventByTagStatements] =
for {
- byTagWithUpper <- preparedSelectFromTagViewWithUpperBound
+ byTagWithUpper <- preparedSelectFromTagViewWithUpperBound.futureResult()
} yield EventByTagStatements(byTagWithUpper)
/**
@@ -393,11 +395,12 @@ class CassandraReadJournal protected (
* INTERNAL API
*/
@InternalApi
- private[pekko] val tagViewScanner: Future[TagViewSequenceNumberScanner] = preparedSelectTagSequenceNrs.map { ps =>
- new TagViewSequenceNumberScanner(
- TagViewSequenceNumberScanner.Session(session, ps, querySettings.readProfile),
- querySettings.pluginDispatcher)
- }
+ private[pekko] val tagViewScanner: Future[TagViewSequenceNumberScanner] =
+ preparedSelectTagSequenceNrs.futureResult().map { ps =>
+ new TagViewSequenceNumberScanner(
+ TagViewSequenceNumberScanner.Session(session, ps, querySettings.readProfile),
+ querySettings.pluginDispatcher)
+ }
/**
* INTERNAL API
@@ -726,7 +729,7 @@ class CassandraReadJournal protected (
"support-all-persistence-ids=off"))
else {
createSource[String, PreparedStatement](
- preparedSelectAllPersistenceIds,
+ preparedSelectAllPersistenceIds.futureResult(),
(s, ps) =>
Source
.fromGraph(new AllPersistenceIdsStage(refreshInterval, ps, s, querySettings.readProfile))
@@ -739,7 +742,7 @@ class CassandraReadJournal protected (
*/
@InternalApi private[pekko] def currentPersistenceIdsFromMessages(): Source[String, NotUsed] =
createSource[String, PreparedStatement](
- preparedSelectDistinctPersistenceIds,
+ preparedSelectDistinctPersistenceIds.futureResult(),
(s, ps) =>
Source
.fromGraph(new AllPersistenceIdsStage(None, ps, s, querySettings.readProfile))
diff --git a/core/src/main/scala/org/apache/pekko/persistence/cassandra/util/RetryableFutureEval.scala b/core/src/main/scala/org/apache/pekko/persistence/cassandra/util/RetryableFutureEval.scala
new file mode 100644
index 0000000..89f8834
--- /dev/null
+++ b/core/src/main/scala/org/apache/pekko/persistence/cassandra/util/RetryableFutureEval.scala
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.cassandra.util
+
+import org.apache.pekko.annotation.InternalApi
+
+import java.util.concurrent.atomic.AtomicReference
+import scala.concurrent.{ ExecutionContext, Future }
+import scala.util.Failure
+
+@InternalApi
+private[cassandra] object RetryableFutureEval {
+ def apply[T](evalFunction: () => Future[T]): RetryableFutureEval[T] = new RetryableFutureEval[T](evalFunction)
+}
+
+/**
+ * An internal utility class that lazily creates an instance of T but that can recover if the evalFunction fails.
+ *
+ * It does not guarantee that the evalFunction succeeds only once.
+ *
+ * This class is not recommended for non-Pekko usage.
+ */
+@InternalApi
+private[cassandra] class RetryableFutureEval[T](evalFunction: () => Future[T]) {
+ private val instance = new AtomicReference[Future[T]]()
+
+ def futureResult()(implicit ec: ExecutionContext): Future[T] = {
+ val f = instance.get()
+ if (f == null) {
+ val result = evalFunction()
+ instance.set(result)
+ result.onComplete {
+ case _: Failure[T] => instance.set(null)
+ case _ =>
+ }
+ result
+ } else {
+ f
+ }
+ }
+}
diff --git a/core/src/test/scala/org/apache/pekko/persistence/cassandra/util/RetryableFutureEvalSpec.scala b/core/src/test/scala/org/apache/pekko/persistence/cassandra/util/RetryableFutureEvalSpec.scala
new file mode 100644
index 0000000..7ae5124
--- /dev/null
+++ b/core/src/test/scala/org/apache/pekko/persistence/cassandra/util/RetryableFutureEvalSpec.scala
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.cassandra.util
+
+import org.scalatest.concurrent.Eventually
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+import java.util.UUID
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration.DurationInt
+import scala.concurrent.{ Await, Future }
+
+class RetryableFutureEvalSpec extends AnyWordSpecLike with Matchers with Eventually {
+ private val timeout = 5.seconds
+
+ "RetryableFutureEval" should {
+ "eval only once when there is just 1 thread" in {
+ val lazyEval = RetryableFutureEval(() => Future.successful(UUID.randomUUID()))
+ val uuid1 = Await.result(lazyEval.futureResult(), timeout)
+ uuid1 should not be null
+ Await.result(lazyEval.futureResult(), timeout) shouldEqual uuid1
+ }
+ "not store a failed eval result" in {
+ val eval = new EvalThatFailsFirstTime
+ val lazyEval = RetryableFutureEval(() => Future(eval.getResult()))
+ val res1 = Await.ready(lazyEval.futureResult(), timeout)
+ res1.value.get.isFailure shouldBe true
+ eventually {
+ // 2nd call should succeed but the cached result is removed asynchronously so we need to retry
+ val res2 = Await.ready(lazyEval.futureResult(), timeout)
+ res2.value.get.isSuccess shouldBe true
+ }
+ }
+ }
+
+ private class EvalThatFailsFirstTime {
+ private var initialized = false
+ def getResult(): Boolean = {
+ if (!initialized) {
+ initialized = true
+ throw new IllegalStateException("fails on first call")
+ }
+ initialized
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org