feat: Configuration of number of pub-sub topics, #252 (#341)

Author: patriknw

core/src/main/resources/reference.conf

@@ -34,6 +34,18 @@ akka.persistence.r2dbc {
       throughput-collect-interval = 10 seconds
     }
 
+    # Group the slices for an entity type into this number of topics. Most efficient is to use
+    # the same number as number of projection instances. If configured to less than the number of
+    # of projection instances the overhead is that events will be sent more than once and discarded
+    # on the destination side. If configured to more than the number of projection instances
+    # the events will only be sent once but there is a risk of exceeding the limits of number
+    # of topics that PubSub can handle (e.g. OversizedPayloadException).
+    # Must be between 1 and 1024 and a whole number divisor of 1024 (number of slices).
+    # This configuration can be changed in a rolling update, but there might be some events
+    # that are not delivered via the pub-sub path and instead delivered later by the queries.
+    # This configuration cannot be defined per journal, but is global for the ActorSystem.
+    publish-events-number-of-topics = 128
+
     # replay filter not needed for this plugin
     replay-filter.mode = off
   }

core/src/main/scala/akka/persistence/r2dbc/internal/PubSub.scala

@@ -49,6 +49,13 @@ import org.slf4j.LoggerFactory
 
   private val settings = new PublishEventsDynamicSettings(
     system.settings.config.getConfig("akka.persistence.r2dbc.journal.publish-events-dynamic"))
+
+  private val sliceRanges = {
+    val numberOfTopics = system.settings.config.getInt("akka.persistence.r2dbc.journal.publish-events-number-of-topics")
+    persistenceExt.sliceRanges(numberOfTopics)
+  }
+  private val sliceRangeLookup = new ConcurrentHashMap[Int, Range]
+
   private val throughputCollectIntervalMillis = settings.throughputCollectInterval.toMillis
   private val throughputThreshold = settings.throughputThreshold.toDouble
   private val throughputSampler =
@@ -64,8 +71,23 @@ import org.slf4j.LoggerFactory
       .narrow[Topic.Command[EventEnvelope[Event]]]
   }
 
-  private def topicName(entityType: String, slice: Int): String =
-    URLEncoder.encode(s"r2dbc-$entityType-$slice", StandardCharsets.UTF_8.name())
+  def eventTopics[Event](
+      entityType: String,
+      minSlice: Int,
+      maxSlice: Int): Set[ActorRef[Topic.Command[EventEnvelope[Event]]]] = {
+    (minSlice to maxSlice).map(eventTopic[Event](entityType, _)).toSet
+  }
+
+  private def topicName(entityType: String, slice: Int): String = {
+    val range = sliceRangeLookup.computeIfAbsent(
+      slice,
+      _ =>
+        sliceRanges
+          .find(_.contains(slice))
+          .getOrElse(throw new IllegalArgumentException(s"Slice [$slice] not found in " +
+          s"slice ranges [${sliceRanges.mkString(", ")}]")))
+    URLEncoder.encode(s"r2dbc-$entityType-${range.min}-${range.max}", StandardCharsets.UTF_8.name())
+  }
 
   def publish(pr: PersistentRepr, timestamp: Instant): Unit = {
 

core/src/main/scala/akka/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala

@@ -162,11 +162,15 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
             bufferSize = settings.querySettings.bufferSize,
             overflowStrategy = OverflowStrategy.dropNew)
           .mapMaterializedValue { ref =>
-            (minSlice to maxSlice).foreach { slice =>
+            pubSub.eventTopics[Event](entityType, minSlice, maxSlice).foreach { topic =>
               import akka.actor.typed.scaladsl.adapter._
-              pubSub.eventTopic(entityType, slice) ! Topic.Subscribe(ref.toTyped[EventEnvelope[Event]])
+              topic ! Topic.Subscribe(ref.toTyped[EventEnvelope[Event]])
             }
           }
+          .filter { env =>
+            val slice = sliceForPersistenceId(env.persistenceId)
+            minSlice <= slice && slice <= maxSlice
+          }
       dbSource
         .mergePrioritized(pubSubSource, leftPriority = 1, rightPriority = 10)
         .via(deduplicate(settings.querySettings.deduplicateCapacity))

core/src/test/scala/akka/persistence/r2dbc/query/EventsBySlicePubSubSpec.scala

@@ -4,6 +4,7 @@
 
 package akka.persistence.r2dbc.query
 
+import scala.collection.immutable
 import java.time.Instant
 
 import scala.concurrent.Await
@@ -15,6 +16,7 @@ import akka.actor.testkit.typed.scaladsl.LoggingTestKit
 import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
 import akka.actor.typed.ActorSystem
 import akka.actor.typed.internal.pubsub.TopicImpl
+import akka.persistence.Persistence
 import akka.persistence.query.NoOffset
 import akka.persistence.query.PersistenceQuery
 import akka.persistence.query.TimestampOffset
@@ -44,6 +46,7 @@ object EventsBySlicePubSubSpec {
     .parseString("""
     akka.persistence.r2dbc {
       journal.publish-events = on
+      #journal.publish-events-number-of-topics = 4
       journal.publish-events-dynamic {
         throughput-threshold = 50
         throughput-collect-interval = 1 second
@@ -234,6 +237,41 @@ class EventsBySlicePubSubSpec
 
     }
 
+    "group slices into topics" in new Setup {
+
+      val numberOfTopics =
+        typedSystem.settings.config.getInt("akka.persistence.r2dbc.journal.publish-events-number-of-topics")
+      //
+      val querySliceRanges = Persistence(typedSystem).sliceRanges(numberOfTopics * 2)
+      val queries: immutable.IndexedSeq[TestSubscriber.Probe[EventEnvelope[String]]] = {
+        querySliceRanges.map { range =>
+          query.eventsBySlices[String](entityType, range.min, range.max, NoOffset).runWith(sinkProbe).request(100)
+        }
+      }
+
+      val topicStatsProbe = createTestProbe[TopicImpl.TopicStats]()
+      eventually {
+        (0 until 1024).foreach { i =>
+          withClue(s"slice $i: ") {
+            PubSub(typedSystem).eventTopic[String](entityType, i) ! TopicImpl.GetTopicStats(topicStatsProbe.ref)
+            topicStatsProbe.receiveMessage().localSubscriberCount shouldBe 2
+          }
+        }
+      }
+
+      for (i <- 1 to 10) {
+        persister ! PersistWithAck(s"e-$i", probe.ref)
+        probe.expectMessage(Done)
+      }
+
+      for (i <- 1 to 10) {
+        val queryIndex = querySliceRanges.indexOf(querySliceRanges.find(_.contains(slice)).get)
+        queries(queryIndex).expectNext().event shouldBe s"e-$i"
+      }
+
+      queries.foreach(_.cancel())
+    }
+
   }
 
 }

docs/src/main/paradox/query.md

@@ -112,6 +112,8 @@ Disable publishing of events with configuration:
 akka.persistence.r2dbc.journal.publish-events = off
 ```
 
+If you use many queries or Projection instances you should consider adjusting the `akka.persistence.r2dbc.journal.publish-events-number-of-topics` configuration, see @ref:[Configuration](#configuration).
+
 ## Durable state queries
 
 @apidoc[R2dbcDurableStateStore] implements the following @extref:[Persistence Queries](akka:durable-state/persistence-query.html):