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-persistence-cassandra.git
The following commit(s) were added to refs/heads/main by this push:
new 0e96d77 add max-buffer-size for tag writing (#478)
0e96d77 is described below
commit 0e96d77b78487ea4788d847ee75a0f6461eff6e0
Author: PJ Fanning <[email protected]>
AuthorDate: Sun Sep 6 12:43:31 2026 +0100
add max-buffer-size for tag writing (#478)
* add max-buffer-size for tag writing
* add tests
* fail tag writes explicitly when the buffer limit is reached
Motivation:
The buffer limit added in this branch dropped the TagWrite message and
relied on the comment "the sender's ask will timeout and the journal will
retry". There is no retry. CassandraJournal.asyncWriteMessages does
`t.ask(extractTagWrites(...))` with tag-write-timeout, and TagWriters
forwards with askTagActor, so a dropped message means the journal stalls
for the full timeout (4s by default) and then fails the write, stopping
the persistent actor. Repeating that for every write while the buffer is
full turns memory pressure into a slow storm of actor failures.
The limit was also measured inconsistently: the idle branch used
buffer.size while the write-in-progress branch used a new Buffer.pendingSize
that folded over the pending vector on every message, which is O(n) in
exactly the situation the limit exists for.
Buffer.size itself could not be used as it stood because it has two
accounting bugs. rebuild carried leftover writes into pending without
counting their events, so size undercounted after a write completed with
events still pending. remove dropped a pid's pending writes without
subtracting them, so size overcounted afterwards. Both also affect the
existing "buffer for tagged events is getting too large" warning.
Modification:
- Fix Buffer.size accounting. rebuild now takes the total from the caller,
which is arithmetic rather than a fold, so completing a write stays cheap
when pending is large. remove partitions pending and subtracts what it
removed. Buffer.pendingSize is gone.
- Reject an over-limit write by replying Status.Failure with a
TagWriterBufferFullException instead of dropping the message. The journal
sees the failure immediately rather than after tag-write-timeout, and the
cause names the tag and the sizes.
- Check the limit before assigning tag pid sequence numbers, in both
branches, so a rejected write consumes none. A consumed sequence nr would
leave a gap in tag_views that stalls eventsByTag until repaired.
- Measure with buffer.size in both branches. It includes the batch being
written, which is held in memory and so counts towards the bound.
- Default max-buffer-size to 0, meaning no limit, so behaviour is unchanged
unless the limit is opted into. Rejection stops persistent actors, which
is not something to turn on for everyone by default.
- Document the setting, including that recovery, not the normal write path,
is what makes the buffer grow.
Result:
The buffer can be bounded, and reaching the bound produces an immediate,
attributable failure rather than a silent drop followed by a timeout. No
events are lost: rejected writes leave no gap and are written by tag
scanning when the persistent actor recovers. Existing deployments are
unaffected until they set max-buffer-size.
Tests:
- Added TagWriterSpec cases: rejection fails the sender with
TagWriterBufferFullException when idle and when a write is in progress, a
rejected write consumes no tag pid sequence nr, writes are accepted again
once the buffer drains, and nothing is rejected when the limit is 0.
- Added BufferSpec cases for both size accounting bugs.
- sbt "core/testOnly ...BufferSpec ...TagWriterSpec ...TagWritersSpec" - 48
passed. The existing "handle being overloaded" case also guards the cost
of the accounting fix; a fold over pending there made it quadratic and it
did not complete.
- sbt +mimaReportBinaryIssues - no issues
- sbt "+core/Test/compile" - Scala 2.13.18 and 3.3.8 compile
- sbt docs/paradox - builds
- sbt scalafmtAll headerCreateAll - clean
- Cassandra integration tests - Not run, no Cassandra or Docker available
locally
References:
Refs #478
* accept unlimited as max-buffer-size value
Support `max-buffer-size = unlimited` (and `off`/`false`) alongside `0`
for no limit, following the spelling already used by other events-by-tag
settings, and make `unlimited` the default so the reference.conf value is
self-describing.
---------
Co-authored-by: PJ Fanning <[email protected]>
---
core/src/main/resources/reference.conf | 18 +++
.../cassandra/EventsByTagSettings.scala | 13 ++
.../persistence/cassandra/journal/Buffer.scala | 22 +++-
.../persistence/cassandra/journal/TagWriter.scala | 99 +++++++++++----
.../cassandra/CassandraPluginSettingsSpec.scala | 22 ++++
.../persistence/cassandra/journal/BufferSpec.scala | 32 +++++
.../cassandra/journal/TagWriterSpec.scala | 133 ++++++++++++++++++++-
.../cassandra/journal/TagWritersSpec.scala | 1 +
docs/src/main/paradox/events-by-tag.md | 24 ++++
9 files changed, 335 insertions(+), 29 deletions(-)
diff --git a/core/src/main/resources/reference.conf
b/core/src/main/resources/reference.conf
index 0aaf24f..b1483b6 100644
--- a/core/src/main/resources/reference.conf
+++ b/core/src/main/resources/reference.conf
@@ -250,6 +250,24 @@ pekko.persistence.cassandra {
# size and this should be reduced if that warning is seen.
max-message-batch-size = 150
+ # Hard upper bound on the number of events buffered by a single tag writer.
+ # Set to "unlimited" (the default), "off" or 0 for no limit, which is the
+ # behaviour of earlier versions.
+ #
+ # Normal writes are already bounded: the journal waits for the tag write
to be
+ # acknowledged, so a persistent actor has at most one write outstanding.
Recovery
+ # sends tag writes without waiting for acknowledgement, so a persistence
id with a
+ # lot of tagged events to replay is what can make this buffer grow.
+ #
+ # When the limit is reached the tag write is rejected and the failure is
returned to
+ # the journal, which fails the write for that persistent actor. Nothing is
silently
+ # dropped: rejected events consume no tag pid sequence nrs and are written
by tag
+ # scanning when the persistent actor next recovers. Enable this only if an
unbounded
+ # buffer is an actual risk for you, as the rejection stops persistent
actors.
+ #
+ # Should be comfortably larger than max-message-batch-size.
+ max-buffer-size = unlimited
+
# Max time to buffer events for before writing.
# Larger values will increase cassandra write efficiency but increase the
delay before
# seeing events in EventsByTag queries.
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/EventsByTagSettings.scala
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/EventsByTagSettings.scala
index f02fd06..2f82cfb 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/EventsByTagSettings.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/EventsByTagSettings.scala
@@ -154,8 +154,11 @@ import com.typesafe.config.Config
val tagWriteTimeout = eventsByTagConfig.getDuration("tag-write-timeout",
TimeUnit.MILLISECONDS).millis
+ val maxBufferSize: Int = unlimitedInt(eventsByTagConfig, "max-buffer-size")
+
val tagWriterSettings = TagWriterSettings(
eventsByTagConfig.getInt("max-message-batch-size"),
+ maxBufferSize,
eventsByTagConfig.getDuration("flush-interval",
TimeUnit.MILLISECONDS).millis,
eventsByTagConfig.getDuration("scanning-flush-interval",
TimeUnit.MILLISECONDS).millis,
eventsByTagConfig.getDuration("stop-tag-writer-when-idle",
TimeUnit.MILLISECONDS).millis,
@@ -218,6 +221,16 @@ import com.typesafe.config.Config
val maxMissingToSearch: Long =
eventsByTagConfig.getLong("max-missing-to-search")
+ /**
+ * An int setting where no limit can be spelled either as `0` or as
`unlimited`/`off`. Returns `0` for no limit.
+ */
+ private def unlimitedInt(cfg: Config, path: String): Int = {
+ cfg.getString(path).toLowerCase match {
+ case "unlimited" | "off" | "false" => 0
+ case _ => cfg.getInt(path)
+ }
+ }
+
private def optionalDuration(cfg: Config, path: String):
Option[FiniteDuration] = {
cfg.getString(path).toLowerCase match {
case "off" | "false" => None
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/Buffer.scala
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/Buffer.scala
index 7ac26e0..95b9f48 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/Buffer.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/Buffer.scala
@@ -44,8 +44,8 @@ private[pekko] case class Buffer(
def remove(pid: String): Buffer = {
val (toFilter, without) =
nextBatch.partition(_.events.head._1.persistenceId == pid)
- val filteredPending = pending.filterNot(_.events.head._1.persistenceId ==
pid)
- val removed = toFilter.foldLeft(0)((acc, next) => acc + next.events.size)
+ val (removedPending, filteredPending) =
pending.partition(_.events.head._1.persistenceId == pid)
+ val removed = Buffer.eventCount(toFilter) +
Buffer.eventCount(removedPending)
copy(size = size - removed, nextBatch = without, pending = filteredPending)
}
@@ -80,7 +80,7 @@ private[pekko] case class Buffer(
// rare case where events have been received out of order, just
re-build the buffer
require(pending.isEmpty)
val allWrites = (nextBatch :+
write).sortBy(_.events.head._1.timeUuid)(timeUuidOrdering)
- rebuild(allWrites)
+ rebuild(allWrites, newSize)
} else if (nextBatch.headOption.exists(_.events.head._1.timeBucket !=
write.events.head._1.timeBucket)) {
// time bucket has changed
copy(size = newSize, pending = pending :+ write, writeRequired = true)
@@ -104,7 +104,12 @@ private[pekko] case class Buffer(
}
}
- private def rebuild(writes: Vector[AwaitingWrite]): Buffer = {
+ /**
+ * `totalSize` is the number of events in `writes`. It is passed in rather
than counted here because
+ * `writes` can be very large when the database is falling behind, and this
runs on every completed
+ * write.
+ */
+ private def rebuild(writes: Vector[AwaitingWrite], totalSize: Int): Buffer =
{
var buffer = Buffer.empty(batchSize)
var i = 0
while (!buffer.shouldWrite() && i < writes.size) {
@@ -112,7 +117,7 @@ private[pekko] case class Buffer(
i += 1
}
// pending may have one in it as the last one may have been a time
bucket change rather than bach full
- val done = buffer.copy(pending = buffer.pending ++ writes.drop(i))
+ val done = buffer.copy(size = totalSize, pending = buffer.pending ++
writes.drop(i))
done
}
@@ -123,7 +128,8 @@ private[pekko] case class Buffer(
def writeComplete(): Buffer = {
// this could be more efficient by adding until a write is required but
this is simpler and
// pending is expected to be small unless the database is falling behind
- rebuild(pending)
+ // nextBatch has just been written and is bounded by batchSize, so what is
left is the rest of size
+ rebuild(pending, size - Buffer.eventCount(nextBatch))
}
}
@@ -132,6 +138,10 @@ private[pekko] case class Buffer(
*/
@InternalApi
private[pekko] object Buffer {
+
+ private def eventCount(writes: Vector[AwaitingWrite]): Int =
+ writes.foldLeft(0)((acc, next) => acc + next.events.size)
+
def empty(batchSize: Int): Buffer = {
require(batchSize > 0)
Buffer(batchSize, 0, Vector.empty, Vector.empty, writeRequired = false)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TagWriter.scala
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TagWriter.scala
index 33a7ab3..8bc083a 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TagWriter.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/journal/TagWriter.scala
@@ -17,7 +17,16 @@ import java.util.UUID
import org.apache.pekko
import pekko.Done
-import pekko.actor.{ Actor, ActorLogging, ActorRef,
NoSerializationVerificationNeeded, Props, ReceiveTimeout, Timers }
+import pekko.actor.{
+ Actor,
+ ActorLogging,
+ ActorRef,
+ NoSerializationVerificationNeeded,
+ Props,
+ ReceiveTimeout,
+ Status,
+ Timers
+}
import pekko.annotation.InternalApi
import pekko.cluster.pubsub.{ DistributedPubSub, DistributedPubSubMediator }
import pekko.event.LoggingAdapter
@@ -29,6 +38,7 @@ import
pekko.persistence.cassandra.journal.TagWriters.TagWritersSession
import pekko.util.{ OptionVal, UUIDComparator }
import scala.concurrent.duration.{ Duration, FiniteDuration, _ }
+import scala.util.control.NoStackTrace
import scala.util.control.NonFatal
import scala.util.{ Failure, Success, Try }
@@ -43,7 +53,6 @@ import scala.util.{ Failure, Success, Try }
* Prevents any concurrent writes.
*
* Possible improvements:
- * - Max buffer size
* - Optimize sorting given they are nearly sorted
*/
@InternalApi private[pekko] object TagWriter {
@@ -51,8 +60,20 @@ import scala.util.{ Failure, Success, Try }
private[pekko] def props(settings: TagWriterSettings, session:
TagWritersSession, tag: Tag, parent: ActorRef): Props =
Props(new TagWriter(settings, session, tag, parent))
+ /**
+ * Returned to the sender when a tag write can not be buffered because
`max-buffer-size` has been
+ * reached. The write is not buffered and no tag pid sequence nrs are
consumed by it, so the events
+ * are picked up again by tag scanning when the persistent actor next
recovers.
+ */
+ private[pekko] final class TagWriterBufferFullException(val tag: Tag, val
bufferSize: Int, val maxBufferSize: Int)
+ extends RuntimeException(
+ s"Tag write for tag [$tag] rejected, buffer is full [$bufferSize >=
$maxBufferSize]. " +
+ "Cassandra is not keeping up with tagged writes.")
+ with NoStackTrace
+
private[pekko] case class TagWriterSettings(
maxBatchSize: Int,
+ maxBufferSize: Int,
flushInterval: FiniteDuration,
scanningFlushInterval: FiniteDuration,
stopTagWriterWhenIdle: FiniteDuration,
@@ -143,6 +164,30 @@ import scala.util.{ Failure, Success, Try }
var lastLoggedBufferNs: Long = -1
val bufferWarningMinDurationNs: Long = 5.seconds.toNanos
+ var lastLoggedBufferFullNs: Long = -1
+
+ private def bufferFull(buffer: Buffer): Boolean =
+ settings.maxBufferSize > 0 && buffer.size >= settings.maxBufferSize
+
+ /**
+ * Fails the sender's ask rather than dropping the write silently, so that
the journal sees the
+ * rejection straight away instead of waiting out `tag-write-timeout`.
Nothing is buffered and no tag
+ * pid sequence nrs are consumed, so the events are not lost: tag scanning
picks them up when the
+ * persistent actor next recovers.
+ */
+ private def rejectWrite(buffer: Buffer, replyTo: ActorRef): Unit = {
+ val now = System.nanoTime()
+ if (now > (lastLoggedBufferFullNs + bufferWarningMinDurationNs)) {
+ lastLoggedBufferFullNs = now
+ log.error(
+ "Buffer for tagged events is full ({} >= {}) for tag [{}], rejecting
writes. Is Cassandra responsive? " +
+ "Are writes failing? Rejected events will be written by tag scanning
when the persistent actor recovers.",
+ buffer.size,
+ settings.maxBufferSize,
+ tag)
+ }
+ replyTo ! Status.Failure(new TagWriter.TagWriterBufferFullException(tag,
buffer.size, settings.maxBufferSize))
+ }
override def preStart(): Unit = {
log.debug("Running TagWriter for [{}] with settings {}", tag, settings)
@@ -171,12 +216,17 @@ import scala.util.{ Failure, Success, Try }
sender() ! FlushComplete
}
case TagWrite(_, payload, _) =>
- val (newTagPidSequenceNrs, events: Seq[(Serialized, TagPidSequenceNr)])
= {
- assignTagPidSequenceNumbers(payload.toVector, tagPidSequenceNrs)
+ // checked before any tag pid sequence nrs are assigned so that a
rejected write consumes none
+ if (bufferFull(buffer)) {
+ rejectWrite(buffer, sender())
+ } else {
+ val (newTagPidSequenceNrs, events: Seq[(Serialized,
TagPidSequenceNr)]) = {
+ assignTagPidSequenceNumbers(payload.toVector, tagPidSequenceNrs)
+ }
+ val newWrite = AwaitingWrite(events, OptionVal(sender()))
+ val newBuffer = buffer.add(newWrite)
+ flushIfRequired(newBuffer, newTagPidSequenceNrs)
}
- val newWrite = AwaitingWrite(events, OptionVal(sender()))
- val newBuffer = buffer.add(newWrite)
- flushIfRequired(newBuffer, newTagPidSequenceNrs)
case twd: TagWriteDone =>
log.error("Received Done when in idle state. This is a bug. Please
report with DEBUG logs: {}", twd)
case ResetPersistenceId(_, tp @ TagProgress(pid, _, tagPidSequenceNr)) =>
@@ -208,22 +258,27 @@ import scala.util.{ Failure, Success, Try }
log.debug("External flush while write in progress. Will flush after
write complete")
become(writeInProgress(buffer, tagPidSequenceNrs, Some(sender())))
case TagWrite(_, payload, _) =>
- val (updatedTagPidSequenceNrs, events) =
- assignTagPidSequenceNumbers(payload.toVector, tagPidSequenceNrs)
- val awaitingWrite = AwaitingWrite(events, OptionVal(sender()))
- val now = System.nanoTime()
- if (buffer.size > (4 * settings.maxBatchSize) && now >
(lastLoggedBufferNs + bufferWarningMinDurationNs)) {
- lastLoggedBufferNs = now
- log.warning(
- "Buffer for tagged events is getting too large ({}), is Cassandra
responsive? Are writes failing? " +
- "If events are buffered for longer than the
eventual-consistency-delay they won't be picked up by live queries. The oldest
event in the buffer is offset: {}",
- buffer.size,
- formatOffset(buffer.nextBatch.head.events.head._1.timeUuid))
+ // checked before any tag pid sequence nrs are assigned so that a
rejected write consumes none
+ if (bufferFull(buffer)) {
+ rejectWrite(buffer, sender())
+ } else {
+ val (updatedTagPidSequenceNrs, events) =
+ assignTagPidSequenceNumbers(payload.toVector, tagPidSequenceNrs)
+ val now = System.nanoTime()
+ val awaitingWrite = AwaitingWrite(events, OptionVal(sender()))
+ if (buffer.size > (4 * settings.maxBatchSize) && now >
(lastLoggedBufferNs + bufferWarningMinDurationNs)) {
+ lastLoggedBufferNs = now
+ log.warning(
+ "Buffer for tagged events is getting too large ({}), is Cassandra
responsive? Are writes failing? " +
+ "If events are buffered for longer than the
eventual-consistency-delay they won't be picked up by live queries. The oldest
event in the buffer is offset: {}",
+ buffer.size,
+ formatOffset(buffer.nextBatch.head.events.head._1.timeUuid))
+ }
+ // buffer until current query is finished
+ // Don't sort until the write has finished
+ val newBuffer = buffer.addPending(awaitingWrite)
+ become(writeInProgress(newBuffer, updatedTagPidSequenceNrs,
awaitingFlush))
}
- // buffer until current query is finished
- // Don't sort until the write has finished
- val newBuffer = buffer.addPending(awaitingWrite)
- become(writeInProgress(newBuffer, updatedTagPidSequenceNrs,
awaitingFlush))
case TagWriteDone(summary, doneNotify) =>
log.debug("Tag write done: {}", summary)
val nextBuffer = buffer.writeComplete()
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/CassandraPluginSettingsSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/CassandraPluginSettingsSpec.scala
index ce97bcb..7b9acab 100644
---
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/CassandraPluginSettingsSpec.scala
+++
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/CassandraPluginSettingsSpec.scala
@@ -256,4 +256,26 @@ class CassandraPluginSettingsSpec
}
}
+ "An EventsByTagSettings" must {
+
+ def settingsWithMaxBufferSize(value: String): EventsByTagSettings =
+ new EventsByTagSettings(
+ system,
+ ConfigFactory.parseString(s"events-by-tag.max-buffer-size =
$value").withFallback(defaultConfig))
+
+ "default max-buffer-size to no limit" in {
+ new EventsByTagSettings(system, defaultConfig).maxBufferSize must be(0)
+ }
+
+ "parse max-buffer-size as a number" in {
+ settingsWithMaxBufferSize("100000").maxBufferSize must be(100000)
+ }
+
+ "parse max-buffer-size no limit aliases" in {
+ forAll(Table("value", "unlimited", "UNLIMITED", "off", "false", "0")) {
value =>
+ settingsWithMaxBufferSize(value).maxBufferSize must be(0)
+ }
+ }
+ }
+
}
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/BufferSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/BufferSpec.scala
index 62e08e8..41d9aef 100644
---
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/BufferSpec.scala
+++
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/BufferSpec.scala
@@ -183,6 +183,38 @@ class BufferSpec extends AnyWordSpec with Matchers with
BeforeAndAfterAll {
}
writes shouldEqual totalWrites / 2
}
+
+ // size is what the tag writer reports and limits on, so it has to stay
equal to the number of
+ // events actually held in nextBatch plus pending
+ "keep size accurate when a write completes with events still pending" in {
+ val bucket = nowBucket()
+ var buffer = Buffer.empty(2)
+ for (i <- 1 to 5) {
+ buffer = buffer.add(awNoSender((event("p1", seqNr = i, payload =
"cats", bucket), i)))
+ }
+ buffer.size shouldEqual 5
+
+ // two events are written, leaving three, of which one does not fit in
the rebuilt batch
+ val afterWrite = buffer.writeComplete()
+ afterWrite.size shouldEqual 3
+ afterWrite.nextBatch.map(_.events.size).sum +
afterWrite.pending.map(_.events.size).sum shouldEqual 3
+ }
+
+ "keep size accurate when a persistence id with pending events is removed"
in {
+ val bucket = nowBucket()
+ val buffer = Buffer
+ .empty(2)
+ .add(awNoSender((event("p1", seqNr = 1, payload = "cats", bucket), 1)))
+ .add(awNoSender((event("p1", seqNr = 2, payload = "cats", bucket), 2)))
+ .add(awNoSender((event("p2", seqNr = 1, payload = "dogs", bucket), 1)))
+ .add(awNoSender((event("p1", seqNr = 3, payload = "cats", bucket), 3)))
+ buffer.size shouldEqual 4
+
+ // removes two from nextBatch and one from pending, leaving only p2
+ val afterRemove = buffer.remove("p1")
+ afterRemove.size shouldEqual 1
+ afterRemove.nextBatch.map(_.events.size).sum +
afterRemove.pending.map(_.events.size).sum shouldEqual 1
+ }
}
override protected def afterAll(): Unit = {
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/TagWriterSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/TagWriterSpec.scala
index 9ee9a98..5ab6e0b 100644
---
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/TagWriterSpec.scala
+++
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/TagWriterSpec.scala
@@ -17,7 +17,8 @@ import java.nio.ByteBuffer
import java.util.UUID
import org.apache.pekko
import pekko.Done
-import pekko.actor.{ ActorRef, ActorSystem }
+import pekko.actor.{ ActorRef, ActorSystem, Status }
+import pekko.event.Logging
import pekko.event.Logging.Warning
import pekko.persistence.cassandra.Day
import pekko.persistence.cassandra.journal.CassandraJournal._
@@ -78,6 +79,7 @@ class TagWriterSpec
val successfulWrite: Statement[?] => Future[Done] = _ =>
Future.successful(Done)
val defaultSettings = TagWriterSettings(
maxBatchSize = 10,
+ maxBufferSize = 0,
flushInterval = 10.seconds,
scanningFlushInterval = 20.seconds,
stopTagWriterWhenIdle = 5.seconds,
@@ -102,6 +104,11 @@ class TagWriterSpec
implicit val senderRef: ActorRef = sender.ref
}
+ private def expectBufferFullLogged(): Unit =
+ logProbe.expectMsgPF(waitDuration) {
+ case Logging.Error(_, _, _, msg) if msg.toString.contains("Buffer for
tagged events is full") => ()
+ }
+
"Tag writer batching" must {
"external flush when idle" in new Setup {
@@ -672,6 +679,130 @@ class TagWriterSpec
}
+ "Tag writer buffer limit" must {
+
+ "reject a write and fail the sender when the buffer is full" in new Setup {
+ val (probe, ref) =
+ setup(settings = defaultSettings.copy(maxBatchSize = 100,
maxBufferSize = 1, flushInterval = 1.hour))
+ val bucket = nowBucket()
+ val e1 = event("p1", 1L, "e-1", bucket)
+ val e2 = event("p2", 1L, "e-2", bucket)
+
+ // fills the buffer, nothing is written yet as the batch is not full
+ ref ! TagWrite(tagName, Vector(e1))
+ probe.expectNoMessage(waitDuration)
+
+ ref ! TagWrite(tagName, Vector(e2))
+ val failure = sender.expectMsgType[Status.Failure]
+
assert(failure.cause.isInstanceOf[TagWriter.TagWriterBufferFullException])
+ probe.expectNoMessage(waitDuration)
+ expectBufferFullLogged()
+ }
+
+ "reject a write and fail the sender when the buffer is full during a
write" in new Setup {
+ val writePromise = Promise[Done]()
+ val (probe, ref) = setup(
+ settings = defaultSettings.copy(maxBatchSize = 1, maxBufferSize = 2,
flushInterval = 1.hour),
+ writeResponse = LazyList(writePromise.future) ++
LazyList.continually(Future.successful(Done)))
+ val bucket = nowBucket()
+ val e1 = event("p1", 1L, "e-1", bucket)
+ val e2 = event("p2", 1L, "e-2", bucket)
+ val e3 = event("p3", 1L, "e-3", bucket)
+
+ // maxBatchSize is 1 so this write starts immediately and stays in the
buffer while in flight
+ ref ! TagWrite(tagName, Vector(e1))
+ probe.expectMsg(Vector(toEw(e1, 1)))
+
+ // buffer holds the in flight batch, one more still fits
+ ref ! TagWrite(tagName, Vector(e2))
+ probe.expectNoMessage(waitDuration)
+
+ ref ! TagWrite(tagName, Vector(e3))
+ val failure = sender.expectMsgType[Status.Failure]
+
assert(failure.cause.isInstanceOf[TagWriter.TagWriterBufferFullException])
+ expectBufferFullLogged()
+
+ // the accepted write is unaffected by the rejection
+ writePromise.success(Done)
+ sender.expectMsg(Done)
+ probe.expectMsg(ProgressWrite("p1", 1, 1, e1.timeUuid))
+ probe.expectMsg(Vector(toEw(e2, 1)))
+ probe.expectMsg(ProgressWrite("p2", 1, 1, e2.timeUuid))
+ sender.expectMsg(Done)
+ }
+
+ "not consume a tag pid sequence nr for a rejected write" in new Setup {
+ val (probe, ref) =
+ setup(settings = defaultSettings.copy(maxBatchSize = 100,
maxBufferSize = 1, flushInterval = 1.hour))
+ val bucket = nowBucket()
+ val e1 = event("p1", 1L, "e-1", bucket)
+ val e2 = event("p1", 2L, "e-2", bucket)
+
+ ref ! TagWrite(tagName, Vector(e1))
+ probe.expectNoMessage(waitDuration)
+
+ // rejected, so it must not take tag pid sequence nr 2
+ ref ! TagWrite(tagName, Vector(e2))
+ sender.expectMsgType[Status.Failure]
+ expectBufferFullLogged()
+
+ ref ! Flush
+ probe.expectMsg(Vector(toEw(e1, 1)))
+ probe.expectMsg(ProgressWrite("p1", 1, 1, e1.timeUuid))
+ sender.expectMsg(Done)
+ sender.expectMsg(FlushComplete)
+
+ // buffer has drained, the retry gets the sequence nr the rejected write
did not take.
+ // a gap here would stall eventsByTag until tag_views was repaired
+ ref ! TagWrite(tagName, Vector(e2))
+ ref ! Flush
+ probe.expectMsg(Vector(toEw(e2, 2)))
+ probe.expectMsg(ProgressWrite("p1", 2, 2, e2.timeUuid))
+ sender.expectMsg(Done)
+ sender.expectMsg(FlushComplete)
+ }
+
+ "accept writes again once the buffer has drained" in new Setup {
+ val (probe, ref) =
+ setup(settings = defaultSettings.copy(maxBatchSize = 100,
maxBufferSize = 1, flushInterval = 1.hour))
+ val bucket = nowBucket()
+ val e1 = event("p1", 1L, "e-1", bucket)
+ val e2 = event("p2", 1L, "e-2", bucket)
+ val e3 = event("p3", 1L, "e-3", bucket)
+
+ ref ! TagWrite(tagName, Vector(e1))
+ ref ! TagWrite(tagName, Vector(e2))
+ sender.expectMsgType[Status.Failure]
+ expectBufferFullLogged()
+
+ ref ! Flush
+ probe.expectMsg(Vector(toEw(e1, 1)))
+ probe.expectMsg(ProgressWrite("p1", 1, 1, e1.timeUuid))
+ sender.expectMsg(Done)
+ sender.expectMsg(FlushComplete)
+
+ ref ! TagWrite(tagName, Vector(e3))
+ ref ! Flush
+ probe.expectMsg(Vector(toEw(e3, 1)))
+ probe.expectMsg(ProgressWrite("p3", 1, 1, e3.timeUuid))
+ sender.expectMsg(Done)
+ sender.expectMsg(FlushComplete)
+ }
+
+ "not reject anything when max-buffer-size is 0" in new Setup {
+ val (probe, ref) =
+ setup(settings = defaultSettings.copy(maxBatchSize = 100,
maxBufferSize = 0, flushInterval = 1.hour))
+ val bucket = nowBucket()
+
+ (1 to 10).foreach { i =>
+ ref ! TagWrite(tagName, Vector(event(s"p$i", 1L, s"e-$i", bucket)))
+ }
+
+ sender.expectNoMessage(waitDuration)
+ probe.expectNoMessage(waitDuration)
+ }
+ }
+
"Tag writer error scenarios" must {
"handle tag writes view failing" in new Setup {
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/TagWritersSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/TagWritersSpec.scala
index 7633221..053d1a9 100644
---
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/TagWritersSpec.scala
+++
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/journal/TagWritersSpec.scala
@@ -54,6 +54,7 @@ class TagWritersSpec
private val defaultSettings = TagWriterSettings(
maxBatchSize = 10,
+ maxBufferSize = 0,
flushInterval = 10.seconds,
scanningFlushInterval = 20.seconds,
stopTagWriterWhenIdle = 5.seconds,
diff --git a/docs/src/main/paradox/events-by-tag.md
b/docs/src/main/paradox/events-by-tag.md
index acc230f..e3df245 100644
--- a/docs/src/main/paradox/events-by-tag.md
+++ b/docs/src/main/paradox/events-by-tag.md
@@ -192,6 +192,30 @@ be taken not to have batches that will be rejected by
Cassandra. Two other cases
* Periodically: By default 250ms. To prevent eventsByTag queries being too out
of date.
* When a starting a new timebucket, which translates to a new partition in
Cassandra, the events for the old timebucket are written.
+### Limiting the tag write buffer
+
+Each tag has a writer actor that buffers events until they are batched into a
write. Normal writes are already
+bounded: the journal waits for the tag write to be acknowledged before
completing the persist, so a persistent
+actor has at most one write outstanding. Recovery is different — it sends tag
writes without waiting for
+acknowledgement, so a persistence id replaying a lot of tagged events is what
can make the buffer grow.
+
+`max-buffer-size` puts a hard upper bound on the number of events one tag
writer will hold. It is `unlimited` by
+default, which means no limit; `off` and `0` mean the same thing.
+
+```
+pekko.persistence.cassandra.events-by-tag.max-buffer-size = 100000
+```
+
+When the limit is reached the tag write is rejected and the failure is
returned to the journal, which fails the
+write for that persistent actor and stops it. Nothing is dropped silently: a
rejected write consumes no tag pid
+sequence nrs, so it leaves no gap in `tag_views`, and the events are written
by tag scanning when the persistent
+actor next recovers.
+
+That is a disruptive outcome, so only set this if an unbounded buffer is a
real risk for you — it trades stopped
+persistent actors for bounded memory. Set it comfortably above
`max-message-batch-size`, as the batch currently
+being written counts towards the buffer. The existing warning about the buffer
getting too large is logged well
+before the limit is reached and is the signal to investigate first.
+
## Cleanup of tag_views table
By default the tag_views table keeps tagged events indefinitely, even when the
original events have been removed.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
