Possibility to opt out from all_persistence_ids table, akka#784 (akka…

…#786)
chbatey · Jun 3, 2020 · 43f2cdf · 43f2cdf
1 parent 68aa9a6
commit 43f2cdf
Show file tree

Hide file tree

Showing 8 changed files with 43 additions and 15 deletions.
diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf
@@ -169,6 +169,15 @@ akka.persistence.cassandra {
     # It must not be off if deletes of events are used or have been used previously.
     # If this is set to off then delete attempts will fail with an IllegalArgumentException.
     support-deletes = on
+
+    # For applications that are not using persistenceIds or currentPersistenceIds queries
+    # this can be set to 'off', which will optimize the write of the first event for each
+    # persistent actor since the all_persistence_id table doesn't have to be populated.
+    # Note that the Cleanup and Reconcilation tools may also use persistenceIds queries
+    # and those will not work if this was disabled when the events were written. In
+    # that case the all_persistence_id table can be reconstructed with
+    # Reconcilation.rebuildAllPersistenceIds.
+    support-all-persistence-ids = on
   }
   //#journal
 

diff --git a/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournal.scala b/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournal.scala
@@ -197,7 +197,8 @@ import akka.stream.scaladsl.Source
       preparedWriteMessageWithMeta
       preparedSelectMessages
       preparedSelectHighestSequenceNr
-      preparedInsertIntoAllPersistenceIds
+      if (settings.journalSettings.supportAllPersistenceIds)
+        preparedInsertIntoAllPersistenceIds
       if (settings.journalSettings.supportDeletes) {
         preparedDeleteMessages
         preparedSelectDeletedTo
@@ -348,7 +349,7 @@ import akka.stream.scaladsl.Source
   private def writeMessages(atomicWrites: Seq[SerializedAtomicWrite]): Future[Unit] = {
     // insert into the all_persistence_ids table for the first event, used by persistenceIds query
     val allPersistenceId =
-      if (atomicWrites.head.payload.head.sequenceNr == 1L)
+      if (settings.journalSettings.supportAllPersistenceIds && atomicWrites.head.payload.head.sequenceNr == 1L)
         preparedInsertIntoAllPersistenceIds.map(_.bind(atomicWrites.head.persistenceId)).flatMap(execute(_))
       else
         FutureUnit

diff --git a/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournalStatements.scala b/core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournalStatements.scala
@@ -353,7 +353,12 @@ import akka.persistence.cassandra.FutureDone
         _ <- keyspace
         _ <- session.executeAsync(createTable).toScala
         _ <- session.executeAsync(createMetadataTable).toScala
-        _ <- session.executeAsync(createAllPersistenceIdsTable).toScala
+        _ <- {
+          if (settings.journalSettings.supportAllPersistenceIds)
+            session.executeAsync(createAllPersistenceIdsTable).toScala
+          else
+            FutureDone
+        }
         _ <- tagStatements
       } yield {
         session.setSchemaMetadataEnabled(null)

diff --git a/core/src/main/scala/akka/persistence/cassandra/journal/JournalSettings.scala b/core/src/main/scala/akka/persistence/cassandra/journal/JournalSettings.scala
@@ -49,6 +49,8 @@ import com.typesafe.config.Config
 
   val supportDeletes: Boolean = journalConfig.getBoolean("support-deletes")
 
+  val supportAllPersistenceIds: Boolean = journalConfig.getBoolean("support-all-persistence-ids")
+
   val coordinatedShutdownOnError: Boolean = config.getBoolean("coordinated-shutdown-on-error")
 
 }
diff --git a/core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala b/core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala
@@ -307,7 +307,10 @@ class CassandraReadJournal protected (
    */
   @InternalApi private[akka] def eventsByTagInternal(tag: String, offset: Offset): Source[UUIDPersistentRepr, NotUsed] =
     if (!eventsByTagSettings.eventsByTagEnabled)
-      Source.failed(new IllegalStateException("Events by tag queries are disabled"))
+      Source.failed(
+        new IllegalStateException(
+          "Events by tag queries are disabled with configuration " +
+          "events-by-tag.enabled=off"))
     else {
       try {
         val (fromOffset, usingOffset) = offsetToInternalOffset(offset)
@@ -696,14 +699,21 @@ class CassandraReadJournal protected (
     persistenceIds(None, "currentPersistenceIds")
 
   private def persistenceIds(refreshInterval: Option[FiniteDuration], name: String): Source[String, NotUsed] =
-    createSource[String, PreparedStatement](
-      preparedSelectAllPersistenceIds,
-      (s, ps) =>
-        Source
-          .fromGraph(new AllPersistenceIdsStage(refreshInterval, ps, s, querySettings.readProfile))
-          .withAttributes(ActorAttributes.dispatcher(querySettings.pluginDispatcher))
-          .mapMaterializedValue(_ => NotUsed)
-          .named(name))
+    if (!settings.journalSettings.supportAllPersistenceIds)
+      Source.failed(
+        new IllegalStateException(
+          "persistenceIds queries are disabled with configuration " +
+          "support-all-persistence-ids=off"))
+    else {
+      createSource[String, PreparedStatement](
+        preparedSelectAllPersistenceIds,
+        (s, ps) =>
+          Source
+            .fromGraph(new AllPersistenceIdsStage(refreshInterval, ps, s, querySettings.readProfile))
+            .withAttributes(ActorAttributes.dispatcher(querySettings.pluginDispatcher))
+            .mapMaterializedValue(_ => NotUsed)
+            .named(name))
+    }
 
   /**
    * INTERNAL API: Needed for migration to 1.0

diff --git a/core/src/test/scala/akka/persistence/cassandra/ReconnectSpec.scala b/core/src/test/scala/akka/persistence/cassandra/ReconnectSpec.scala
@@ -9,7 +9,7 @@ import java.io.File
 import akka.actor.{ ActorSystem, Props }
 import akka.persistence.cassandra.CassandraLifecycle.AwaitPersistenceInit
 import akka.persistence.cassandra.testkit.CassandraLauncher
-import akka.testkit.{ ImplicitSender, SocketUtil, TestKit, TestKitBase }
+import akka.testkit.{ ImplicitSender, SocketUtil, TestKit }
 import com.typesafe.config.ConfigFactory
 import org.scalatest.Suite
 import org.scalatest.concurrent.ScalaFutures

diff --git a/core/src/test/scala/akka/persistence/cassandra/journal/ManyActorsLoadSpec.scala b/core/src/test/scala/akka/persistence/cassandra/journal/ManyActorsLoadSpec.scala
@@ -19,6 +19,7 @@ object ManyActorsLoadSpec {
   val config = ConfigFactory.parseString(s"""
       akka.persistence.cassandra.journal.keyspace=ManyActorsLoadSpec
       akka.persistence.cassandra.events-by-tag.enabled = on
+      akka.persistence.cassandra.journal.support-all-persistence-ids = off
       # increase this to 3s when benchmarking
       akka.persistence.cassandra.events-by-tag.scanning-flush-interval = 1s
       #akka.persistence.cassandra.log-queries = on

diff --git a/core/src/test/scala/akka/persistence/cassandra/query/EventsByTagSpec.scala b/core/src/test/scala/akka/persistence/cassandra/query/EventsByTagSpec.scala
@@ -1408,14 +1408,14 @@ class EventsByTagDisabledSpec extends AbstractEventsByTagSpec(EventsByTagSpec.di
       val greenSrc = queries.currentEventsByTag(tag = "green", offset = NoOffset)
       val probe = greenSrc.runWith(TestSink.probe[Any])
       probe.request(1)
-      probe.expectError().getMessage shouldEqual "Events by tag queries are disabled"
+      probe.expectError().getMessage should include("Events by tag queries are disabled")
     }
 
     "fail live events by tag queries" in {
       val greenSrc = queries.eventsByTag(tag = "green", offset = NoOffset)
       val probe = greenSrc.runWith(TestSink.probe[Any])
       probe.request(1)
-      probe.expectError().getMessage shouldEqual "Events by tag queries are disabled"
+      probe.expectError().getMessage should include("Events by tag queries are disabled")
     }
 
     "allow recovery" in {