This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 05b300dafb Port scalable fan-out persistence query abstraction from
Akka (#3277)
05b300dafb is described below
commit 05b300dafb7330fad8a47247f562f061cb85d4f8
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Jul 1 11:31:20 2026 +0100
Port scalable fan-out persistence query abstraction from Akka (#3277)
* Port scalable fan-out persistence query abstraction from Akka
manual changes
more
more
more
more
compile issues
Update EventsBySliceFirehoseSpec.scala
build issues
Update EventsBySliceFirehoseSpec.scala
unused imports
remove new cluster-sharding files
revert
* javadoc
* scalafmt
---------
Co-authored-by: copilot-swe-agent[bot]
<[email protected]>
---
docs/src/main/paradox/persistence-query.md | 4 +-
.../src/main/resources/reference.conf | 53 ++
.../EventsBySliceFirehoseReadJournalProvider.scala | 33 +
.../typed/internal/EventsBySliceFirehose.scala | 839 +++++++++++++++++++++
...sByPersistenceIdStartingFromSnapshotQuery.scala | 39 +
...ntEventsBySliceStartingFromSnapshotsQuery.scala | 52 ++
...sByPersistenceIdStartingFromSnapshotQuery.scala | 39 +
.../typed/javadsl/EventsBySliceFirehoseQuery.scala | 95 +++
.../query/typed/javadsl/EventsBySliceQuery.scala | 2 +
.../EventsBySliceStartingFromSnapshotsQuery.scala | 54 ++
...sByPersistenceIdStartingFromSnapshotQuery.scala | 39 +
...ntEventsBySliceStartingFromSnapshotsQuery.scala | 53 ++
...sByPersistenceIdStartingFromSnapshotQuery.scala | 39 +
.../scaladsl/EventsBySliceFirehoseQuery.scala | 119 +++
.../query/typed/scaladsl/EventsBySliceQuery.scala | 2 +
.../EventsBySliceStartingFromSnapshotsQuery.scala | 55 ++
.../apache/pekko/persistence/query/TestClock.scala | 59 ++
.../typed/internal/EventsBySliceFirehoseSpec.scala | 440 +++++++++++
18 files changed, 2015 insertions(+), 1 deletion(-)
diff --git a/docs/src/main/paradox/persistence-query.md
b/docs/src/main/paradox/persistence-query.md
index 801dfe5509..44f2de0bc1 100644
--- a/docs/src/main/paradox/persistence-query.md
+++ b/docs/src/main/paradox/persistence-query.md
@@ -168,7 +168,9 @@ If your usage does not require a live stream, you can use
the @apidoc[currentEve
Query events for given entity type and slices. A slice is deterministically
defined based on the persistence id.
The purpose is to evenly distribute all persistence ids over the slices.
-See @apidoc[persistence.query.typed.*.EventsBySliceQuery] and
@apidoc[persistence.query.typed.*.CurrentEventsBySliceQuery].
+See @apidoc[pekko.persistence.query.typed.*.EventsBySliceQuery] and
@apidoc[pekko.persistence.query.typed.*.CurrentEventsBySliceQuery].
+
+A variation of these are
@apidoc[pekko.persistence.query.typed.*.EventsBySliceStartingFromSnapshotsQuery]
and
@apidoc[pekko.persistence.query.typed.*.CurrentEventsBySliceStartingFromSnapshotsQuery].
### Materialized values of queries
diff --git a/persistence-query/src/main/resources/reference.conf
b/persistence-query/src/main/resources/reference.conf
index 2dfff044c1..c9bb5af9ee 100644
--- a/persistence-query/src/main/resources/reference.conf
+++ b/persistence-query/src/main/resources/reference.conf
@@ -30,6 +30,59 @@ pekko.persistence.query.journal.leveldb {
}
#//#query-leveldb
+pekko.persistence.query.events-by-slice-firehose {
+ class =
"org.apache.pekko.persistence.query.typed.EventsBySliceFirehoseReadJournalProvider"
+
+ # The identifier (config path) of the underlying EventsBySlice query plugin.
+ # This must be defined by the application.
+ delegate-query-plugin-id = ""
+
+ # Buffer size of the BroadcastHub that will fan out the shared firehose
stream
+ # to attached consumer streams. If too small, some consumers may slow down
other
+ # consumers before the slow consumers have been aborted. If too large, it
will
+ # use more memory by holding more events in the buffer memory.
+ # Must be a power of two and less than 4096.
+ broadcast-buffer-size = 256
+
+ # The shared firehose stream will be closed after this timeout when all
consumer
+ # streams have been closed. It will be started again when new consumers
attach,
+ # but there is some overhead of stopping and starting so it's good to keep it
+ # around for a while. For example, keep around long enough to cover
Projection
+ # restarts.
+ firehose-linger-timeout = 40s
+
+ # When the catchup stream for a new consumer has caught up to the shared
firehose
+ # stream events will be retrieved from both during this time of overlap. The
reason
+ # is to ensure that no events are missed when switching over. After that,
+ # the catchup stream will be closed. Time is based on the timestamps of the
+ # EventEnvelope.
+ catchup-overlap = 10s
+
+ # Approximately number of entries of the deduplication cache.
+ # During the overlap period events will be deduplicated by keeping track of
emitted
+ # persistenceId and seqNr.
+ deduplication-capacity = 10000
+
+ # Slow consumers are detected and aborted by a background task that is
running
+ # with this interval. Should be less than `slow-consumer-lag-threshold`.
+ slow-consumer-reaper-interval = 2s
+
+ # Slow consumer candidates are determined if the fastest consumer has a lag
greater
+ # than this duration, and the slow consumer is behind the fastest consumer
by more
+ # than half of the `broadcast-buffer-size`.
+ # Slow consumers are then confirmed to be slow if they stay as such for at
+ # least `abort-slow-consumer-after`.
+ slow-consumer-lag-threshold = 5s
+
+ # See `slow-consumer-lag-threshold`.
+ # This duration is based on wall clock time.
+ abort-slow-consumer-after = 2s
+
+ # Provide a higher level of details in the debug logs, often per event. Be
careful about enabling
+ # in production systems.
+ verbose-debug-logging = off
+}
+
pekko.actor {
serializers {
pekko-persistence-query =
"org.apache.pekko.persistence.query.internal.QuerySerializer"
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/EventsBySliceFirehoseReadJournalProvider.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/EventsBySliceFirehoseReadJournalProvider.scala
new file mode 100644
index 0000000000..19f4e7db51
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/EventsBySliceFirehoseReadJournalProvider.scala
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed
+import com.typesafe.config.Config
+
+import org.apache.pekko
+import pekko.actor.ExtendedActorSystem
+import pekko.persistence.query.ReadJournalProvider
+
+final class EventsBySliceFirehoseReadJournalProvider(system:
ExtendedActorSystem, config: Config, cfgPath: String)
+ extends ReadJournalProvider {
+
+ private lazy val scaladslReadJournalInstance:
scaladsl.EventsBySliceFirehoseQuery =
+ new scaladsl.EventsBySliceFirehoseQuery(system, config, cfgPath)
+
+ override def scaladslReadJournal(): scaladsl.EventsBySliceFirehoseQuery =
scaladslReadJournalInstance
+
+ private lazy val javadslReadJournalInstance =
+ new javadsl.EventsBySliceFirehoseQuery(new
scaladsl.EventsBySliceFirehoseQuery(system, config, cfgPath))
+
+ override def javadslReadJournal(): javadsl.EventsBySliceFirehoseQuery =
javadslReadJournalInstance
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/internal/EventsBySliceFirehose.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/internal/EventsBySliceFirehose.scala
new file mode 100644
index 0000000000..94728c1334
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/internal/EventsBySliceFirehose.scala
@@ -0,0 +1,839 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.internal
+
+import java.time.{ Duration => JDuration }
+import java.time.Instant
+import java.util.UUID
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.annotation.{ nowarn, tailrec }
+import scala.collection.mutable
+import scala.concurrent.duration._
+import scala.jdk.DurationConverters._
+import scala.util.control.NoStackTrace
+
+import com.typesafe.config.Config
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.actor.ActorSystem
+import pekko.actor.Cancellable
+import pekko.actor.ClassicActorSystemProvider
+import pekko.actor.ExtendedActorSystem
+import pekko.actor.Extension
+import pekko.actor.ExtensionId
+import pekko.actor.ExtensionIdProvider
+import pekko.annotation.InternalApi
+import pekko.event.Logging
+import pekko.event.LoggingAdapter
+import pekko.persistence.query.Offset
+import pekko.persistence.query.PersistenceQuery
+import pekko.persistence.query.TimestampOffset
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.persistence.query.typed.scaladsl.EventsBySliceQuery
+import
pekko.persistence.query.typed.scaladsl.EventsBySliceStartingFromSnapshotsQuery
+import pekko.stream.Attributes
+import pekko.stream.FanInShape2
+import pekko.stream.FlowShape
+import pekko.stream.Inlet
+import pekko.stream.KillSwitch
+import pekko.stream.KillSwitches
+import pekko.stream.Outlet
+import pekko.stream.scaladsl.BroadcastHub
+import pekko.stream.scaladsl.GraphDSL
+import pekko.stream.scaladsl.Keep
+import pekko.stream.scaladsl.Source
+import pekko.stream.stage.GraphStage
+import pekko.stream.stage.GraphStageLogic
+import pekko.stream.stage.InHandler
+import pekko.stream.stage.OutHandler
+import pekko.stream.stage.StageLogging
+import pekko.util.OptionVal
+
+/**
+ * INTERNAL API
+ *
+ * The purpose is to share the stream of events from the database and fan out
to connected consumer
+ * streams. Thereby less queries and loading of events from the database. The
shared stream is called
+ * the firehose stream.
+ *
+ * The fan out of the firehose stream is via a `BroadcastHub` that consumer
streams dynamically attach to.
+ *
+ * A new consumer starts a catchup stream since the start offset typically is
behind the live
+ * firehose stream. In the beginning it will emit events only from the catchup
stream. Offset progress
+ * for the firehose stream is tracked and when the catchup stream has caught
up with the firehose
+ * stream it will switch over to emitting from firehose stream and close the
catchup stream. During
+ * an overlap period of time it will use events from both catchup and firehose
streams to make sure
+ * that no events are missed. During this overlap time there is best effort
deduplication.
+ *
+ * The `BroadcastHub` has a limited buffer that holds events between the
slowest and fastest consumer.
+ * When the buffer is full the fastest consumer can't progress faster than the
slowest. Short periods of
+ * slow down can be fine, but after a while the slow consumers are detected
and aborted. They have to
+ * connect again and try catching up, but without slowing down other streams.
+ *
+ * The firehose stream is started on demand when the first consumer is
attaching. It will be stopped
+ * when the last consumer is stopped, but it stays around for a while to make
it more efficient for
+ * new or restarted consumers to attach again.
+ */
+@InternalApi private[pekko] object EventsBySliceFirehose
+ extends ExtensionId[EventsBySliceFirehose]
+ with ExtensionIdProvider {
+
+ override def get(system: ActorSystem): EventsBySliceFirehose =
super.get(system)
+
+ override def get(system: ClassicActorSystemProvider): EventsBySliceFirehose
= super.get(system)
+
+ override def lookup = EventsBySliceFirehose
+
+ override def createExtension(system: ExtendedActorSystem):
EventsBySliceFirehose =
+ new EventsBySliceFirehose(system)
+
+ object Settings {
+ def apply(system: ActorSystem, pluginId: String): Settings =
+ apply(system.settings.config.getConfig(pluginId))
+
+ def apply(config: Config): Settings =
+ Settings(
+ delegateQueryPluginId = delegateQueryPluginId(config),
+ broadcastBufferSize = config.getInt("broadcast-buffer-size"),
+ firehoseLingerTimeout =
config.getDuration("firehose-linger-timeout").toScala,
+ catchupOverlap = config.getDuration("catchup-overlap"),
+ deduplicationCapacity = config.getInt("deduplication-capacity"),
+ slowConsumerReaperInterval =
config.getDuration("slow-consumer-reaper-interval").toScala,
+ slowConsumerLagThreshold =
config.getDuration("slow-consumer-lag-threshold"),
+ abortSlowConsumerAfter =
config.getDuration("abort-slow-consumer-after"),
+ verboseLogging = config.getBoolean("verbose-debug-logging"))
+
+ def delegateQueryPluginId(config: Config): String =
+ config.getString("delegate-query-plugin-id")
+ }
+
+ final case class Settings(
+ delegateQueryPluginId: String,
+ broadcastBufferSize: Int,
+ firehoseLingerTimeout: FiniteDuration,
+ catchupOverlap: JDuration,
+ deduplicationCapacity: Int,
+ slowConsumerReaperInterval: FiniteDuration,
+ slowConsumerLagThreshold: JDuration,
+ abortSlowConsumerAfter: JDuration,
+ verboseLogging: Boolean) {
+ require(
+ delegateQueryPluginId != null && delegateQueryPluginId.nonEmpty,
+ "Configuration of delegate-query-plugin-id must defined.")
+ }
+
+ final class SlowConsumerException(message: String) extends
RuntimeException(message) with NoStackTrace
+
+ final case class FirehoseKey(pluginId: String, entityType: String,
sliceRange: Range)
+
+ final case class ConsumerTracking(
+ consumerId: String,
+ history: Vector[TimestampOffset],
+ firehoseOnly: Boolean,
+ consumerKillSwitch: KillSwitch,
+ slowConsumerCandidate: Option[Instant]) {
+ def offsetTimestamp: Instant =
+ if (history.isEmpty) Instant.EPOCH
+ else history.last.timestamp
+ }
+
+ final class Firehose(
+ val firehoseKey: FirehoseKey,
+ val settings: Settings,
+ val firehoseHub: Source[EventEnvelope[Any], NotUsed],
+ firehoseKillSwitch: KillSwitch,
+ log: LoggingAdapter) {
+
+ val consumerTracking: ConcurrentHashMap[String, ConsumerTracking] = new
ConcurrentHashMap
+ @volatile private var firehoseIsShutdown = false
+
+ private def entityType = firehoseKey.entityType
+ private val sliceRangeStr =
s"${firehoseKey.sliceRange.min}-${firehoseKey.sliceRange.max}"
+
+ private def consumerTrackingValues(): Vector[ConsumerTracking] = {
+ import scala.jdk.CollectionConverters._
+ consumerTracking.values.iterator.asScala.filter(h => h.history.nonEmpty
&& h.firehoseOnly).toVector
+ }
+
+ def consumerStarted(consumerId: String, consumerKillSwitch: KillSwitch):
Unit = {
+ log.debug("Firehose entityType [{}] sliceRange [{}] consumer [{}]
started", entityType, sliceRangeStr, consumerId)
+ consumerTracking.putIfAbsent(
+ consumerId,
+ ConsumerTracking(consumerId, history = Vector.empty, firehoseOnly =
false, consumerKillSwitch, None))
+ }
+
+ def consumerTerminated(consumerId: String): Int = {
+ log.debug(
+ "Firehose entityType [{}] sliceRange [{}] consumer [{}] terminated",
+ entityType,
+ sliceRangeStr,
+ consumerId)
+ consumerTracking.remove(consumerId)
+ consumerTracking.size
+ }
+
+ def shutdownFirehoseIfNoConsumers(): Boolean = {
+ if (consumerTracking.isEmpty) {
+ log.debug("Firehose entityType [{}] sliceRange [{}] is shutting down,
no consumers", entityType, sliceRangeStr)
+ firehoseIsShutdown = true
+ firehoseKillSwitch.shutdown()
+ true
+ } else
+ false
+ }
+
+ def isShutdown: Boolean =
+ firehoseIsShutdown
+
+ def updateConsumerTracking(
+ consumerId: String,
+ now: Instant,
+ offset: TimestampOffset,
+ consumerKillSwitch: KillSwitch): Unit = {
+
+ val newTracking = consumerTracking.compute(
+ consumerId,
+ (_, existing) => {
+ if (existing == null)
+ ConsumerTracking(consumerId, history = Vector(offset),
firehoseOnly = false, consumerKillSwitch, None)
+ else {
+ val newHistory =
+ if (existing.history.size <= settings.broadcastBufferSize)
+ existing.history :+ offset
+ else
+ existing.history.tail :+ offset // drop one, add one
+ existing.copy(history = newHistory)
+ }
+ })
+ logUpdateConsumerTracking(consumerId, now, newTracking)
+ }
+
+ private def logUpdateConsumerTracking(consumerId: String, now: Instant,
tracking: ConsumerTracking): Unit = {
+ if (settings.verboseLogging && log.isDebugEnabled) {
+ val trackingValues = consumerTrackingValues()
+ if (trackingValues.size > 1) {
+ val slowestConsumer = trackingValues.minBy(_.offsetTimestamp)
+ val fastestConsumer = trackingValues.maxBy(_.offsetTimestamp)
+ val behind = elementsBehind(fastestConsumer.history,
slowestConsumer.history)
+ if (behind > 0) {
+ val diffSlowestFastestsMillis =
fastestConsumer.offsetTimestamp.toEpochMilli -
+ slowestConsumer.offsetTimestamp.toEpochMilli
+ val fastestLagMillis = now.toEpochMilli -
fastestConsumer.offsetTimestamp.toEpochMilli
+
+ val diffFastest = fastestConsumer.offsetTimestamp.toEpochMilli -
tracking.offsetTimestamp.toEpochMilli
+ val diffFastestStr =
+ if (diffFastest > 0) s"behind fastest [$diffFastest] ms"
+ else if (diffFastest < 0) s"ahead of fastest [$diffFastest] ms"
// not possible
+ else "same as fastest"
+ val diffSlowest = slowestConsumer.offsetTimestamp.toEpochMilli -
tracking.offsetTimestamp.toEpochMilli
+ val diffSlowestStr =
+ if (diffSlowest > 0) s"behind slowest [$diffSlowest] ms" // not
possible
+ else if (diffSlowest < 0) s"ahead of slowest [${-diffSlowest}]
ms"
+ else "same as slowest"
+ val consumerBehind = elementsBehind(fastestConsumer.history,
tracking.history)
+
+ log.debug(
+ s"Firehose entityType [$entityType] sliceRange [$sliceRangeStr]
updateConsumerTracking [$consumerId], " +
+ s"behind [$consumerBehind] events from fastest, " +
+ s"$diffFastestStr, $diffSlowestStr, " +
+ s"slowest [${slowestConsumer.consumerId}] is behind [$behind] " +
+ s"events from fastest [${fastestConsumer.consumerId}],
[$diffSlowestFastestsMillis] ms. " +
+ s"Consumer lag of fastest [$fastestLagMillis] ms.")
+ }
+ }
+ }
+ }
+
+ def detectSlowConsumers(now: Instant): Unit = {
+ val trackingValues = consumerTrackingValues()
+ if (trackingValues.size > 1) {
+ val slowestConsumer = trackingValues.minBy(_.offsetTimestamp)
+ val fastestConsumer = trackingValues.maxBy(_.offsetTimestamp)
+
+ val behind = elementsBehind(fastestConsumer.history,
slowestConsumer.history)
+ val fastestLagMillis = now.toEpochMilli -
fastestConsumer.offsetTimestamp.toEpochMilli
+
+ if (behind >= settings.broadcastBufferSize / 2 &&
+ fastestLagMillis > settings.slowConsumerLagThreshold.toMillis) {
+ logDetectSlowConsumers(trackingValues, slowestConsumer,
fastestConsumer, behind, fastestLagMillis)
+
+ val changedConsumerTrackingValues = trackingValues.flatMap {
tracking =>
+ val consumerBehind = elementsBehind(fastestConsumer.history,
tracking.history)
+ if (consumerBehind >= settings.broadcastBufferSize / 2) {
+ if (tracking.slowConsumerCandidate.isDefined)
+ None // keep original
+ else
+ Some(tracking.copy(slowConsumerCandidate = Some(now)))
+ } else if (tracking.slowConsumerCandidate.isDefined) {
+ Some(tracking.copy(slowConsumerCandidate = None)) // not slow
any more
+ } else {
+ None
+ }
+ }
+
+ changedConsumerTrackingValues.foreach { tracking =>
+ consumerTracking.computeIfPresent(
+ tracking.consumerId,
+ (_, existing) => existing.copy(slowConsumerCandidate =
tracking.slowConsumerCandidate))
+ }
+
+ val newTrackingValues = consumerTrackingValues()
+
+ val confirmedSlowConsumers = newTrackingValues.filter {
+ _.slowConsumerCandidate.exists(isDurationGreaterThan(_, now,
settings.abortSlowConsumerAfter))
+ }
+
+ if (confirmedSlowConsumers.nonEmpty) {
+ if (log.isInfoEnabled) {
+ val behindMillis = fastestConsumer.offsetTimestamp.toEpochMilli -
+ confirmedSlowConsumers
+ .maxBy(_.offsetTimestamp)
+ .offsetTimestamp
+ .toEpochMilli
+ log.info(
+ s"Firehose entityType [$entityType] sliceRange
[$sliceRangeStr], [${confirmedSlowConsumers.size}] " +
+ s"slow consumers are aborted
[${confirmedSlowConsumers.map(_.consumerId).mkString(", ")}], " +
+ s"behind by [$behind] events [$behindMillis] ms.")
+ }
+
+ confirmedSlowConsumers.foreach { tracking =>
+ tracking.consumerKillSwitch.abort(
+ new SlowConsumerException(s"Consumer [${tracking.consumerId}]
is too slow."))
+ }
+ }
+
+ } else if (behind < 0) {
+ // can happen if fastest is a new consumer, but ignoring should be ok
+ log.debug(
+ s"Firehose entityType [{}] sliceRange [{}] missing history for
fastest consumer [{}] corresponding to slowest consumer [{}]",
+ entityType,
+ sliceRangeStr,
+ fastestConsumer.consumerId,
+ slowestConsumer.consumerId)
+ } else if (settings.verboseLogging) {
+ logDetectSlowConsumers(trackingValues, slowestConsumer,
fastestConsumer, behind, fastestLagMillis)
+ }
+
+ }
+ }
+
+ private def logDetectSlowConsumers(
+ consumerTrackingValues: Vector[ConsumerTracking],
+ slowestConsumer: ConsumerTracking,
+ fastestConsumer: ConsumerTracking,
+ behind: Int,
+ fastestLagMillis: Long): Unit = {
+ if (behind > 0) {
+ val behindMillis = fastestConsumer.offsetTimestamp.toEpochMilli -
slowestConsumer.offsetTimestamp.toEpochMilli
+ log.debug(
+ s"Firehose entityType [$entityType] sliceRange [$sliceRangeStr]
detectSlowConsumers, " +
+ s"slowest [${slowestConsumer.consumerId}] is behind [$behind] " +
+ s"events from fastest [${fastestConsumer.consumerId}],
[$behindMillis] ms. " +
+ s"Consumer lag of fastest [$fastestLagMillis] ms")
+
+ consumerTrackingValues.foreach { tracking =>
+ val diffFastest = fastestConsumer.offsetTimestamp.toEpochMilli -
tracking.offsetTimestamp.toEpochMilli
+ val diffFastestStr =
+ if (diffFastest > 0) s"behind fastest [$diffFastest] ms"
+ else if (diffFastest < 0) s"ahead of fastest [$diffFastest] ms" //
not possible
+ else "same as fastest"
+ val diffSlowest = slowestConsumer.offsetTimestamp.toEpochMilli -
tracking.offsetTimestamp.toEpochMilli
+ val diffSlowestStr =
+ if (diffSlowest > 0) s"behind slowest [$diffSlowest] ms" // not
possible
+ else if (diffSlowest < 0) s"ahead of slowest [${-diffSlowest}] ms"
+ else "same as slowest"
+ val consumerBehind = elementsBehind(fastestConsumer.history,
tracking.history)
+
+ val logMessage =
+ s"Firehose entityType [$entityType] sliceRange [$sliceRangeStr]
consumer [${tracking.consumerId}], " +
+ s"behind [$consumerBehind] events from fastest, " +
+ s"$diffFastestStr, $diffSlowestStr, firehoseOnly
[${tracking.firehoseOnly}]"
+
+ log.debug(logMessage)
+ }
+ }
+ }
+
+ private def elementsBehind(fastest: Vector[TimestampOffset], slowest:
Vector[TimestampOffset]): Int = {
+ if (fastest.last == slowest.last) {
+ 0
+ } else {
+ val i = fastest.indexOf(slowest.last)
+ if (i >= 0)
+ fastest.size - i - 1
+ else
+ -1
+ }
+ }
+
+ @tailrec def updateConsumerFirehoseOnly(consumerId: String): Unit = {
+ val existingTracking = consumerTracking.get(consumerId)
+ val tracking = existingTracking match {
+ case null =>
+ throw new IllegalStateException(s"Expected existing tracking for
consumer [$consumerId]")
+ case existing =>
+ existing.copy(firehoseOnly = true)
+ }
+
+ if (!consumerTracking.replace(consumerId, existingTracking, tracking))
+ // concurrent update, try again
+ updateConsumerFirehoseOnly(consumerId)
+ }
+
+ }
+
+ def timestampOffset(env: EventEnvelope[Any]): TimestampOffset =
+ env match {
+ case eventEnvelope: EventEnvelope[?] if
eventEnvelope.offset.isInstanceOf[TimestampOffset] =>
+ eventEnvelope.offset.asInstanceOf[TimestampOffset]
+ case _ =>
+ throw new IllegalArgumentException(s"Expected TimestampOffset, but was
[${env.offset.getClass.getName}]")
+ }
+
+ def isDurationGreaterThan(from: Instant, to: Instant, duration: JDuration):
Boolean =
+ JDuration.between(from, to).compareTo(duration) > 0
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] class EventsBySliceFirehose(system: ActorSystem)
extends Extension {
+ import EventsBySliceFirehose._
+ private val log = Logging(system, classOf[EventsBySliceFirehose])
+ private val firehoses = new ConcurrentHashMap[FirehoseKey, Firehose]()
+
+ @tailrec final def getFirehose(firehoseKey: FirehoseKey): Firehose = {
+ val firehose = firehoses.computeIfAbsent(firehoseKey, key =>
createFirehose(key))
+ if (firehose.isShutdown) {
+ // concurrency race condition, but it should be removed
+ firehoses.remove(firehoseKey, firehose)
+ getFirehose(firehoseKey) // try again
+ } else
+ firehose
+ }
+
+ private def createFirehose(key: FirehoseKey): Firehose = {
+ implicit val sys: ActorSystem = system
+ val sliceRangeStr = s"${key.sliceRange.min}-${key.sliceRange.max}"
+
+ log.debug("Create firehose entityType [{}], sliceRange [{}]",
key.entityType, sliceRangeStr)
+
+ val settings = Settings(sys, key.pluginId)
+
+ val firehoseKillSwitch = KillSwitches.shared("firehoseKillSwitch")
+
+ val firehoseHub: Source[EventEnvelope[Any], NotUsed] =
+ underlyingEventsBySlices[Any](
+ settings.delegateQueryPluginId,
+ key.entityType,
+ key.sliceRange.min,
+ key.sliceRange.max,
+ TimestampOffset(Instant.now(), Map.empty),
+ firehose = true)
+ .via(firehoseKillSwitch.flow)
+
.runWith(BroadcastHub.sink[EventEnvelope[Any]](settings.broadcastBufferSize))
+
+ val firehose = new Firehose(key, settings, firehoseHub,
firehoseKillSwitch, log)
+
+ val reaperInterval = settings.slowConsumerReaperInterval
+ // var because it is used inside the scheduled block to cancel itself
+ var reaperTask: Cancellable = null
+ reaperTask = system.scheduler.scheduleWithFixedDelay(reaperInterval,
reaperInterval) { () =>
+ if (reaperTask == null)
+ () // theoretical possibility but would only mean that the first tick
is ignored
+ else if (firehose.isShutdown)
+ reaperTask.cancel()
+ else
+ firehose.detectSlowConsumers(Instant.now)
+ }(system.dispatcher)
+
+ firehose
+ }
+
+ def eventsBySlices[Event](
+ pluginId: String,
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset): Source[EventEnvelope[Event], NotUsed] = {
+ val sliceRange = minSlice to maxSlice
+ val firehoseKey = FirehoseKey(pluginId, entityType, sliceRange)
+ val firehose = getFirehose(firehoseKey)
+ val catchupSource = underlyingEventsBySlices[Event](
+ firehose.settings.delegateQueryPluginId,
+ entityType,
+ minSlice,
+ maxSlice,
+ offset,
+ firehose = false)
+ eventsBySlicesImpl(firehose, catchupSource)
+ }
+
+ def eventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ pluginId: String,
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed] = {
+ val sliceRange = minSlice to maxSlice
+ val firehoseKey = FirehoseKey(pluginId, entityType, sliceRange)
+ val firehose = getFirehose(firehoseKey)
+ val catchupSource =
underlyingEventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ firehose.settings.delegateQueryPluginId,
+ entityType,
+ minSlice,
+ maxSlice,
+ offset,
+ transformSnapshot)
+ eventsBySlicesImpl(firehose, catchupSource)
+ }
+
+ private def eventsBySlicesImpl[Event](
+ firehose: Firehose,
+ catchupSource: Source[EventEnvelope[Event], NotUsed]):
Source[EventEnvelope[Event], NotUsed] = {
+
+ val firehoseKey = firehose.firehoseKey
+ val settings = firehose.settings
+ val consumerId = UUID.randomUUID().toString
+
+ def consumerTerminated(): Unit = {
+ if (firehose.consumerTerminated(consumerId) == 0) {
+ // Don't shutdown firehose immediately because Projection it should
survive Projection restart
+ system.scheduler.scheduleOnce(settings.firehoseLingerTimeout) {
+ if (firehose.shutdownFirehoseIfNoConsumers())
+ firehoses.remove(firehoseKey)
+ }(system.dispatcher)
+ }
+ }
+
+ val catchupKillSwitch = KillSwitches.shared("catchupKillSwitch")
+ val catchupSourceWithKillSwitch =
+ catchupSource.asInstanceOf[Source[EventEnvelope[Any],
NotUsed]].via(catchupKillSwitch.flow)
+
+ val consumerKillSwitch = KillSwitches.shared("consumerKillSwitch")
+
+ val firehoseSource = firehose.firehoseHub
+
+ import GraphDSL.Implicits._
+ val catchupOrFirehose = GraphDSL.createGraph(catchupSourceWithKillSwitch)
{ implicit b => r =>
+ val merge = b.add(new CatchupOrFirehose(consumerId, firehose,
catchupKillSwitch))
+ r ~> merge.in1
+ FlowShape(merge.in0, merge.out)
+ }
+
+ firehoseSource
+ .via(catchupOrFirehose)
+ .map { env =>
+ // don't look at pub-sub or backtracking events
+ if (env.source == "")
+ firehose.updateConsumerTracking(consumerId, Instant.now(),
timestampOffset(env), consumerKillSwitch)
+ env.asInstanceOf[EventEnvelope[Event]]
+ }
+ .via(consumerKillSwitch.flow)
+ .watchTermination(Keep.right)
+ .mapMaterializedValue { termination =>
+ firehose.consumerStarted(consumerId, consumerKillSwitch)
+ termination.onComplete { _ =>
+ consumerTerminated()
+ }(system.dispatcher)
+ NotUsed
+ }
+ }
+
+ // can be overridden in tests
+ protected def underlyingEventsBySlices[Event](
+ pluginId: String,
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ @nowarn("msg=never used") firehose: Boolean):
Source[EventEnvelope[Event], NotUsed] = {
+ PersistenceQuery(system)
+ .readJournalFor[EventsBySliceQuery](pluginId)
+ .eventsBySlices(entityType, minSlice, maxSlice, offset)
+ }
+
+ private def underlyingEventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ pluginId: String,
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed] = {
+ PersistenceQuery(system)
+ .readJournalFor[EventsBySliceStartingFromSnapshotsQuery](pluginId)
+ .eventsBySlicesStartingFromSnapshots(entityType, minSlice, maxSlice,
offset, transformSnapshot)
+ }
+
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] object CatchupOrFirehose {
+ private sealed trait Mode
+ private case object CatchUpOnly extends Mode
+ private final case class Both(caughtUpTimestamp: Instant) extends Mode
+ private case object FirehoseOnly extends Mode
+
+ private case class DeduplicationCacheEntry(pid: String, seqNr: Long, source:
String)
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] class CatchupOrFirehose(
+ consumerId: String,
+ firehose: EventsBySliceFirehose.Firehose,
+ catchupKillSwitch: KillSwitch)
+ extends GraphStage[FanInShape2[EventEnvelope[Any], EventEnvelope[Any],
EventEnvelope[Any]]] {
+ import CatchupOrFirehose._
+ import EventsBySliceFirehose.isDurationGreaterThan
+ import EventsBySliceFirehose.timestampOffset
+ import firehose.firehoseKey.entityType
+ import firehose.settings
+
+ override def initialAttributes = Attributes.name("CatchupOrFirehose")
+ override val shape: FanInShape2[EventEnvelope[Any], EventEnvelope[Any],
EventEnvelope[Any]] =
+ new FanInShape2[EventEnvelope[Any], EventEnvelope[Any],
EventEnvelope[Any]]("CatchupOrFirehose")
+ def out: Outlet[EventEnvelope[Any]] = shape.out
+ val firehoseInlet: Inlet[EventEnvelope[Any]] = shape.in0
+ val catchupInlet: Inlet[EventEnvelope[Any]] = shape.in1
+
+ private val sliceRangeStr =
s"${firehose.firehoseKey.sliceRange.min}-${firehose.firehoseKey.sliceRange.max}"
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with StageLogging {
+
+ // Without this the completion signalling would take one extra pull
+ private def willShutDown: Boolean = isClosed(firehoseInlet)
+
+ private val firehoseHandler = new FirehoseHandler(firehoseInlet)
+ private val catchupHandler = new CatchupHandler(catchupInlet)
+
+ private var mode: Mode = CatchUpOnly
+
+ // cache of seen pid/seqNr
+ private var deduplicationCache =
mutable.LinkedHashSet.empty[DeduplicationCacheEntry]
+ private val deduplicationCacheEvictThreshold =
(settings.deduplicationCapacity * 1.1).toInt
+
+ override protected def logSource: Class[?] = classOf[CatchupOrFirehose]
+
+ setHandler(out,
+ new OutHandler {
+ override def onPull(): Unit = {
+ tryPushOutput()
+ tryPullAllIfNeeded()
+ }
+ })
+
+ setHandler(firehoseInlet, firehoseHandler)
+ setHandler(catchupInlet, catchupHandler)
+
+ private def tryPushOutput(): Unit = {
+ def tryPushFirehoseValue(deduplicate: Boolean): Boolean =
+ firehoseHandler.value match {
+ case OptionVal.Some(env) =>
+ firehoseHandler.value = OptionVal.None
+ if (!(deduplicate && isDuplicate(env))) {
+ if (settings.verboseLogging && log.isDebugEnabled)
+ log.debug(
+ s"Firehose entityType [$entityType] sliceRange
[$sliceRangeStr] consumer [$consumerId] push from " +
+ s" firehose [${env.persistenceId}] seqNr
[${env.sequenceNr}], source [${env.source}]")
+ push(out, env)
+ true
+ } else
+ false
+ case _ =>
+ false
+ }
+
+ def tryPushCatchupValue(deduplicate: Boolean): Boolean =
+ catchupHandler.value match {
+ case OptionVal.Some(env) =>
+ catchupHandler.value = OptionVal.None
+ if (!(deduplicate && isDuplicate(env))) {
+ if (settings.verboseLogging && log.isDebugEnabled)
+ log.debug(
+ s"Firehose entityType [$entityType] sliceRange
[$sliceRangeStr] consumer [$consumerId] push from " +
+ s"catchup [${env.persistenceId}] seqNr
[${env.sequenceNr}], source [${env.source}]")
+ push(out, env)
+ true
+ } else
+ false
+ case _ =>
+ false
+ }
+
+ if (isAvailable(out)) {
+ mode match {
+ case FirehoseOnly =>
+ // there can be one final value from catchup when switching to
FirehoseOnly
+ if (!tryPushCatchupValue(deduplicate = false))
+ tryPushFirehoseValue(deduplicate = false)
+ case Both(_) =>
+ if (!tryPushFirehoseValue(deduplicate = true))
+ tryPushCatchupValue(deduplicate = true)
+ case CatchUpOnly =>
+ tryPushCatchupValue(deduplicate = false)
+ }
+ }
+
+ if (willShutDown) completeStage()
+ }
+
+ def isDuplicate(env: EventEnvelope[Any]): Boolean = {
+ if (settings.deduplicationCapacity == 0)
+ false
+ else {
+ val entry = DeduplicationCacheEntry(env.persistenceId,
env.sequenceNr, env.source)
+ val result = {
+ if (deduplicationCache.contains(entry)) {
+ true
+ } else {
+ deduplicationCache.add(entry)
+ false
+ }
+ }
+
+ if (deduplicationCache.size >= deduplicationCacheEvictThreshold) {
+ // weird that add modifies the instance but drop returns a new
instance
+ deduplicationCache =
deduplicationCache.drop(deduplicationCache.size -
settings.deduplicationCapacity)
+ }
+
+ result
+ }
+ }
+
+ private def tryPullAllIfNeeded(): Unit = {
+ if (isClosed(firehoseInlet)) {
+ completeStage()
+ } else {
+ if (!hasBeenPulled(firehoseInlet) && firehoseHandler.value.isEmpty) {
+ tryPull(firehoseInlet)
+ }
+ if (mode != FirehoseOnly && !hasBeenPulled(catchupInlet) &&
catchupHandler.value.isEmpty) {
+ tryPull(catchupInlet)
+ }
+ }
+ }
+
+ def isCaughtUp(env: EventEnvelope[Any]): Boolean = {
+ if (env.source == "") {
+ val offset = timestampOffset(env)
+ firehoseHandler.firehoseOffset.timestamp != Instant.EPOCH &&
+ !firehoseHandler.firehoseOffset.timestamp.isAfter(offset.timestamp)
+ } else
+ false // don't look at pub-sub or backtracking events
+ }
+
+ private class FirehoseHandler(in: Inlet[EventEnvelope[Any]]) extends
InHandler {
+ var value: OptionVal[EventEnvelope[Any]] = OptionVal.None
+ var firehoseOffset: TimestampOffset = TimestampOffset(Instant.EPOCH,
Map.empty)
+
+ def updateFirehoseOffset(env: EventEnvelope[Any]): Unit = {
+ // don't look at pub-sub or backtracking events
+ if (env.source == "")
+ firehoseOffset = timestampOffset(env)
+ }
+
+ override def onPush(): Unit = {
+ if (value.isDefined)
+ throw new IllegalStateException("FirehoseInlet.onPush but has
already value. This is a bug.")
+
+ val env = grab(in)
+
+ mode match {
+ case FirehoseOnly =>
+ value = OptionVal.Some(env)
+ case Both(_) =>
+ updateFirehoseOffset(env)
+ value = OptionVal.Some(env)
+ case CatchUpOnly =>
+ updateFirehoseOffset(env)
+ }
+
+ tryPushOutput()
+ tryPullAllIfNeeded()
+ }
+ }
+
+ private class CatchupHandler(in: Inlet[EventEnvelope[Any]]) extends
InHandler {
+ var value: OptionVal[EventEnvelope[Any]] = OptionVal.None
+
+ override def onPush(): Unit = {
+ if (value.isDefined)
+ throw new IllegalStateException("CatchupInlet.onPush but has
already value. This is a bug.")
+
+ val env = grab(in)
+
+ mode match {
+ case CatchUpOnly =>
+ if (isCaughtUp(env)) {
+ val timestamp = timestampOffset(env).timestamp
+ log.debug(
+ "Firehose entityType [{}] sliceRange [{}] consumer [{}]
caught up at [{}]",
+ entityType,
+ sliceRangeStr,
+ consumerId,
+ timestamp)
+ mode = Both(timestamp)
+ }
+ value = OptionVal.Some(env)
+
+ case Both(caughtUpTimestamp) =>
+ // don't look at pub-sub or backtracking events
+ if (env.source == "") {
+ val timestamp = timestampOffset(env).timestamp
+ if (isDurationGreaterThan(caughtUpTimestamp, timestamp,
settings.catchupOverlap)) {
+ firehose.updateConsumerFirehoseOnly(consumerId)
+ log.debug(
+ "Firehose entityType [{}] sliceRange [{}] consumer [{}]
switching to firehose only [{}]",
+ entityType,
+ sliceRangeStr,
+ consumerId,
+ timestamp)
+ catchupKillSwitch.shutdown()
+ mode = FirehoseOnly
+ deduplicationCache =
mutable.LinkedHashSet.empty[DeduplicationCacheEntry]
+ }
+ }
+
+ value = OptionVal.Some(env)
+
+ case FirehoseOnly =>
+ // skip
+ }
+
+ tryPushOutput()
+ tryPullAllIfNeeded()
+ }
+
+ override def onUpstreamFinish(): Unit = {
+ // important to override onUpstreamFinish, otherwise it will close
everything
+ log.debug(
+ "Firehose entityType [{}] sliceRange [{}] consumer [{}] catchup
closed",
+ entityType,
+ sliceRangeStr,
+ consumerId)
+ }
+ }
+
+ }
+
+ override def toString = "CatchupOrFirehose"
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/CurrentEventsByPersistenceIdStartingFromSnapshotQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/CurrentEventsByPersistenceIdStartingFromSnapshotQuery.scala
new file mode 100644
index 0000000000..88d8f5c262
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/CurrentEventsByPersistenceIdStartingFromSnapshotQuery.scala
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.javadsl
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.persistence.query.javadsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.stream.javadsl.Source
+
+/**
+ * A plugin may optionally support this query by implementing this trait.
+ */
+@ApiMayChange
+trait CurrentEventsByPersistenceIdStartingFromSnapshotQuery extends
ReadJournal {
+
+ /**
+ * Same as [[CurrentEventsByPersistenceIdTypedQuery]] but with the purpose
to use snapshot as starting point
+ * and thereby reducing number of events that have to be loaded.
+ */
+ def currentEventsByPersistenceIdStartingFromSnapshot[Snapshot, Event](
+ persistenceId: String,
+ fromSequenceNr: Long,
+ toSequenceNr: Long,
+ transformSnapshot: java.util.function.Function[Snapshot, Event]):
Source[EventEnvelope[Event], NotUsed]
+
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/CurrentEventsBySliceStartingFromSnapshotsQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/CurrentEventsBySliceStartingFromSnapshotsQuery.scala
new file mode 100644
index 0000000000..c7e57636be
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/CurrentEventsBySliceStartingFromSnapshotsQuery.scala
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.javadsl
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.japi.Pair
+import pekko.persistence.query.Offset
+import pekko.persistence.query.javadsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.stream.javadsl.Source
+
+/**
+ * A plugin may optionally support this query by implementing this trait.
+ *
+ * API May Change
+ */
+@ApiMayChange
+trait CurrentEventsBySliceStartingFromSnapshotsQuery extends ReadJournal {
+
+ /**
+ * Same as [[EventsBySliceStartingFromSnapshotsQuery]] but with the purpose
to use snapshots as starting points
+ * and thereby reducing number of events that have to be loaded. This can be
useful if the consumer start
+ * from zero without any previously processed offset or if it has been
disconnected for a long while and
+ * its offset is far behind.
+ *
+ * Same type of query as
[[EventsBySliceStartingFromSnapshotsQuery.eventsBySlicesStartingFromSnapshots]]
but
+ * the event stream is completed immediately when it reaches the end of the
"result set".
+ */
+ def currentEventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: java.util.function.Function[Snapshot, Event]):
Source[EventEnvelope[Event], NotUsed]
+
+ def sliceForPersistenceId(persistenceId: String): Int
+
+ def sliceRanges(numberOfRanges: Int): java.util.List[Pair[Integer, Integer]]
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsByPersistenceIdStartingFromSnapshotQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsByPersistenceIdStartingFromSnapshotQuery.scala
new file mode 100644
index 0000000000..ebbe98a6fc
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsByPersistenceIdStartingFromSnapshotQuery.scala
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.javadsl
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.persistence.query.javadsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.stream.javadsl.Source
+
+/**
+ * A plugin may optionally support this query by implementing this trait.
+ */
+@ApiMayChange
+trait EventsByPersistenceIdStartingFromSnapshotQuery extends ReadJournal {
+
+ /**
+ * Same as [[EventsByPersistenceIdTypedQuery]] but with the purpose to use
snapshot as starting point
+ * and thereby reducing number of events that have to be loaded.
+ */
+ def eventsByPersistenceIdStartingFromSnapshot[Snapshot, Event](
+ persistenceId: String,
+ fromSequenceNr: Long,
+ toSequenceNr: Long,
+ transformSnapshot: java.util.function.Function[Snapshot, Event]):
Source[EventEnvelope[Event], NotUsed]
+
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceFirehoseQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceFirehoseQuery.scala
new file mode 100644
index 0000000000..f08a5426bc
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceFirehoseQuery.scala
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.javadsl
+
+import java.time.Instant
+import java.util
+import java.util.Optional
+import java.util.concurrent.CompletionStage
+
+import scala.concurrent.ExecutionContext
+import scala.jdk.FutureConverters._
+import scala.jdk.OptionConverters._
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.japi.Pair
+import pekko.persistence.query.Offset
+import pekko.persistence.query.javadsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.persistence.query.typed.scaladsl
+import pekko.stream.javadsl.Source
+
+object EventsBySliceFirehoseQuery {
+ val Identifier: String = scaladsl.EventsBySliceFirehoseQuery.Identifier
+}
+
+/**
+ * This wrapper of [[EventsBySliceQuery]] gives better scalability when many
consumers retrieve the
+ * same events, for example many Projections of the same entity type. The
purpose is to share
+ * the stream of events from the database and fan out to connected consumer
streams. Thereby fewer
+ * queries and loading of events from the database.
+ *
+ * It is retrieved with:
+ * {{{
+ * EventsBySliceQuery queries =
+ * PersistenceQuery.get(system).getReadJournalFor(EventsBySliceQuery.class,
EventsBySliceFirehoseQuery.Identifier());
+ * }}}
+ *
+ * Corresponding Scala API is in
[[pekko.persistence.query.typed.scaladsl.EventsBySliceFirehoseQuery]].
+ *
+ * Configuration settings can be defined in the configuration section with the
+ * absolute path corresponding to the identifier, which is
`"pekko.persistence.query.events-by-slice-firehose"`
+ * for the default [[EventsBySliceFirehoseQuery#Identifier]]. See
`reference.conf`.
+ */
+final class EventsBySliceFirehoseQuery(delegate:
scaladsl.EventsBySliceFirehoseQuery)
+ extends ReadJournal
+ with EventsBySliceQuery
+ with EventsBySliceStartingFromSnapshotsQuery
+ with EventTimestampQuery
+ with LoadEventQuery {
+
+ override def sliceForPersistenceId(persistenceId: String): Int =
+ delegate.sliceForPersistenceId(persistenceId)
+
+ override def eventsBySlices[Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset): Source[EventEnvelope[Event], NotUsed] =
+ delegate.eventsBySlices(entityType, minSlice, maxSlice, offset).asJava
+
+ override def eventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: java.util.function.Function[Snapshot, Event]):
Source[EventEnvelope[Event], NotUsed] =
+ delegate.eventsBySlicesStartingFromSnapshots(entityType, minSlice,
maxSlice, offset, transformSnapshot(_)).asJava
+
+ override def sliceRanges(numberOfRanges: Int): util.List[Pair[Integer,
Integer]] = {
+ import scala.jdk.CollectionConverters._
+ delegate
+ .sliceRanges(numberOfRanges)
+ .map(range => Pair(Integer.valueOf(range.min),
Integer.valueOf(range.max)))
+ .asJava
+ }
+
+ override def timestampOf(persistenceId: String, sequenceNr: Long):
CompletionStage[Optional[Instant]] =
+ delegate.timestampOf(persistenceId,
sequenceNr).map(_.toJava)(ExecutionContext.parasitic).asJava
+
+ override def loadEnvelope[Event](persistenceId: String, sequenceNr: Long):
CompletionStage[EventEnvelope[Event]] =
+ delegate.loadEnvelope[Event](persistenceId, sequenceNr).asJava
+
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceQuery.scala
index b0a4afbfdb..190851788a 100644
---
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceQuery.scala
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceQuery.scala
@@ -28,6 +28,8 @@ import pekko.stream.javadsl.Source
* `EventsBySliceQuery` that is using a timestamp based offset should also
implement [[EventTimestampQuery]] and
* [[LoadEventQuery]].
*
+ * See also [[EventsBySliceFirehoseQuery]].
+ *
* API May Change
*/
@ApiMayChange
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceStartingFromSnapshotsQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceStartingFromSnapshotsQuery.scala
new file mode 100644
index 0000000000..84e36ccb1d
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/javadsl/EventsBySliceStartingFromSnapshotsQuery.scala
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.javadsl
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.japi.Pair
+import pekko.persistence.query.Offset
+import pekko.persistence.query.javadsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.stream.javadsl.Source
+
+/**
+ * A plugin may optionally support this query by implementing this trait.
+ *
+ * `EventsBySliceQuery` that is using a timestamp based offset should also
implement [[EventTimestampQuery]] and
+ * [[LoadEventQuery]].
+ *
+ * See also [[EventsBySliceFirehoseQuery]].
+ *
+ * API May Change
+ */
+@ApiMayChange
+trait EventsBySliceStartingFromSnapshotsQuery extends ReadJournal {
+
+ /**
+ * Same as [[EventsBySliceQuery]] but with the purpose to use snapshots as
starting points and thereby reducing number of
+ * events that have to be loaded. This can be useful if the consumer start
from zero without any previously processed
+ * offset or if it has been disconnected for a long while and its offset is
far behind.
+ */
+ def eventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: java.util.function.Function[Snapshot, Event]):
Source[EventEnvelope[Event], NotUsed]
+
+ def sliceForPersistenceId(persistenceId: String): Int
+
+ def sliceRanges(numberOfRanges: Int): java.util.List[Pair[Integer, Integer]]
+
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/CurrentEventsByPersistenceIdStartingFromSnapshotQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/CurrentEventsByPersistenceIdStartingFromSnapshotQuery.scala
new file mode 100644
index 0000000000..d2075bf2ea
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/CurrentEventsByPersistenceIdStartingFromSnapshotQuery.scala
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.scaladsl
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.persistence.query.scaladsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.stream.scaladsl.Source
+
+/**
+ * A plugin may optionally support this query by implementing this trait.
+ */
+@ApiMayChange
+trait CurrentEventsByPersistenceIdStartingFromSnapshotQuery extends
ReadJournal {
+
+ /**
+ * Same as [[CurrentEventsByPersistenceIdTypedQuery]] but with the purpose
to use snapshot as starting point
+ * and thereby reducing number of events that have to be loaded.
+ */
+ def currentEventsByPersistenceIdStartingFromSnapshot[Snapshot, Event](
+ persistenceId: String,
+ fromSequenceNr: Long,
+ toSequenceNr: Long,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed]
+
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/CurrentEventsBySliceStartingFromSnapshotsQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/CurrentEventsBySliceStartingFromSnapshotsQuery.scala
new file mode 100644
index 0000000000..d79f01d830
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/CurrentEventsBySliceStartingFromSnapshotsQuery.scala
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.scaladsl
+
+import scala.collection.immutable
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.persistence.query.Offset
+import pekko.persistence.query.scaladsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.stream.scaladsl.Source
+
+/**
+ * A plugin may optionally support this query by implementing this trait.
+ *
+ * API May Change
+ */
+@ApiMayChange
+trait CurrentEventsBySliceStartingFromSnapshotsQuery extends ReadJournal {
+
+ /**
+ * Same as [[EventsBySliceStartingFromSnapshotsQuery]] but with the purpose
to use snapshots as starting points
+ * and thereby reducing number of events that have to be loaded. This can be
useful if the consumer start
+ * from zero without any previously processed offset or if it has been
disconnected for a long while and
+ * its offset is far behind.
+ *
+ * Same type of query as
[[EventsBySliceStartingFromSnapshotsQuery.eventsBySlicesStartingFromSnapshots]]
but
+ * the event stream is completed immediately when it reaches the end of the
"result set".
+ */
+ def currentEventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed]
+
+ def sliceForPersistenceId(persistenceId: String): Int
+
+ def sliceRanges(numberOfRanges: Int): immutable.Seq[Range]
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsByPersistenceIdStartingFromSnapshotQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsByPersistenceIdStartingFromSnapshotQuery.scala
new file mode 100644
index 0000000000..e33fe28abf
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsByPersistenceIdStartingFromSnapshotQuery.scala
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.scaladsl
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.persistence.query.scaladsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.stream.scaladsl.Source
+
+/**
+ * A plugin may optionally support this query by implementing this trait.
+ */
+@ApiMayChange
+trait EventsByPersistenceIdStartingFromSnapshotQuery extends ReadJournal {
+
+ /**
+ * Same as [[EventsByPersistenceIdTypedQuery]] but with the purpose to use
snapshot as starting point
+ * and thereby reducing number of events that have to be loaded.
+ */
+ def eventsByPersistenceIdStartingFromSnapshot[Snapshot, Event](
+ persistenceId: String,
+ fromSequenceNr: Long,
+ toSequenceNr: Long,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed]
+
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceFirehoseQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceFirehoseQuery.scala
new file mode 100644
index 0000000000..5eb68e98f7
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceFirehoseQuery.scala
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.scaladsl
+
+import java.time.Instant
+
+import scala.annotation.nowarn
+import scala.collection.immutable
+import scala.concurrent.Future
+
+import com.typesafe.config.Config
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.actor.ExtendedActorSystem
+import pekko.persistence.Persistence
+import pekko.persistence.query.Offset
+import pekko.persistence.query.PersistenceQuery
+import pekko.persistence.query.scaladsl._
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.persistence.query.typed.internal.EventsBySliceFirehose
+import pekko.stream.scaladsl.Source
+
+object EventsBySliceFirehoseQuery {
+ val Identifier = "pekko.persistence.query.events-by-slice-firehose"
+
+}
+
+/**
+ * This wrapper of [[EventsBySliceQuery]] gives better scalability when many
consumers retrieve the
+ * same events, for example many Projections of the same entity type. The
purpose is to share
+ * the stream of events from the database and fan out to connected consumer
streams. Thereby fewer
+ * queries and loading of events from the database.
+ *
+ * It is retrieved with:
+ * {{{
+ * val queries =
PersistenceQuery(system).readJournalFor[EventsBySliceQuery](EventsBySliceFirehoseQuery.Identifier)
+ * }}}
+ *
+ * Corresponding Java API is in
[[org.apache.pekko.persistence.query.typed.javadsl.EventsBySliceFirehoseQuery]].
+ *
+ * Configuration settings can be defined in the configuration section with the
+ * absolute path corresponding to the identifier, which is
`"pekko.persistence.query.events-by-slice-firehose"`
+ * for the default [[EventsBySliceFirehoseQuery#Identifier]]. See
`reference.conf`.
+ */
+@nowarn("msg=never used")
+final class EventsBySliceFirehoseQuery(system: ExtendedActorSystem, config:
Config, cfgPath: String)
+ extends ReadJournal
+ with EventsBySliceQuery
+ with EventsBySliceStartingFromSnapshotsQuery
+ with EventTimestampQuery
+ with LoadEventQuery {
+
+ private lazy val persistenceExt = Persistence(system)
+ private lazy val settings = EventsBySliceFirehose.Settings(system, cfgPath)
+
+ override def eventsBySlices[Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset): Source[EventEnvelope[Event], NotUsed] =
+ EventsBySliceFirehose(system).eventsBySlices(cfgPath, entityType,
minSlice, maxSlice, offset)
+
+ override def eventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed] =
+ EventsBySliceFirehose(system).eventsBySlicesStartingFromSnapshots(
+ cfgPath,
+ entityType,
+ minSlice,
+ maxSlice,
+ offset,
+ transformSnapshot)
+
+ override def sliceForPersistenceId(persistenceId: String): Int =
+ persistenceExt.sliceForPersistenceId(persistenceId)
+
+ override def sliceRanges(numberOfRanges: Int): immutable.Seq[Range] =
+ persistenceExt.sliceRanges(numberOfRanges)
+
+ override def timestampOf(persistenceId: String, sequenceNr: Long):
Future[Option[Instant]] = {
+ eventsBySliceQuery match {
+ case q: EventTimestampQuery => q.timestampOf(persistenceId, sequenceNr)
+ case _ =>
+ throw new IllegalArgumentException(
+ s"Underlying ReadJournal [${settings.delegateQueryPluginId}] doesn't
implement EventTimestampQuery")
+ }
+ }
+
+ override def loadEnvelope[Event](persistenceId: String, sequenceNr: Long):
Future[EventEnvelope[Event]] =
+ eventsBySliceQuery match {
+ case q: LoadEventQuery => q.loadEnvelope(persistenceId, sequenceNr)
+ case _ =>
+ throw new IllegalArgumentException(
+ s"Underlying ReadJournal [${settings.delegateQueryPluginId}] " +
+ "doesn't implement LoadEventQuery")
+ }
+
+ private def eventsBySliceQuery: EventsBySliceQuery = {
+ val delegateQueryPluginId =
+
EventsBySliceFirehose.Settings.delegateQueryPluginId(system.settings.config.getConfig(cfgPath))
+
PersistenceQuery(system).readJournalFor[EventsBySliceQuery](delegateQueryPluginId)
+ }
+
+}
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceQuery.scala
index 3591a17642..7d76b020d7 100644
---
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceQuery.scala
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceQuery.scala
@@ -29,6 +29,8 @@ import pekko.stream.scaladsl.Source
* `EventsBySliceQuery` that is using a timestamp based offset should also
implement [[EventTimestampQuery]] and
* [[LoadEventQuery]].
*
+ * See also [[EventsBySliceFirehoseQuery]].
+ *
* API May Change
*/
@ApiMayChange
diff --git
a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceStartingFromSnapshotsQuery.scala
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceStartingFromSnapshotsQuery.scala
new file mode 100644
index 0000000000..cb53fbd02a
--- /dev/null
+++
b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/typed/scaladsl/EventsBySliceStartingFromSnapshotsQuery.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.scaladsl
+
+import scala.collection.immutable
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.annotation.ApiMayChange
+import pekko.persistence.query.Offset
+import pekko.persistence.query.scaladsl.ReadJournal
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.stream.scaladsl.Source
+
+/**
+ * A plugin may optionally support this query by implementing this trait.
+ *
+ * `EventsBySliceStartingFromSnapshotsQuery` that is using a timestamp based
offset should also implement [[EventTimestampQuery]] and
+ * [[LoadEventQuery]].
+ *
+ * See also [[EventsBySliceFirehoseQuery]].
+ *
+ * API May Change
+ */
+@ApiMayChange
+trait EventsBySliceStartingFromSnapshotsQuery extends ReadJournal {
+
+ /**
+ * Same as [[EventsBySliceQuery]] but with the purpose to use snapshots as
starting points and thereby reducing number of
+ * events that have to be loaded. This can be useful if the consumer start
from zero without any previously processed
+ * offset or if it has been disconnected for a long while and its offset is
far behind.
+ */
+ def eventsBySlicesStartingFromSnapshots[Snapshot, Event](
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ transformSnapshot: Snapshot => Event): Source[EventEnvelope[Event],
NotUsed]
+
+ def sliceForPersistenceId(persistenceId: String): Int
+
+ def sliceRanges(numberOfRanges: Int): immutable.Seq[Range]
+
+}
diff --git
a/persistence-query/src/test/scala/org/apache/pekko/persistence/query/TestClock.scala
b/persistence-query/src/test/scala/org/apache/pekko/persistence/query/TestClock.scala
new file mode 100644
index 0000000000..55fcd4fa3c
--- /dev/null
+++
b/persistence-query/src/test/scala/org/apache/pekko/persistence/query/TestClock.scala
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query
+
+import java.time.{ Duration => JDuration }
+import java.time.Clock
+import java.time.Instant
+import java.time.ZoneId
+import java.time.ZoneOffset
+
+import scala.concurrent.duration.FiniteDuration
+
+import org.apache.pekko.annotation.InternalApi
+
+/**
+ * INTERNAL API
+ */
+@InternalApi private[pekko] class TestClock extends Clock {
+
+ @volatile private var _instant = roundToMillis(Instant.now())
+
+ override def getZone: ZoneId = ZoneOffset.UTC
+
+ override def withZone(zone: ZoneId): Clock =
+ throw new UnsupportedOperationException("withZone not supported")
+
+ override def instant(): Instant =
+ _instant
+
+ def setInstant(newInstant: Instant): Unit =
+ _instant = roundToMillis(newInstant)
+
+ def tick(duration: JDuration): Instant = {
+ val newInstant = roundToMillis(_instant.plus(duration))
+ _instant = newInstant
+ newInstant
+ }
+
+ def tick(duration: FiniteDuration): Instant =
+ tick(JDuration.ofMillis(duration.toMillis))
+
+ private def roundToMillis(i: Instant): Instant = {
+ // algo taken from java.time.Clock.tick
+ val epochMilli = i.toEpochMilli
+ Instant.ofEpochMilli(epochMilli - Math.floorMod(epochMilli, 1L))
+ }
+
+}
diff --git
a/persistence-query/src/test/scala/org/apache/pekko/persistence/query/typed/internal/EventsBySliceFirehoseSpec.scala
b/persistence-query/src/test/scala/org/apache/pekko/persistence/query/typed/internal/EventsBySliceFirehoseSpec.scala
new file mode 100644
index 0000000000..fc96abab8d
--- /dev/null
+++
b/persistence-query/src/test/scala/org/apache/pekko/persistence/query/typed/internal/EventsBySliceFirehoseSpec.scala
@@ -0,0 +1,440 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.query.typed.internal
+
+import java.time.{ Duration => JDuration }
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.concurrent.{ ExecutionContext, Promise }
+import scala.concurrent.duration._
+
+import com.typesafe.config.ConfigFactory
+import org.scalatest.concurrent.Eventually
+
+import org.apache.pekko
+import pekko.NotUsed
+import pekko.persistence.Persistence
+import pekko.persistence.query.NoOffset
+import pekko.persistence.query.Offset
+import pekko.persistence.query.TestClock
+import pekko.persistence.query.TimestampOffset
+import pekko.persistence.query.typed.EventEnvelope
+import pekko.persistence.query.typed.internal.EventsBySliceFirehose.FirehoseKey
+import
pekko.persistence.query.typed.internal.EventsBySliceFirehose.SlowConsumerException
+import pekko.persistence.query.typed.scaladsl.EventsBySliceFirehoseQuery
+import pekko.stream.scaladsl.Keep
+import pekko.stream.scaladsl.Source
+import pekko.stream.testkit.TestPublisher
+import pekko.stream.testkit.TestSubscriber
+import pekko.stream.testkit.TestSubscriber.OnNext
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.stream.testkit.scaladsl.TestSource
+import pekko.testkit.PekkoSpec
+import pekko.testkit.WithLogCapturing
+
+object EventsBySliceFirehoseSpec {
+ // real PersistenceId is in pekko-persistence-typed, and no dependency to
that from here
+ final case class PersistenceId(entityTypeHint: String, entityId: String) {
+ def id: String = s"$entityTypeHint|$entityId"
+ }
+
+ private val config = ConfigFactory.parseString(s"""
+ pekko.loglevel = DEBUG
+ pekko.loggers = ["org.apache.pekko.testkit.SilenceAllTestEventListener"]
+
+ pekko.persistence.query.events-by-slice-firehose {
+ # test-query-plugin is not used by this test, but must be defined
+ delegate-query-plugin-id = test-query-plugin
+
+ broadcast-buffer-size = 64
+
+ firehose-linger-timeout = 2s
+
+ # don't deplicate in this test
+ deduplication-capacity = 0
+
+ # disable reaper in this test, will be triggered by code
+ slow-consumer-reaper-interval = 1 day
+
+ verbose-debug-logging = on
+ }
+
+ events-by-slice-firehose-with-deduplication =
$${pekko.persistence.query.events-by-slice-firehose}
+ events-by-slice-firehose-with-deduplication {
+ deduplication-capacity = 10000
+ }
+ """)
+}
+
+class EventsBySliceFirehoseSpec
+ extends PekkoSpec(EventsBySliceFirehoseSpec.config)
+ with WithLogCapturing
+ with Eventually {
+ import EventsBySliceFirehoseSpec._
+
+ private val entityType = "EntityA"
+
+ private val persistence = Persistence(system)
+
+ private val clock = new TestClock
+
+ private def createEnvelope(
+ pid: PersistenceId,
+ seqNr: Long,
+ evt: String,
+ tags: Set[String] = Set.empty): EventEnvelope[Any] = {
+ clock.tick(JDuration.ofMillis(1))
+ val now = clock.instant()
+ EventEnvelope(
+ TimestampOffset(now, Map(pid.id -> seqNr)),
+ pid.id,
+ seqNr,
+ evt,
+ now.toEpochMilli,
+ pid.entityTypeHint,
+ persistence.sliceForPersistenceId(pid.id),
+ filtered = false,
+ source = "",
+ tags = tags)
+ }
+
+ private val envelopes = Vector(
+ createEnvelope(PersistenceId(entityType, "a"), 1, "a1"),
+ createEnvelope(PersistenceId(entityType, "b"), 1, "b1"),
+ createEnvelope(PersistenceId(entityType, "c"), 1, "c1"),
+ createEnvelope(PersistenceId(entityType, "d"), 1, "d1"))
+
+ private class Setup {
+ def allEnvelopes: Vector[EventEnvelope[Any]] = envelopes
+
+ def sliceRange = 0 to 1023
+
+ def pluginId: String = EventsBySliceFirehoseQuery.Identifier
+
+ class ConsumerSetup {
+ private val catchupPublisherPromise =
Promise[TestPublisher.Probe[EventEnvelope[Any]]]()
+ val catchupSource: Source[EventEnvelope[Any], NotUsed] =
+ TestSource[EventEnvelope[Any]]().mapMaterializedValue { probe =>
+ catchupPublisherPromise.success(probe)
+ NotUsed
+ }
+
+ lazy val outProbe =
+ eventsBySliceFirehose
+ .eventsBySlices[Any](pluginId, entityType, sliceRange.min,
sliceRange.max, NoOffset)
+ .runWith(TestSink())
+
+ lazy val catchupPublisher = {
+ outProbe // materialize
+ catchupPublisherPromise.future.futureValue
+ }
+ }
+
+ private val consumerCount = new AtomicInteger
+ private val consumers = Vector.fill(100)(new ConsumerSetup)
+
+ def catchupPublisher(consumerIndex: Int):
TestPublisher.Probe[EventEnvelope[Any]] =
+ consumers(consumerIndex).catchupPublisher
+ def catchupPublisher: TestPublisher.Probe[EventEnvelope[Any]] =
catchupPublisher(0)
+
+ def outProbe(consumerIndex: Int): TestSubscriber.Probe[EventEnvelope[Any]]
= consumers(consumerIndex).outProbe
+ def outProbe: TestSubscriber.Probe[EventEnvelope[Any]] = outProbe(0)
+
+ val firehoseRunning = new AtomicBoolean
+ private val firehosePublisherPromise =
Promise[TestPublisher.Probe[EventEnvelope[Any]]]()
+ private val firehoseSource: Source[EventEnvelope[Any], NotUsed] =
+
TestSource[EventEnvelope[Any]]().watchTermination(Keep.both).mapMaterializedValue
{
+ case (probe, termination) =>
+ firehoseRunning.set(true)
+ termination.onComplete(_ =>
firehoseRunning.set(false))(ExecutionContext.parasitic)
+ firehosePublisherPromise.success(probe)
+ NotUsed
+ }
+
+ val eventsBySliceFirehose = new
EventsBySliceFirehose(system.classicSystem) {
+ override protected def underlyingEventsBySlices[Event](
+ pluginId: String,
+ entityType: String,
+ minSlice: Int,
+ maxSlice: Int,
+ offset: Offset,
+ firehose: Boolean): Source[EventEnvelope[Event], NotUsed] = {
+ if (firehose)
+ firehoseSource.map(_.asInstanceOf[EventEnvelope[Event]])
+ else {
+ val i = consumerCount.getAndIncrement()
+ consumers(i).catchupSource.map(_.asInstanceOf[EventEnvelope[Event]])
+ }
+ }
+ }
+
+ lazy val firehosePublisher = {
+ outProbe // materialize at least one
+ firehosePublisherPromise.future.futureValue
+ }
+ }
+
+ "EventsBySliceFirehose" must {
+ "emit from catchup" in new Setup {
+ allEnvelopes.foreach(catchupPublisher.sendNext)
+ outProbe.request(10)
+ outProbe.expectNextN(envelopes.size) shouldBe envelopes
+ }
+
+ "emit from catchup and then switch over to firehose" in new Setup {
+ catchupPublisher.sendNext(allEnvelopes(0))
+ catchupPublisher.sendNext(allEnvelopes(1))
+ outProbe.request(10)
+ outProbe.expectNext(allEnvelopes(0))
+ outProbe.expectNext(allEnvelopes(1))
+
+ firehosePublisher.sendNext(allEnvelopes(2))
+ outProbe.expectNoMessage()
+
+ catchupPublisher.sendNext(allEnvelopes(2))
+ outProbe.expectNext(allEnvelopes(2)) // still from catchup
+
+ firehosePublisher.sendNext(allEnvelopes(3))
+ outProbe.expectNext(allEnvelopes(3)) // from firehose
+
+ catchupPublisher.sendNext(allEnvelopes(3))
+ outProbe.expectNext(allEnvelopes(3)) // from catchup, emitting from both
+
+ clock.tick(JDuration.ofSeconds(60))
+ val env5 = createEnvelope(PersistenceId(entityType, "a"), 2, "a2")
+ catchupPublisher.sendNext(env5)
+ outProbe.expectNext(env5)
+
+ val env6 = createEnvelope(PersistenceId(entityType, "a"), 3, "a3")
+ catchupPublisher.sendNext(env6)
+ outProbe.expectNoMessage() // catchup closed
+ firehosePublisher.sendNext(env6)
+ outProbe.expectNext(env6)
+ }
+
+ "track consumer progress" in new Setup {
+ // using two consumers
+ outProbe(0).request(2)
+ outProbe(1).request(2)
+
+ // FIXME is there a way to know that it has been added to the hub?
+ Thread.sleep(1000)
+
+ catchupPublisher(0).sendNext(allEnvelopes(0))
+ catchupPublisher(1).sendNext(allEnvelopes(0))
+ outProbe(0).expectNext(allEnvelopes(0))
+ outProbe(1).expectNext(allEnvelopes(0))
+
+ firehosePublisher.sendNext(allEnvelopes(0))
+ // this "sleep" is needed so that the firehose envelope is received first
+ outProbe.expectNoMessage()
+
+ catchupPublisher(0).sendNext(allEnvelopes(1))
+ catchupPublisher(1).sendNext(allEnvelopes(1))
+ outProbe(0).expectNext(allEnvelopes(1))
+ outProbe(1).expectNext(allEnvelopes(1))
+
+ val firehose =
+
eventsBySliceFirehose.getFirehose(FirehoseKey(EventsBySliceFirehoseQuery.Identifier,
entityType, sliceRange))
+ firehose.consumerTracking.size shouldBe 2
+ import scala.jdk.CollectionConverters._
+ firehose.consumerTracking.values.asScala.foreach { tracking =>
+ tracking.offsetTimestamp shouldBe
allEnvelopes(1).offset.asInstanceOf[TimestampOffset].timestamp
+ }
+
+ // only requesting for outProbe(0)
+ outProbe(0).request(100)
+ // less than BroadcastHub buffer size
+ val moreEnvelopes = (1 to 20).map(n =>
createEnvelope(PersistenceId(entityType, "x"), n, s"x$n"))
+ moreEnvelopes.foreach(firehosePublisher.sendNext)
+ outProbe(0).expectNextN(moreEnvelopes.size) shouldBe moreEnvelopes
+ firehose.consumerTracking.size shouldBe 2
+ firehose.consumerTracking.values.asScala
+ .count(_.offsetTimestamp ==
moreEnvelopes.last.offset.asInstanceOf[TimestampOffset].timestamp) shouldBe 1
+ }
+
+ "abort slow consumers" in new Setup {
+ // using two consumers
+ outProbe(0).request(3)
+ outProbe(1).request(3)
+
+ // FIXME is there a way to know that it has been added to the hub?
+ Thread.sleep(1000)
+
+ catchupPublisher(0).sendNext(allEnvelopes(0))
+ catchupPublisher(1).sendNext(allEnvelopes(0))
+ outProbe(0).expectNext(allEnvelopes(0))
+ outProbe(1).expectNext(allEnvelopes(0))
+
+ firehosePublisher.sendNext(allEnvelopes(0))
+ // this "sleep" is needed so that the firehose envelope is received first
+ outProbe.expectNoMessage()
+
+ catchupPublisher(0).sendNext(allEnvelopes(1))
+ catchupPublisher(1).sendNext(allEnvelopes(1))
+ outProbe(0).expectNext(allEnvelopes(1))
+ outProbe(1).expectNext(allEnvelopes(1))
+
+ // switch to firehose only
+ clock.tick(JDuration.ofSeconds(60))
+ // more than half BroadcastHub buffer size
+ val moreEnvelopes = (1 to 40).map { n =>
+ clock.tick(JDuration.ofSeconds(1))
+ createEnvelope(PersistenceId(entityType, "x"), n, s"x$n")
+ }
+ // both consumers will now switch to firehose only
+ catchupPublisher(0).sendNext(moreEnvelopes.head)
+ catchupPublisher(1).sendNext(moreEnvelopes.head)
+ outProbe(0).expectNext(moreEnvelopes.head)
+ outProbe(1).expectNext(moreEnvelopes.head)
+ // only requesting for outProbe(0)
+ outProbe(0).request(100)
+ moreEnvelopes.foreach(firehosePublisher.sendNext)
+ outProbe(0).expectNextN(moreEnvelopes.size) shouldBe moreEnvelopes
+ val firehose =
+
eventsBySliceFirehose.getFirehose(FirehoseKey(EventsBySliceFirehoseQuery.Identifier,
entityType, sliceRange))
+ clock.tick(JDuration.ofSeconds(6)) // simulate consumer lag
+ firehose.detectSlowConsumers(clock.instant())
+ clock.tick(JDuration.ofSeconds(2))
+ firehose.detectSlowConsumers(clock.instant())
+ clock.tick(JDuration.ofSeconds(2))
+ firehose.detectSlowConsumers(clock.instant())
+ outProbe(1).expectError().getClass shouldBe
classOf[SlowConsumerException]
+ }
+
+ "not abort consumers when fast" in new Setup {
+ val numberOfConsumers = 10
+ (0 until numberOfConsumers).foreach { i =>
+ outProbe(i).request(10000)
+ }
+
+ // FIXME is there a way to know that it has been added to the hub?
+ Thread.sleep(1000)
+
+ val moreEnvelopes = (1 to 100).map { n =>
+ createEnvelope(PersistenceId(entityType, "x"), n, s"x$n")
+ }
+
+ moreEnvelopes.foreach { env =>
+ firehosePublisher.sendNext(env)
+ (0 until numberOfConsumers).foreach { i =>
+ catchupPublisher(i).sendNext(env)
+ }
+ }
+
+ (0 until numberOfConsumers).foreach { i =>
+ // there may be duplicates
+ val received = outProbe(i).receiveWhile(max = 10.seconds, idle =
100.millis) { case OnNext(env) => env }
+ received.toSet shouldBe moreEnvelopes.toSet
+ }
+ }
+
+ "not close when catchup is closed" in new Setup {
+ allEnvelopes.foreach(catchupPublisher.sendNext)
+ outProbe.request(10)
+ catchupPublisher.sendComplete()
+ outProbe.expectNextN(envelopes.size) shouldBe envelopes
+ outProbe.expectNoMessage() // not OnComplete
+ }
+
+ "close when firehose is closed" in new Setup {
+ allEnvelopes.foreach(catchupPublisher.sendNext)
+ outProbe.request(10)
+ outProbe.expectNextN(envelopes.size) shouldBe envelopes
+ firehosePublisher.sendComplete()
+ outProbe.expectComplete()
+ }
+
+ "close when last consumer is removed, but more consumers can be added
later" in new Setup {
+ // using two consumers first
+ outProbe(0).request(10)
+ outProbe(1).request(10)
+
+ // FIXME is there a way to know that it has been added to the hub?
+ Thread.sleep(1000)
+
+ catchupPublisher(0).sendNext(allEnvelopes(0))
+ catchupPublisher(1).sendNext(allEnvelopes(0))
+ outProbe(0).expectNext(allEnvelopes(0))
+ outProbe(1).expectNext(allEnvelopes(0))
+ firehoseRunning.get shouldBe true
+
+ outProbe(0).cancel()
+ catchupPublisher(1).sendNext(allEnvelopes(1))
+ outProbe(1).expectNext(allEnvelopes(1))
+ firehoseRunning.get shouldBe true
+
+ outProbe(1).cancel()
+
+ // another consumer
+ outProbe(2).request(10)
+ // FIXME is there a way to know that it has been added to the hub?
+ Thread.sleep(1000)
+ catchupPublisher(2).sendNext(allEnvelopes(2))
+ outProbe(2).expectNext(allEnvelopes(2))
+ firehoseRunning.get shouldBe true
+
+ outProbe(2).cancel()
+ // after a while the firehose will be shutdown
+ eventually {
+ firehoseRunning.get shouldBe false
+ }
+
+ // FIXME add another (more integration test) that verifies similar that
consumers can be added to the
+ // EventsBySliceFirehose extension after the firehose has been shutdown
+ }
+
+ "deduplicate when emitting from both" in new Setup {
+ override def pluginId = "events-by-slice-firehose-with-deduplication"
+
+ catchupPublisher.sendNext(allEnvelopes(0))
+ catchupPublisher.sendNext(allEnvelopes(1))
+ outProbe.request(10)
+ outProbe.expectNext(allEnvelopes(0))
+ outProbe.expectNext(allEnvelopes(1))
+
+ firehosePublisher.sendNext(allEnvelopes(2))
+ outProbe.expectNoMessage()
+
+ catchupPublisher.sendNext(allEnvelopes(2))
+ outProbe.expectNext(allEnvelopes(2)) // still from catchup
+
+ firehosePublisher.sendNext(allEnvelopes(3))
+ outProbe.expectNext(allEnvelopes(3)) // from firehose
+
+ catchupPublisher.sendNext(allEnvelopes(3))
+ outProbe.expectNoMessage() // duplicate from catchup
+
+
clock.setInstant(allEnvelopes.last.offset.asInstanceOf[TimestampOffset].timestamp)
+ val env4 = createEnvelope(PersistenceId(entityType, "a"), 2, "a2")
+ catchupPublisher.sendNext(env4)
+ firehosePublisher.sendNext(env4)
+ outProbe.expectNext(env4)
+ outProbe.expectNoMessage() // deduplicate
+
+ clock.tick(JDuration.ofSeconds(60))
+ val env5 = createEnvelope(PersistenceId(entityType, "b"), 2, "b2")
+ catchupPublisher.sendNext(env5)
+ outProbe.expectNext(env5)
+
+ val env6 = createEnvelope(PersistenceId(entityType, "b"), 3, "b3")
+ catchupPublisher.sendNext(env6)
+ outProbe.expectNoMessage() // catchup closed
+ firehosePublisher.sendNext(env6)
+ outProbe.expectNext(env6)
+ }
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
