This is an automated email from the ASF dual-hosted git repository.
He-Pin 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 c04735c301 feat: improve EventWriter: fill sequence number gaps,
handle lookup failures and snapshot events, bypass the circuit breaker (#3653)
c04735c301 is described below
commit c04735c3013475a6a0b9e64d70696b6ccec636f4
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 14:31:08 2026 +0100
feat: improve EventWriter: fill sequence number gaps, handle lookup
failures and snapshot events, bypass the circuit breaker (#3653)
* feat: fill sequence number gaps in EventWriter
Motivation:
EventWriter is used to write events produced elsewhere (for example
replicated or filtered events) into the journal. When some events are
filtered out upstream the sequence numbers that arrive have gaps, and the
journal then ends up with gaps for the persistence id, which breaks
replay and sequence number based consumers.
Modification:
- Add `pekko.persistence.typed.event-writer.fill-sequence-number-gaps`
(default off) and `latest-sequence-number-cache-capacity` (default
1000) config, and corresponding EventWriterSettings fields.
- When enabled, track the latest written sequence number per persistence
id, look up the highest sequence number from the journal when it is not
known, and fill detected gaps with FilteredPayload events written in
the same atomic write as the event after the gap.
- When enabled, acknowledge writes of sequence numbers lower than the
next expected one as duplicates without writing them again.
- Evict least recently used idle persistence ids from the in-memory state
when the cache capacity is exceeded.
- Only count non-FilteredPayload events towards the max batch size.
- Keep a 2-arg EventWriterSettings constructor and apply for binary
compatibility; add MiMa excludes for the internal EventWriter changes.
- Add EventWriterFillGapsSpec and adjust EventWriterSpec.
Result:
EventWriter can optionally fill sequence number gaps with FilteredPayload;
default behaviour is unchanged.
Tests:
- Not run locally - left to CI; scalafmt and headerCreateAll applied
References:
None - EventWriter improvement for writing events with sequence number gaps
* fix: handle ReplayMessagesFailure in EventWriter
Motivation:
When a write fails, EventWriter asks the journal for the highest sequence
number of the persistence id to detect duplicate writes. If that lookup
failed with ReplayMessagesFailure (or the ask failed/timed out), the
writer threw an exception and was restarted, leaving the waiting
clients without a reply until their own ask timed out.
Modification:
- Map ReplayMessagesFailure and ask failures for the highest sequence
number lookup to a new internal MaxSeqNrForPidFailed command.
- On MaxSeqNrForPidFailed, reply with an error to all writes waiting for
the persistence id, log a warning and continue with any batched writes.
- Include the original write error in the unexpected-reply exception.
- Add tests for a failing highest sequence number lookup to
EventWriterSpec and EventWriterFillGapsSpec.
Result:
Clients get a prompt "Journal write failed" error instead of a timeout,
and the EventWriter is no longer restarted for this case.
Tests:
- Not run locally - left to CI; scalafmt applied
References:
None - EventWriter robustness improvement
* fix: fill sequence number gaps before snapshot event in EventWriter
Motivation:
When a snapshot is stored together with an event written through
EventWriter, the events before the snapshot event may not have been
written (for example because they were filtered). Recovery then finds a
gap before the snapshot sequence number. The gaps before such an event
must be filled even when general gap filling is disabled.
Modification:
- Add `isSnapshotEvent` to EventWriter.Write. Gaps before a snapshot
event are filled with FilteredPayload regardless of the
`fill-sequence-number-gaps` setting.
- Keep a 6-arg Write constructor and apply (not a snapshot event) for
binary compatibility; add MiMa excludes for the changed copy/unapply
of the internal Write message.
- Replace the optional error description of the internal highest
sequence number lookup with an AskMaxSeqNrReason (write failure, fill
gaps or snapshot event).
- Add snapshot event tests to EventWriterSpec.
Result:
Snapshot events written through EventWriter no longer leave sequence
number gaps in the journal.
Tests:
- Not run locally - left to CI; scalafmt applied
References:
None - EventWriter improvement for snapshot events
* feat: bypass journal circuit breaker for EventWriter writes
Motivation:
EventWriter relies on the journal rejecting duplicate writes (for
example duplicate key violations) and then checks the highest sequence
number to acknowledge the duplicates. These expected failures count
towards the journal circuit breaker, which can then open and fail
unrelated writes.
Modification:
- Add `bypassCircuitBreaker` (default false) to
JournalProtocol.WriteMessages; AsyncWriteJournal calls
asyncWriteMessages without the circuit breaker when it is set.
- WriteMessages is changed from a case class to a final class that keeps
the 3-arg constructor, 3-arg apply, 3-arg copy (keeping the flag),
product methods and pattern matching on the first 3 fields, with
Scala 2 and Scala 3 specific companion parents matching the former
case class companions (Function3/unapply returning Option on Scala 2,
Mirror.Product/unapply returning the instance on Scala 3), so journal
plugins and the TCK compiled against earlier versions keep working.
- EventWriter bypasses the circuit breaker, except after a failed
highest sequence number lookup, until the next successful write or
lookup.
- Use type patterns for WriteMessages in AsyncWriteJournal and
PersistencePluginProxy.
- Add WriteMessagesSpec and an EventWriterSpec test.
Result:
Expected duplicate write failures from EventWriter no longer trip the
journal circuit breaker, without breaking binary compatibility of
WriteMessages.
Tests:
- Not run locally - left to CI; scalafmt and headerCreateAll applied
References:
None - EventWriter and journal circuit breaker interaction
* docs: list the new event writer settings in the migration guide
Motivation:
The event writer gained two settings for filling sequence number gaps, but
the 1.x to 2.x migration guide only lists the existing event writer
settings.
Modification:
Add `fill-sequence-number-gaps` and `latest-sequence-number-cache-capacity`
to the persistence-typed part of the New configuration section.
Result:
The migration guide lists all new event writer settings.
Tests:
- Not run - docs only
References:
Refs #3653
* fix: address EventWriter review findings
Motivation:
Review of #3653 found several cases where EventWriter callers were answered
too early or not at all, that bypassed journal writes lost the circuit
breaker
call timeout, and that hashing a PersistentRepr without a sender threw.
Modification:
- A failed highest sequence number lookup (ReplayMessagesFailure or ask
timeout) fails every waiting write, including those queued for the lookup,
keeps the failure as the cause of "Journal write failed", and no longer
restarts the writer on an empty batch.
- Writes queued during a lookup keep their own isSnapshotEvent flag, so an
ordinary event queued behind a snapshot event is not gap filled when
fill-sequence-number-gaps is off.
- A duplicate of a write that is still pending is answered with the outcome
of
that write instead of being acknowledged right away.
- WriteMessageRejected fails the waiting writes ("Journal rejected write",
with
the rejection as cause), e.g. for a rejected gap fill batch.
- AsyncWriteJournal applies the circuit-breaker call-timeout to writes that
bypass the circuit breaker, so a hanging write cannot hold back the
responses
of later writes.
- PersistentImpl.hashCode accepts a null sender (ActorRef.noSender), with
the
same hash as before for other senders.
- Update the Lightbend notices of the files adapted from akka-core#32153 to
the
2023 source notices.
Result:
EventWriter callers always get a reply that matches the journal outcome.
Tests:
- New tests in EventWriterFillGapsSpec (initial lookup failure and timeout,
duplicates of pending writes, rejected gap fill batch), EventWriterSpec
(snapshot and ordinary events queued for one lookup),
AsyncWriteJournalResponseOrderSpec (hanging bypassed write) and
WriteMessagesSpec (hashCode without sender). They fail against the
previous
code.
- sbt "persistence/testOnly org.apache.pekko.persistence.WriteMessagesSpec
org.apache.pekko.persistence.journal.AsyncWriteJournalResponseOrderSpec" (9
passed)
- sbt "persistence-typed-tests/testOnly
org.apache.pekko.persistence.typed.internal.EventWriterSpec
org.apache.pekko.persistence.typed.internal.EventWriterFillGapsSpec" (45 passed)
- sbt "persistence/mimaReportBinaryIssues"
"persistence-typed/mimaReportBinaryIssues"
- sbt "persistence/headerCheckAll" "persistence-typed/headerCheckAll"
References:
https://github.com/apache/pekko/pull/3653
* fix: evict pid state after duplicate-only lookups and limit gapped batches
Motivation:
Further review of #3653 found that a highest sequence number lookup whose
queued writes were all duplicates left the idle pid state in the cache
without eviction, and that a gapped write added to a pending batch skipped
the max-batch-size check.
Modification:
- After replaying the writes queued for a lookup, an idle pid state goes
through handleUpdatedStateForPid, so latest-sequence-number-cache-capacity
eviction applies. The lookup also refreshes the state's used timestamp.
- A gapped write added to a pending batch is rejected with the max batch
error before filling, when the batch already holds max-batch-size real
events. Filled gaps still don't count towards the limit.
Result:
Duplicate-only writes no longer retain unbounded pid state, and gapped
writes get the same backpressure as ordinary writes.
Tests:
- New EventWriterFillGapsSpec tests (duplicate-only lookups for six pids
with cache capacity 2, gapped writes at max-batch-size 2). Both fail
against the previous code.
- sbt "persistence-typed-tests/testOnly
org.apache.pekko.persistence.typed.internal.EventWriter*" (47 passed)
- scalafmt on the changed files
References:
https://github.com/apache/pekko/pull/3653
---
.../paradox/migration/migration-guide-1.x-2.x.md | 5 +
.../typed/internal/EventWriterFillGapsSpec.scala | 567 +++++++++++++++++++++
.../typed/internal/EventWriterSpec.scala | 278 +++++++++-
.../eventwriter-fill-gaps.excludes | 27 +
.../eventwriter-snapshot-event.excludes | 22 +
.../src/main/resources/reference.conf | 5 +
.../persistence/typed/internal/EventWriter.scala | 541 +++++++++++++++++---
.../pekko/persistence/WriteMessagesCompanion.scala | 40 ++
.../pekko/persistence/WriteMessagesCompanion.scala | 44 ++
.../apache/pekko/persistence/JournalProtocol.scala | 91 +++-
.../org/apache/pekko/persistence/Persistent.scala | 3 +-
.../persistence/journal/AsyncWriteJournal.scala | 34 +-
.../journal/PersistencePluginProxy.scala | 15 +-
.../pekko/persistence/WriteMessagesSpec.scala | 80 +++
.../AsyncWriteJournalResponseOrderSpec.scala | 35 +-
15 files changed, 1668 insertions(+), 119 deletions(-)
diff --git a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
index 8947550241..b09caba440 100644
--- a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
+++ b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
@@ -208,6 +208,11 @@ events-by-slice firehose query that fans out one shared
journal query to many co
* `pekko.persistence.typed.event-writer.max-batch-size` and
`pekko.persistence.typed.event-writer.ask-timeout` configure the event writer.
([PR3432](https://github.com/apache/pekko/pull/3432))
+* `pekko.persistence.typed.event-writer.fill-sequence-number-gaps` (default
`off`) makes the event
+writer fill gaps in sequence numbers with `FilteredPayload` events, and
+`pekko.persistence.typed.event-writer.latest-sequence-number-cache-capacity`
(default `1000`) sets how
+many persistence ids it keeps the latest sequence number in memory for when
gap filling is enabled.
+([PR3653](https://github.com/apache/pekko/pull/3653))
* `replay-batch-size` in the journal plugin fallback bounds the number of
replayed events queued in
a recovering persistent actor's mailbox.
([PR3325](https://github.com/apache/pekko/pull/3325))
* `only-one-snapshot` in the snapshot-store plugin fallback enables retention
optimizations for
diff --git
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterFillGapsSpec.scala
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterFillGapsSpec.scala
new file mode 100644
index 0000000000..0d78b87040
--- /dev/null
+++
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterFillGapsSpec.scala
@@ -0,0 +1,567 @@
+/*
+ * 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.typed.internal
+
+import scala.concurrent.ExecutionContext
+import scala.concurrent.Future
+import scala.concurrent.TimeoutException
+import scala.concurrent.duration.DurationInt
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.TestException
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.pattern.StatusReply
+import pekko.persistence.AtomicWrite
+import pekko.persistence.JournalProtocol
+
+import com.typesafe.config.ConfigFactory
+import org.scalatest.wordspec.AnyWordSpecLike
+
+object EventWriterFillGapsSpec {
+ def config =
+ ConfigFactory.parseString("""
+ pekko.persistence.journal.inmem.delay-writes=10ms
+ """).withFallback(ConfigFactory.load()).resolve()
+}
+
+class EventWriterFillGapsSpec
+ extends ScalaTestWithActorTestKit(EventWriterFillGapsSpec.config)
+ with AnyWordSpecLike
+ with LogCapturing {
+
+ val settings = EventWriter.EventWriterSettings(
+ 10,
+ 5.seconds,
+ fillSequenceNumberGaps = true,
+ latestSequenceNumberCacheCapacity = 100)
+ implicit val ec: ExecutionContext = testKit.system.executionContext
+
+ "The event writer" should {
+
+ "handle events without gaps when starting at 1" in new TestSetup {
+ sendWrite(1)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ sendWrite(2)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ sendWrite(3)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "handle events without gaps when starting at > 1" in new TestSetup {
+ sendWrite(3)
+ journalHighestSeqNr(2)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ sendWrite(4)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "handle duplicates" in new TestSetup {
+ sendWrite(1)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ // should also be ack:ed
+ sendWrite(1)
+ // duplicates detected before journal write
+ fakeJournal.expectNoMessage()
+ clientExpectSuccess(1)
+ }
+
+ "handle batched duplicates" in new TestSetup {
+ // first write, and wait for ack so that this test doesn't have to
lookup latest seqNr
+ sendWrite(1)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ sendWrite(2)
+
+ // first batch
+ for (n <- 3L to 10L) {
+ sendWrite(n)
+ }
+ // 2 will be written directly
+ journalAckWrite() should ===(1)
+ clientExpectSuccess(1)
+
+ // completing 1 triggers write of batch with 3-10
+ // second
+ for (n <- 1L to 10L) {
+ sendWrite(n)
+ }
+ // batch 3-10 in flight, writes in the meanwhile go in a new batch
+ journalAckWrite() should ===(8)
+ // duplicates detected before journal write
+
+ clientExpectSuccess(18)
+ }
+
+ "handle batches with half duplicates" in new TestSetup {
+ // first write, and wait for ack so that this test doesn't have to
lookup latest seqNr
+ sendWrite(1)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ sendWrite(1)
+ for (n <- 2L to 10L) {
+ sendWrite(n)
+ }
+ journalAckWrite() should ===(1)
+ journalAckWrite() should ===(8)
+ clientExpectSuccess(9)
+
+ for (n <- 5L until 15L) {
+ sendWrite(n)
+ }
+ // duplicates detected before journal write
+ journalAckWrite() should ===(1) // new write of 11 (non duplicates)
+ journalAckWrite() should ===(3) // new write of 12-15 (non duplicates)
+
+ // all writes succeeded
+ clientExpectSuccess(10)
+ }
+
+ "pass real errors from journal back" in new TestSetup {
+ sendWrite(1L)
+ journalFailWrite("error error")
+ // duplicate handling will ask for highest seq nr, can't know it is an
actual error
+ journalHighestSeqNr(0L)
+ val response = clientProbe.receiveMessage()
+ response.isError should ===(true)
+ response.getError.getMessage should ===("Journal write failed")
+ }
+
+ "pass real errors from journal back when highestSeqNr fails" in new
TestSetup {
+ sendWrite(1L)
+ journalFailWrite("the error")
+ // duplicate handling will ask for highest seq nr, can't know it is an
actual error
+ journalFailHighestSeqNr("highest error")
+ val response = clientProbe.receiveMessage()
+ response.isError should ===(true)
+ response.getError.getMessage should ===("Journal write failed")
+ }
+
+ "fill gaps when next expected is known" in new TestSetup {
+ sendWrite(1)
+ sendWrite(5)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ journalAckWrite(expectedSequenceNumbers = Vector(2, 3, 4, 5))
+ clientExpectSuccess(1)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps when next expected is unknown" in new TestSetup {
+ sendWrite(5)
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3, 4, 5))
+ clientExpectSuccess(1)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps when next expected is not trusted" in new TestSetup {
+ sendWrite(1)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ // event though we know about 1 some other node might have written 2, 3,
...
+ sendWrite(5)
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3, 4, 5))
+ clientExpectSuccess(1)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps when next expected is unknown and no highest" in new TestSetup {
+ sendWrite(5)
+ journalHighestSeqNr(0) // no events stored
+ journalAckWrite(expectedSequenceNumbers = Vector(1, 2, 3, 4, 5))
+ clientExpectSuccess(1)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps when next expected is unknown and more writes afterwards" in
new TestSetup {
+ sendWrite(4)
+ sendWrite(5)
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3, 4))
+ journalAckWrite(expectedSequenceNumbers = Vector(5))
+ clientExpectSuccess(2)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps when pending write" in new TestSetup {
+ sendWrite(1)
+
+ // no ack of 1 yet, but since it's pending we don't have to lookup max
+ sendWrite(5)
+ journalAckWrite() shouldBe 1
+ journalAckWrite() shouldBe 4
+ clientExpectSuccess(2)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps when pending batch" in new TestSetup {
+ sendWrite(1)
+ // 2 and 3 in next batch
+ sendWrite(2)
+ sendWrite(3)
+
+ sendWrite(5)
+ journalAckWrite() shouldBe 1
+ journalAckWrite() shouldBe 4 // all go into next batch, including the
filled 4
+ clientExpectSuccess(4)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps when more writes after max lookup" in new TestSetup {
+ sendWrite(5)
+ sendWrite(6)
+ sendWrite(7)
+ sendWrite(9) // 8 is another gap
+ journalHighestSeqNr(3) // 4 is a gap
+ journalAckWrite(expectedSequenceNumbers = Vector(4L, 5L))
+ // remaining into next batch
+ journalAckWrite() shouldBe 4
+ clientExpectSuccess(4)
+
+ sendWrite(10)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps when duplicate detected by journal" in new TestSetup {
+ sendWrite(1)
+ // 2 and 3 in next batch
+ sendWrite(2)
+ sendWrite(3)
+
+ sendWrite(5)
+ journalAckWrite(expectedSequenceNumbers = Vector(1))
+ // let's say 2 is a duplicate, detected by journal
+ journalFailWrite("duplicate") shouldBe 4
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3, 4, 5))
+ clientExpectSuccess(4)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "not count filled gaps in max batch check" in new TestSetup {
+ settings.maxBatchSize should ===(10)
+ sendWrite(1)
+
+ sendWrite(15)
+ sendWrite(16)
+ journalAckWrite(expectedSequenceNumbers = Vector(1))
+ journalAckWrite(expectedSequenceNumbers = (2L to 16L).toVector)
+ clientExpectSuccess(2)
+
+ sendWrite(17)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "evict least recently used entries" in new TestSetup {
+ // this test is based on capacity of 100
+ settings.latestSequenceNumberCacheCapacity should ===(100)
+ (1 to 13).foreach { n =>
+ sendWrite(1, pid = s"pid$n")
+ journalAckWrite(s"pid$n")
+ }
+ Thread.sleep(100) // makes the first 13 least recently used
+
+ (14 to 110).foreach { n =>
+ sendWrite(1, pid = s"pid$n")
+ journalAckWrite(s"pid$n")
+ }
+
+ // touch pid1
+ sendWrite(2, "pid1")
+ journalAckWrite("pid1")
+ // touch pid2
+ sendWrite(2, "pid2")
+ journalAckWrite("pid2")
+ // one more
+ sendWrite(1, "pid111")
+ journalAckWrite(pid = "pid111")
+
+ // now exceeded the cache threshold, and evicted entries will result in
lookup
+ sendWrite(2, pid = "pid3")
+ journalHighestSeqNr(1)
+ journalAckWrite("pid3")
+ sendWrite(2, pid = "pid12")
+ journalHighestSeqNr(1)
+ journalAckWrite("pid12")
+ sendWrite(2, pid = "pid13")
+ journalHighestSeqNr(1)
+ journalAckWrite("pid13")
+
+ // still in cache
+ sendWrite(2, "pid14")
+ journalAckWrite("pid14")
+ sendWrite(3, "pid1")
+ journalAckWrite("pid1")
+ sendWrite(3, "pid2")
+ journalAckWrite("pid2")
+ }
+
+ "evict entries where the lookup only found duplicates" in new TestSetup {
+ override def writerSettings =
settings.copy(latestSequenceNumberCacheCapacity = 2)
+ val pids = (1 to 6).map(n => s"pid$n")
+ pids.foreach { pid =>
+ sendWrite(5, pid)
+ journalHighestSeqNr(5)
+ clientExpectSuccess(1) // duplicate
+ }
+
+ // only the entries still in the cache are written directly, the evicted
need a lookup
+ pids.foreach(pid => sendWrite(6, pid))
+ val journalMessages = fakeJournal.receiveMessages(6)
+ journalMessages.count(_.isInstanceOf[JournalProtocol.WriteMessages])
should ===(2)
+ journalMessages.count(_.isInstanceOf[JournalProtocol.ReplayMessages])
should ===(4)
+ }
+
+ "apply the max batch size to writes with gaps" in new TestSetup {
+ override def writerSettings = settings.copy(maxBatchSize = 2)
+ sendWrite(1)
+ // 1 in flight, 3 and 5 are batched with the filled gaps 2 and 4
+ sendWrite(3)
+ sendWrite(5)
+ sendWrite(7)
+ clientExpectError(
+ 1,
+ "Max batch reached for pid pid1, at most 2 writes for the same pid may
be in flight at the same time")
+
+ journalAckWrite(expectedSequenceNumbers = Vector(1L))
+ journalAckWrite(expectedSequenceNumbers = Vector(2L, 3L, 4L, 5L))
+ clientExpectSuccess(3)
+ }
+
+ "handle writes to many pids" in {
+ // no flow control in this test so just no limit on batch size
+ val writer = spawn(EventWriter("pekko.persistence.journal.inmem",
settings.copy(maxBatchSize = Int.MaxValue)))
+ val probe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
+ (1 to 1000).map { pidN =>
+ Future {
+ for (n <- 1 to 20) {
+ writer ! EventWriter.Write(s"A|pid$pidN", n.toLong, n.toString,
false, None, Set.empty, probe.ref)
+ }
+ }
+ }
+ val replies = probe.receiveMessages(20 * 1000, 20.seconds)
+ replies.exists(_.isError) should ===(false)
+ }
+
+ "handle writes to many pids and fill gaps" in {
+ // no flow control in this test so just no limit on batch size
+ val writer = spawn(EventWriter("pekko.persistence.journal.inmem",
settings.copy(maxBatchSize = Int.MaxValue)))
+ val probe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
+ (1 to 1000).map { pidN =>
+ Future {
+ for (n <- 1 to 20) {
+ val gap =
+ if (pidN <= 500 && n <= 3) true
+ else if (pidN > 500 && 9 <= n && n <= 11) true
+ else false
+
+ if (!gap)
+ writer ! EventWriter.Write(s"B|pid$pidN", n.toLong, n.toString,
false, None, Set.empty, probe.ref)
+ }
+ }
+ }
+ // 20 - 3 because 3 gaps for each pid
+ val replies = probe.receiveMessages((20 - 3) * 1000, 20.seconds)
+ replies.exists(_.isError) should ===(false)
+ }
+ }
+
+ "The event writer with a failing highest sequence number lookup" should {
+
+ "fail all writes waiting for the initial lookup and accept a later write"
in new TestSetup {
+ sendWrite(3)
+ sendWrite(4)
+ journalFailHighestSeqNr("lookup failed")
+ clientExpectError(2, "Journal write
failed").foreach(_.getCause.getMessage should ===("lookup failed"))
+
+ sendWrite(5)
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3L, 4L, 5L))
+ clientExpectSuccess(1)
+ }
+
+ "fail the waiting write when the initial lookup times out and accept a
later write" in new TestSetup {
+ override def writerSettings = settings.copy(askTimeout = 300.millis)
+ sendWrite(3)
+ fakeJournal.expectMessageType[JournalProtocol.ReplayMessages] // no reply
+ clientExpectError(1, "Journal write failed").foreach(_.getCause shouldBe
a[TimeoutException])
+
+ sendWrite(4)
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3L, 4L))
+ clientExpectSuccess(1)
+ }
+ }
+
+ "The event writer with duplicates of pending writes" should {
+
+ "not acknowledge a duplicate before the pending write succeeds" in new
TestSetup {
+ val duplicateProbe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
+ sendWrite(1)
+ writer ! EventWriter.Write(pid1, 1L, "1", false, None, Set.empty,
duplicateProbe.ref)
+ duplicateProbe.expectNoMessage(100.millis)
+
+ journalAckWrite(expectedSequenceNumbers = Vector(1L))
+ clientExpectSuccess(1)
+ duplicateProbe.receiveMessage().isSuccess should ===(true)
+ }
+
+ "fail a duplicate when the pending write fails" in new TestSetup {
+ val duplicateProbe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
+ sendWrite(1)
+ writer ! EventWriter.Write(pid1, 1L, "1", false, None, Set.empty,
duplicateProbe.ref)
+ duplicateProbe.expectNoMessage(100.millis)
+
+ journalFailWrite("write failed")
+ // the write was not stored
+ journalHighestSeqNr(0)
+ clientExpectError(1, "Journal write failed")
+ duplicateProbe.receiveMessage().isError should ===(true)
+ }
+ }
+
+ "The event writer with rejected writes" should {
+
+ "fail the writes of a rejected gap fill batch and accept a later write" in
new TestSetup {
+ sendWrite(3)
+ journalHighestSeqNr(0)
+ journalRejectWrite("FilteredPayload not supported") should ===(3) // 1-2
gap fillers and 3
+ clientExpectError(1, "Journal rejected write").foreach(
+ _.getCause.getMessage should ===("FilteredPayload not supported"))
+
+ sendWrite(4)
+ journalHighestSeqNr(0)
+ journalAckWrite(expectedSequenceNumbers = Vector(1L, 2L, 3L, 4L))
+ clientExpectSuccess(1)
+ }
+ }
+
+ trait TestSetup {
+ def pid1 = "pid1"
+ def writerSettings: EventWriter.EventWriterSettings = settings
+ val fakeJournal = createTestProbe[JournalProtocol.Message]()
+ val writer = spawn(EventWriter(fakeJournal.ref, writerSettings))
+ val clientProbe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
+ def sendWrite(seqNr: Long, pid: String = pid1): Unit = {
+ writer ! EventWriter.Write(pid, seqNr, seqNr.toString, false, None,
Set.empty, clientProbe.ref)
+ }
+ def journalAckWrite(pid: String = pid1, expectedSequenceNumbers:
Vector[Long] = Vector.empty): Int = {
+ val write = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ write.messages should have size 1
+ val atomicWrite = write.messages.head.asInstanceOf[AtomicWrite]
+
+ val seqNrs =
+ atomicWrite.payload.map { repr =>
+ repr.persistenceId should ===(pid)
+ write.persistentActor ! JournalProtocol.WriteMessageSuccess(repr,
write.actorInstanceId)
+ repr.sequenceNr
+ }
+ if (expectedSequenceNumbers.nonEmpty)
+ seqNrs should ===(expectedSequenceNumbers)
+ write.persistentActor ! JournalProtocol.WriteMessagesSuccessful
+ atomicWrite.payload.size
+ }
+
+ def journalFailWrite(reason: String, pid: String = pid1): Int = {
+ val write = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ write.messages should have size 1
+ val atomicWrite = write.messages.head.asInstanceOf[AtomicWrite]
+ atomicWrite.payload.foreach { repr =>
+ repr.persistenceId should ===(pid)
+ write.persistentActor ! JournalProtocol.WriteMessageFailure(repr,
TestException(reason), write.actorInstanceId)
+ }
+ write.persistentActor ! JournalProtocol.WriteMessagesFailed(
+ TestException(reason),
+ atomicWrite.payload.size)
+ atomicWrite.payload.size
+ }
+
+ def journalRejectWrite(reason: String, pid: String = pid1): Int = {
+ val write = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ write.messages should have size 1
+ val atomicWrite = write.messages.head.asInstanceOf[AtomicWrite]
+ write.persistentActor ! JournalProtocol.WriteMessagesSuccessful
+ atomicWrite.payload.foreach { repr =>
+ repr.persistenceId should ===(pid)
+ write.persistentActor ! JournalProtocol.WriteMessageRejected(repr,
TestException(reason), write.actorInstanceId)
+ }
+ atomicWrite.payload.size
+ }
+
+ def clientExpectError(n: Int, message: String): Seq[Throwable] =
+ clientProbe.receiveMessages(n).map { reply =>
+ reply.isError should ===(true)
+ reply.getError.getMessage should ===(message)
+ reply.getError
+ }
+
+ def journalHighestSeqNr(highestSeqNr: Long): Unit = {
+ val replay =
fakeJournal.expectMessageType[JournalProtocol.ReplayMessages]
+ replay.persistentActor ! JournalProtocol.RecoverySuccess(highestSeqNr)
+ }
+
+ def journalFailHighestSeqNr(reason: String): Unit = {
+ val replay =
fakeJournal.expectMessageType[JournalProtocol.ReplayMessages]
+ replay.persistentActor !
JournalProtocol.ReplayMessagesFailure(TestException(reason))
+ }
+
+ def clientExpectSuccess(n: Int): Unit = {
+ clientProbe.receiveMessages(n).foreach { reply =>
+ reply.isSuccess should be(true)
+ }
+ }
+ }
+
+}
diff --git
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterSpec.scala
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterSpec.scala
index 918fb4b4b4..c4086ae022 100644
---
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterSpec.scala
+++
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/internal/EventWriterSpec.scala
@@ -18,6 +18,7 @@ import scala.concurrent.Future
import scala.concurrent.duration.DurationInt
import org.apache.pekko
+import pekko.actor.testkit.typed.TestException
import pekko.actor.testkit.typed.scaladsl.LogCapturing
import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
import pekko.pattern.StatusReply
@@ -42,7 +43,11 @@ class EventWriterSpec
with AnyWordSpecLike
with LogCapturing {
- private val settings = EventWriter.EventWriterSettings(10, 5.seconds)
+ private val settings = EventWriter.EventWriterSettings(
+ 10,
+ 5.seconds,
+ fillSequenceNumberGaps = false,
+ latestSequenceNumberCacheCapacity = 1000)
implicit val ec: ExecutionContext = testKit.system.executionContext
"The event writer" should {
@@ -65,11 +70,11 @@ class EventWriterSpec
for (n <- 2L to 10L) {
sendWrite(n)
}
- // 0 will be written directly
+ // 1 will be written directly
journalAckWrite() should ===(1)
clientExpectSuccess(1)
- // completing 1 triggers write of batch with 0-9
+ // completing 1 triggers write of batch with 2-10
// second
for (n <- 1L to 10L) {
sendWrite(n)
@@ -113,6 +118,40 @@ class EventWriterSpec
response.getError.getMessage should ===("Journal write failed")
}
+ "pass real errors from journal back when highestSeqNr fails" in new
TestSetup {
+ sendWrite(1L)
+ journalFailWrite("the error")
+ // duplicate handling will ask for highest seq nr, can't know it is an
actual error
+ journalFailHighestSeqNr("highest error")
+ val response = clientProbe.receiveMessage()
+ response.isError should ===(true)
+ response.getError.getMessage should ===("Journal write failed")
+ }
+
+ "bypass circuit breaker unless highest seqNr lookup failed" in new
TestSetup {
+ sendWrite(1L)
+ val write1 = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ write1.bypassCircuitBreaker should ===(true)
+ val repr1 = write1.messages.head.asInstanceOf[AtomicWrite].payload.head
+ write1.persistentActor !
+ JournalProtocol.WriteMessageFailure(repr1, TestException("the error"),
write1.actorInstanceId)
+ journalFailHighestSeqNr("highest error")
+ clientProbe.receiveMessage().isError should ===(true)
+
+ // journal seems to have real problems, use the circuit breaker
+ sendWrite(1L)
+ val write2 = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ write2.bypassCircuitBreaker should ===(false)
+ val repr2 = write2.messages.head.asInstanceOf[AtomicWrite].payload.head
+ write2.persistentActor ! JournalProtocol.WriteMessageSuccess(repr2,
write2.actorInstanceId)
+ clientExpectSuccess(1)
+
+ // successful write, bypass again
+ sendWrite(2L)
+ val write3 = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
+ write3.bypassCircuitBreaker should ===(true)
+ }
+
"ignores old failures when replay triggered" in new TestSetup {
sendWrite(1L) // triggers write
@@ -148,17 +187,207 @@ class EventWriterSpec
}
"handle writes to many pids" in {
- val writer = spawn(EventWriter("pekko.persistence.journal.inmem",
settings))
+ // no flow control in this test so just no limit on batch size
+ val writer = spawn(EventWriter("pekko.persistence.journal.inmem",
settings.copy(maxBatchSize = Int.MaxValue)))
val probe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
- (0 to 1000).map { pidN =>
+ (1 to 1000).map { pidN =>
Future {
- for (n <- 0 until 20) {
- writer ! EventWriter.Write(s"pid$pidN", n.toLong, n.toString,
None, Set.empty, probe.ref)
+ for (n <- 1 to 20) {
+ writer ! EventWriter.Write(s"A|pid$pidN", n.toLong, n.toString,
false, None, Set.empty, probe.ref)
}
}
}
- probe.receiveMessages(20 * 1000, 20.seconds)
+ val replies = probe.receiveMessages(20 * 1000, 20.seconds)
+ replies.exists(_.isError) should ===(false)
}
+
+ }
+
+ // These tests are for Write with isSnapshot = true
+ // They are similar to EventWriterFillGapsSpec
+ "Write of snapshot event" should {
+
+ "fill gaps before snapshot when next expected is known" in new TestSetup {
+ sendWrite(1)
+ sendSnapshotWrite(5)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ journalAckWrite(expectedSequenceNumbers = Vector(2, 3, 4, 5))
+ clientExpectSuccess(1)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps before snapshot when next expected is unknown" in new TestSetup
{
+ sendSnapshotWrite(5)
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3, 4, 5))
+ clientExpectSuccess(1)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps before snapshot when next expected is not trusted" in new
TestSetup {
+ sendWrite(1)
+ journalAckWrite()
+ clientExpectSuccess(1)
+
+ // event though we know about 1 some other node might have written 2, 3,
...
+ sendSnapshotWrite(5)
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3, 4, 5))
+ clientExpectSuccess(1)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps before snapshot when next expected is unknown and no highest"
in new TestSetup {
+ sendSnapshotWrite(5)
+ journalHighestSeqNr(0) // no events stored
+ journalAckWrite(expectedSequenceNumbers = Vector(1, 2, 3, 4, 5))
+ clientExpectSuccess(1)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps before snapshot when next expected is unknown and more writes
afterwards" in new TestSetup {
+ sendSnapshotWrite(4)
+ sendWrite(5)
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3, 4))
+ journalAckWrite(expectedSequenceNumbers = Vector(5))
+ clientExpectSuccess(2)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps before snapshot when pending write" in new TestSetup {
+ sendWrite(1)
+
+ // no ack of 1 yet, but since it's pending we don't have to lookup max
+ sendSnapshotWrite(5)
+ journalAckWrite() shouldBe 1
+ journalAckWrite() shouldBe 4
+ clientExpectSuccess(2)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps before snapshot when pending batch" in new TestSetup {
+ sendWrite(1)
+ // 2 and 3 in next batch
+ sendWrite(2)
+ sendWrite(3)
+
+ sendSnapshotWrite(5)
+ journalAckWrite() shouldBe 1
+ journalAckWrite() shouldBe 4 // all go into next batch, including the
filled 4
+ clientExpectSuccess(4)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps before snapshot when more writes after max lookup" in new
TestSetup {
+ sendSnapshotWrite(5)
+ sendWrite(6)
+ sendWrite(7)
+ sendSnapshotWrite(9) // 8 is another gap
+ journalHighestSeqNr(3) // 4 is a gap
+ journalAckWrite(expectedSequenceNumbers = Vector(4L, 5L))
+ // remaining into next batch
+ journalAckWrite() shouldBe 4
+ clientExpectSuccess(4)
+
+ sendWrite(10)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "fill gaps before snapshot when duplicate detected by journal" in new
TestSetup {
+ sendWrite(1)
+ // 2 and 3 in next batch
+ sendWrite(2)
+ sendWrite(3)
+
+ sendSnapshotWrite(5)
+ journalAckWrite(expectedSequenceNumbers = Vector(1))
+ // let's say 2 is a duplicate, detected by journal
+ journalFailWrite("duplicate") shouldBe 4
+ journalHighestSeqNr(2)
+ journalAckWrite(expectedSequenceNumbers = Vector(3, 4, 5))
+ clientExpectSuccess(4)
+
+ sendWrite(6)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "only fill gaps before the snapshot event, not before ordinary events
queued during its lookup" in new TestSetup {
+ sendSnapshotWrite(5)
+ // queued while the lookup for the snapshot event is in progress
+ sendWrite(7)
+ journalHighestSeqNr(3)
+ journalAckWrite(expectedSequenceNumbers = Vector(4L, 5L))
+ // gap filling is disabled, so 6 is not filled for the ordinary event
+ journalAckWrite(expectedSequenceNumbers = Vector(7L))
+ clientExpectSuccess(2)
+ }
+
+ "not count filled gaps before snapshot in max batch check" in new
TestSetup {
+ settings.maxBatchSize should ===(10)
+ sendWrite(1)
+
+ sendSnapshotWrite(15)
+ sendWrite(16)
+ journalAckWrite(expectedSequenceNumbers = Vector(1))
+ journalAckWrite(expectedSequenceNumbers = (2L to 16L).toVector)
+ clientExpectSuccess(2)
+
+ sendWrite(17)
+ journalAckWrite()
+ clientExpectSuccess(1)
+ }
+
+ "handle writes to many pids and fill gaps before snapshot" in {
+ // no flow control in this test so just no limit on batch size
+ val writer = spawn(EventWriter("pekko.persistence.journal.inmem",
settings.copy(maxBatchSize = Int.MaxValue)))
+ val probe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
+ (1 to 1000).map { pidN =>
+ Future {
+ for (n <- 1 to 20) {
+ val gap =
+ if (pidN <= 500 && n <= 3) true
+ else if (pidN > 500 && 9 <= n && n <= 11) true
+ else false
+ val snapshot =
+ if (pidN <= 500 && n == 4) true
+ else if (pidN > 500 && n == 12) true
+ else false
+
+ if (!gap)
+ writer ! EventWriter.Write(s"B|pid$pidN", n.toLong, n.toString,
snapshot, None, Set.empty, probe.ref)
+ }
+ }
+ }
+ // 20 - 3 because 3 gaps for each pid
+ val replies = probe.receiveMessages((20 - 3) * 1000, 20.seconds)
+ replies.exists(_.isError) should ===(false)
+ }
+
}
trait TestSetup {
@@ -168,17 +397,26 @@ class EventWriterSpec
val clientProbe = createTestProbe[StatusReply[EventWriter.WriteAck]]()
def sendWrite(seqNr: Long, pid: String = pid1): Unit = {
- writer ! EventWriter.Write(pid, seqNr, seqNr.toString, None, Set.empty,
clientProbe.ref)
+ writer ! EventWriter.Write(pid, seqNr, seqNr.toString, false, None,
Set.empty, clientProbe.ref)
+ }
+
+ def sendSnapshotWrite(seqNr: Long, pid: String = pid1): Unit = {
+ writer ! EventWriter.Write(pid, seqNr, seqNr.toString, isSnapshotEvent =
true, None, Set.empty, clientProbe.ref)
}
- def journalAckWrite(pid: String = pid1): Int = {
+ def journalAckWrite(pid: String = pid1, expectedSequenceNumbers:
Vector[Long] = Vector.empty): Int = {
val write = fakeJournal.expectMessageType[JournalProtocol.WriteMessages]
write.messages should have size 1
val atomicWrite = write.messages.head.asInstanceOf[AtomicWrite]
- atomicWrite.payload.foreach { repr =>
- repr.persistenceId should ===(pid)
- write.persistentActor ! JournalProtocol.WriteMessageSuccess(repr,
write.actorInstanceId)
- }
+
+ val seqNrs =
+ atomicWrite.payload.map { repr =>
+ repr.persistenceId should ===(pid)
+ write.persistentActor ! JournalProtocol.WriteMessageSuccess(repr,
write.actorInstanceId)
+ repr.sequenceNr
+ }
+ if (expectedSequenceNumbers.nonEmpty)
+ seqNrs should ===(expectedSequenceNumbers)
write.persistentActor ! JournalProtocol.WriteMessagesSuccessful
atomicWrite.payload.size
}
@@ -189,13 +427,10 @@ class EventWriterSpec
val atomicWrite = write.messages.head.asInstanceOf[AtomicWrite]
atomicWrite.payload.foreach { repr =>
repr.persistenceId should ===(pid)
- write.persistentActor ! JournalProtocol.WriteMessageFailure(
- repr,
- new RuntimeException(reason),
- write.actorInstanceId)
+ write.persistentActor ! JournalProtocol.WriteMessageFailure(repr,
TestException(reason), write.actorInstanceId)
}
write.persistentActor ! JournalProtocol.WriteMessagesFailed(
- new RuntimeException(reason),
+ TestException(reason),
atomicWrite.payload.size)
atomicWrite.payload.size
}
@@ -205,6 +440,11 @@ class EventWriterSpec
replay.persistentActor ! JournalProtocol.RecoverySuccess(highestSeqNr)
}
+ def journalFailHighestSeqNr(reason: String): Unit = {
+ val replay =
fakeJournal.expectMessageType[JournalProtocol.ReplayMessages]
+ replay.persistentActor !
JournalProtocol.ReplayMessagesFailure(TestException(reason))
+ }
+
def clientExpectSuccess(n: Int): Unit = {
clientProbe.receiveMessages(n).foreach { reply =>
reply.isSuccess should be(true)
diff --git
a/persistence-typed/src/main/mima-filters/2.0.x.backwards.excludes/eventwriter-fill-gaps.excludes
b/persistence-typed/src/main/mima-filters/2.0.x.backwards.excludes/eventwriter-fill-gaps.excludes
new file mode 100644
index 0000000000..b121f8884f
--- /dev/null
+++
b/persistence-typed/src/main/mima-filters/2.0.x.backwards.excludes/eventwriter-fill-gaps.excludes
@@ -0,0 +1,27 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# EventWriter is internal API, new fields for filling sequence number gaps
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.typed.internal.EventWriter#EventWriterSettings.copy")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.persistence.typed.internal.EventWriter#EventWriterSettings.unapply")
+ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.pekko.persistence.typed.internal.EventWriter#EventWriterSettings.unapply")
+# private classes and methods of EventWriter
+ProblemFilters.exclude[Problem]("org.apache.pekko.persistence.typed.internal.EventWriter#StateForPid*")
+ProblemFilters.exclude[Problem]("org.apache.pekko.persistence.typed.internal.EventWriter$StateForPid*")
+ProblemFilters.exclude[Problem]("org.apache.pekko.persistence.typed.internal.EventWriter#MaxSeqNrForPid*")
+ProblemFilters.exclude[Problem]("org.apache.pekko.persistence.typed.internal.EventWriter$MaxSeqNrForPid*")
+ProblemFilters.exclude[Problem]("org.apache.pekko.persistence.typed.internal.EventWriter.*emptyWaitingForWrite")
diff --git
a/persistence-typed/src/main/mima-filters/2.0.x.backwards.excludes/eventwriter-snapshot-event.excludes
b/persistence-typed/src/main/mima-filters/2.0.x.backwards.excludes/eventwriter-snapshot-event.excludes
new file mode 100644
index 0000000000..7505940a06
--- /dev/null
+++
b/persistence-typed/src/main/mima-filters/2.0.x.backwards.excludes/eventwriter-snapshot-event.excludes
@@ -0,0 +1,22 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# EventWriter is internal API, new isSnapshotEvent field in Write (old
constructor and apply are kept)
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.persistence.typed.internal.EventWriter#Write.copy")
+ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.pekko.persistence.typed.internal.EventWriter#Write.copy$default$*")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.persistence.typed.internal.EventWriter#Write.unapply")
+ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.pekko.persistence.typed.internal.EventWriter#Write._*")
diff --git a/persistence-typed/src/main/resources/reference.conf
b/persistence-typed/src/main/resources/reference.conf
index c810d71c01..7ee77a7c9f 100644
--- a/persistence-typed/src/main/resources/reference.conf
+++ b/persistence-typed/src/main/resources/reference.conf
@@ -61,6 +61,11 @@ pekko.persistence.typed {
# The event-writer occasionally needs to ask the journal about highest
sequence number to handle duplicate
# writes, this timeout is for that interaction
ask-timeout = 20s
+ # Detect gaps in sequence numbers and fill them with FilteredPayload
+ fill-sequence-number-gaps = off
+ # When fill-sequence-number-gaps is enabled it will keep latest sequence
+ # number in memory for this many persistence ids.
+ latest-sequence-number-cache-capacity = 1000
}
}
diff --git
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala
index 4bb5abd34d..fa3221f968 100644
---
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala
+++
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala
@@ -36,6 +36,7 @@ import pekko.actor.typed.scaladsl.adapter.TypedActorRefOps
import pekko.annotation.InternalStableApi
import pekko.pattern.StatusReply
import pekko.persistence.AtomicWrite
+import pekko.persistence.FilteredPayload
import pekko.persistence.JournalProtocol
import pekko.persistence.Persistence
import pekko.persistence.PersistentRepr
@@ -82,58 +83,203 @@ private[pekko] object EventWriter {
type Pid = String
object EventWriterSettings {
- def apply(system: ActorSystem[?]): EventWriterSettings =
+ def apply(system: ActorSystem[?]): EventWriterSettings = {
+ val config =
system.settings.config.getConfig("pekko.persistence.typed.event-writer")
EventWriterSettings(
- maxBatchSize =
system.settings.config.getInt("pekko.persistence.typed.event-writer.max-batch-size"),
- askTimeout =
system.settings.config.getDuration("pekko.persistence.typed.event-writer.ask-timeout").toScala)
+ maxBatchSize = config.getInt("max-batch-size"),
+ askTimeout = config.getDuration("ask-timeout").toScala,
+ fillSequenceNumberGaps =
config.getBoolean("fill-sequence-number-gaps"),
+ latestSequenceNumberCacheCapacity =
config.getInt("latest-sequence-number-cache-capacity"))
+ }
+
+ /**
+ * For binary compatibility, gap filling disabled.
+ */
+ def apply(maxBatchSize: Int, askTimeout: FiniteDuration):
EventWriterSettings =
+ new EventWriterSettings(maxBatchSize, askTimeout)
}
- final case class EventWriterSettings(maxBatchSize: Int, askTimeout:
FiniteDuration)
+ final case class EventWriterSettings(
+ maxBatchSize: Int,
+ askTimeout: FiniteDuration,
+ fillSequenceNumberGaps: Boolean,
+ latestSequenceNumberCacheCapacity: Int) {
+
+ /**
+ * For binary compatibility, gap filling disabled.
+ */
+ def this(maxBatchSize: Int, askTimeout: FiniteDuration) =
+ this(maxBatchSize, askTimeout, fillSequenceNumberGaps = false,
latestSequenceNumberCacheCapacity = 1000)
+ }
sealed trait Command
+
+ // extends AbstractFunction6 to keep the companion binary compatible with
the previous shape
+ object Write
+ extends scala.runtime.AbstractFunction6[
+ Pid,
+ SeqNr,
+ Any,
+ Option[Any],
+ Set[String],
+ ActorRef[StatusReply[WriteAck]],
+ Write] {
+
+ /**
+ * For binary compatibility, the event is not a snapshot event.
+ */
+ def apply(
+ persistenceId: Pid,
+ sequenceNumber: SeqNr,
+ event: Any,
+ metadata: Option[Any],
+ tags: Set[String],
+ replyTo: ActorRef[StatusReply[WriteAck]]): Write =
+ new Write(persistenceId, sequenceNumber, event, isSnapshotEvent = false,
metadata, tags, replyTo)
+ }
+
+ /**
+ * @param isSnapshotEvent when `true` gaps before the event will be filled
with `FilteredPayload` even
+ * if `fill-sequence-number-gaps` is disabled
+ */
final case class Write(
persistenceId: Pid,
sequenceNumber: SeqNr,
event: Any,
+ isSnapshotEvent: Boolean,
metadata: Option[Any],
tags: Set[String],
replyTo: ActorRef[StatusReply[WriteAck]])
- extends Command
+ extends Command {
+
+ /**
+ * For binary compatibility, the event is not a snapshot event.
+ */
+ def this(
+ persistenceId: Pid,
+ sequenceNumber: SeqNr,
+ event: Any,
+ metadata: Option[Any],
+ tags: Set[String],
+ replyTo: ActorRef[StatusReply[WriteAck]]) =
+ this(persistenceId, sequenceNumber, event, isSnapshotEvent = false,
metadata, tags, replyTo)
+ }
final case class WriteAck(persistenceId: Pid, sequenceNumber: SeqNr)
- private case class MaxSeqNrForPid(persistenceId: Pid, sequenceNumber: SeqNr,
originalErrorDesc: String)
+ private final case class MaxSeqNrForPid(persistenceId: Pid, sequenceNumber:
SeqNr, reason: AskMaxSeqNrReason)
extends Command
+ private final case class MaxSeqNrForPidFailed(persistenceId: Pid, reason:
AskMaxSeqNrReason, cause: Throwable)
+ extends Command
+
+ private sealed trait AskMaxSeqNrReason
+ private object AskMaxSeqNrReason {
+ final case class WriteFailure(originalErrorDesc: String) extends
AskMaxSeqNrReason
+ case object FillGaps extends AskMaxSeqNrReason
+ case object SnapshotEvent extends AskMaxSeqNrReason
+ }
- private def emptyWaitingForWrite = Vector.empty[(PersistentRepr,
ActorRef[StatusReply[WriteAck]])]
+ /**
+ * A write waiting for the highest sequence number lookup, with whether it
is a snapshot event.
+ */
+ private final case class PendingLookup(
+ repr: PersistentRepr,
+ replyTo: ActorRef[StatusReply[WriteAck]],
+ isSnapshotEvent: Boolean)
+
+ /**
+ * @param duplicateWaiters callers that sent a sequence number that is
already pending in
+ * `waitingForReply` or `waitingForWrite`, they get
the same outcome as that write
+ */
private case class StateForPid(
waitingForReply: Map[SeqNr, (PersistentRepr,
ActorRef[StatusReply[WriteAck]])],
- waitingForWrite: Vector[(PersistentRepr,
ActorRef[StatusReply[WriteAck]])] = emptyWaitingForWrite,
+ waitingForWrite: Vector[(PersistentRepr,
ActorRef[StatusReply[WriteAck]])] = Vector.empty,
writeErrorHandlingInProgress: Boolean = false,
- currentTransactionId: Int = 0)
+ currentTransactionId: Int = 0,
+ latestSeqNr: SeqNr = -1L,
+ usedTimestamp: Long = 0L,
+ waitingForSeqNrLookup: Vector[PendingLookup] = Vector.empty,
+ duplicateWaiters: Map[SeqNr, Vector[ActorRef[StatusReply[WriteAck]]]] =
Map.empty) {
+ def idle: Boolean =
+ waitingForReply.isEmpty && waitingForWrite.isEmpty
+
+ def seqNrlookupInProgress: Boolean =
+ waitingForSeqNrLookup.nonEmpty
+
+ def nextExpectedSeqNr: SeqNr =
+ if (waitingForWrite.nonEmpty)
+ waitingForWrite.last._1.sequenceNr + 1
+ else if (waitingForReply.nonEmpty)
+ waitingForReply.keysIterator.max + 1
+ else if (latestSeqNr == -1L)
+ 1L
+ else
+ latestSeqNr + 1
+
+ def isPending(seqNr: SeqNr): Boolean =
+ waitingForReply.contains(seqNr) || waitingForWrite.exists { case (repr,
_) => repr.sequenceNr == seqNr }
+
+ /**
+ * Replies to the waiting caller of `seqNr` and to its duplicate callers,
and forgets the duplicate callers.
+ */
+ def replyTo(seqNr: SeqNr, reply: StatusReply[WriteAck]): StateForPid = {
+ waitingForReply.get(seqNr).foreach { case (_, replyTo) => replyTo !
reply }
+ duplicateWaiters.get(seqNr) match {
+ case Some(duplicates) =>
+ duplicates.foreach(_ ! reply)
+ copy(duplicateWaiters = duplicateWaiters - seqNr)
+ case None => this
+ }
+ }
+
+ def waitingForWriteExceedingMaxBatchSize(batchSize: Int): Boolean =
+ waitingForWrite.size >= batchSize &&
+ waitingForWrite.count { case (repr, _) => repr.payload !=
FilteredPayload } >= batchSize
+ }
- def apply(journalPluginId: String, eventWriterSettings:
EventWriterSettings): Behavior[Command] =
+ def apply(journalPluginId: String, settings: EventWriterSettings):
Behavior[Command] =
Behaviors
.supervise(Behaviors.setup[Command] { context =>
val journal =
Persistence(context.system.classicSystem).journalFor(journalPluginId)
context.log.debug("Event writer for journal [{}] starting up",
journalPluginId)
- apply(journal.toTyped, eventWriterSettings)
+ apply(journal.toTyped, settings)
})
.onFailure[Exception](SupervisorStrategy.restart)
- def apply(journal: ActorRef[JournalProtocol.Message], eventWriterSettings:
EventWriterSettings): Behavior[Command] =
+ def apply(journal: ActorRef[JournalProtocol.Message], settings:
EventWriterSettings): Behavior[Command] =
Behaviors
.setup[AnyRef] { context =>
val writerUuid = UUID.randomUUID().toString
var perPidWriteState = Map.empty[Pid, StateForPid]
- implicit val askTimeout: Timeout = eventWriterSettings.askTimeout
-
- def sendToJournal(transactionId: Int, reprs: Vector[PersistentRepr]) =
{
+ // duplicate key violations are expected, and would otherwise trip the
journal circuit breaker
+ var bypassCircuitBreaker = true
+ implicit val askTimeout: Timeout = settings.askTimeout
+ import settings.fillSequenceNumberGaps
+
+ def journalWriteFailed(cause: Throwable): Throwable =
+ new IllegalStateException("Journal write failed", cause)
+
+ def journalWriteRejected(cause: Throwable): Throwable =
+ new IllegalStateException("Journal rejected write", cause)
+
+ def replyMaxBatchReached(persistenceId: Pid, replyTo:
ActorRef[StatusReply[WriteAck]]): Unit =
+ replyTo ! StatusReply.error(
+ s"Max batch reached for pid $persistenceId, at most
${settings.maxBatchSize} writes for " +
+ "the same pid may be in flight at the same time")
+
+ def sendToJournal(transactionId: Int, reprs: Vector[PersistentRepr]):
Unit = {
+ if (context.log.isTraceEnabled)
+ context.log.traceN(
+ "Writing events persistence id [{}], sequence nrs [{}-{}]",
+ reprs.head.persistenceId,
+ reprs.head.sequenceNr,
+ reprs.last.sequenceNr)
journal ! JournalProtocol.WriteMessages(
AtomicWrite(reprs) :: Nil,
context.self.toClassic,
// Note: we use actorInstanceId to correlate replies from one
write request
- actorInstanceId = transactionId)
+ actorInstanceId = transactionId,
+ bypassCircuitBreaker)
}
def handleUpdatedStateForPid(pid: Pid, newStateForPid: StateForPid):
Unit = {
@@ -141,7 +287,10 @@ private[pekko] object EventWriter {
// more waiting replyTo before we could batch it or scrap the entry
perPidWriteState = perPidWriteState.updated(pid, newStateForPid)
} else {
- if (newStateForPid.waitingForWrite.isEmpty) {
+ if (newStateForPid.waitingForWrite.isEmpty &&
fillSequenceNumberGaps) {
+ perPidWriteState = perPidWriteState.updated(pid, newStateForPid)
+ evictLeastRecentlyUsedPids()
+ } else if (newStateForPid.waitingForWrite.isEmpty) {
perPidWriteState = perPidWriteState - pid
} else {
// batch waiting for pid
@@ -151,7 +300,7 @@ private[pekko] object EventWriter {
val updatedState =
newStateForPid.copy(
newReplyTo,
- emptyWaitingForWrite,
+ waitingForWrite = Vector.empty,
currentTransactionId = newStateForPid.currentTransactionId +
1)
if (context.log.isTraceEnabled())
@@ -171,9 +320,59 @@ private[pekko] object EventWriter {
}
}
+ def askMaxSeqNr(pid: Pid, reason: AskMaxSeqNrReason): Unit = {
+ context.log.trace("Request highest sequence number for pid [{}]",
pid)
+ context.ask(
+ journal,
+ (replyTo: ActorRef[JournalProtocol.Response]) =>
+ JournalProtocol.ReplayMessages(0L, 0L, 1L, pid,
replyTo.toClassic)) {
+ case Success(JournalProtocol.RecoverySuccess(highestSequenceNr)) =>
+ MaxSeqNrForPid(pid, highestSequenceNr, reason)
+ case Success(JournalProtocol.ReplayMessagesFailure(cause)) =>
+ MaxSeqNrForPidFailed(pid, reason, cause)
+ case Failure(cause) =>
+ MaxSeqNrForPidFailed(pid, reason, cause)
+ case Success(unexpected) =>
+ throw new IllegalArgumentException(
+ s"Got unexpected reply from journal ${unexpected.getClass} for
pid $pid. " +
+ s"original reason: $reason")
+ }
+ }
+
+ def evictLeastRecentlyUsedPids(): Unit = {
+ import settings.latestSequenceNumberCacheCapacity
+ val accumulationFactor = 1.1
+ if (perPidWriteState.size >= latestSequenceNumberCacheCapacity *
accumulationFactor) {
+ val idleEntries =
+ perPidWriteState.iterator.filter {
+ case (_, stateForPid) => stateForPid.idle &&
stateForPid.waitingForSeqNrLookup.isEmpty
+ }.toVector
+
+ if (idleEntries.size >= latestSequenceNumberCacheCapacity *
accumulationFactor) {
+ val pidsToRemove =
+ idleEntries
+ .sortBy {
+ case (_, stateForPid) => stateForPid.usedTimestamp
+ }
+ .take(idleEntries.size - latestSequenceNumberCacheCapacity)
+ .map { case (pid, _) => pid }
+
+ if (context.log.isTraceEnabled)
+ context.log.traceN(
+ "Evicted cache from [{}] to [{}], persistence ids [{}]",
+ idleEntries.size,
+ idleEntries.size - pidsToRemove.size,
+ pidsToRemove.mkString(", "))
+
+ perPidWriteState --= pidsToRemove
+ }
+ }
+ }
+
def handleJournalResponse(response: JournalProtocol.Response):
Behavior[AnyRef] =
response match {
case JournalProtocol.WriteMessageSuccess(message, transactionId) =>
+ bypassCircuitBreaker = true
val pid = message.persistenceId
val sequenceNr = message.sequenceNr
perPidWriteState.get(pid) match {
@@ -188,17 +387,21 @@ private[pekko] object EventWriter {
stateForPid.waitingForReply.get(sequenceNr) match {
case None =>
context.log.debugN(
- "Got write error reply for event with no waiting
request for seq nr, ignoring (pid [{}], seq nr [{}], tx id [{}])",
+ "Got write reply for event with no waiting request
for seq nr, ignoring (pid [{}], seq nr [{}], tx id [{}])",
pid,
sequenceNr,
transactionId)
- case Some((_, waiting)) =>
+ case Some(_) =>
context.log.trace2(
"Successfully wrote event persistence id [{}],
sequence nr [{}]",
pid,
- message.sequenceNr)
- waiting ! StatusReply.success(WriteAck(pid,
sequenceNr))
- val newState = stateForPid.copy(waitingForReply =
stateForPid.waitingForReply - sequenceNr)
+ sequenceNr)
+ val newState = stateForPid
+ .replyTo(sequenceNr,
StatusReply.success(WriteAck(pid, sequenceNr)))
+ .copy(
+ waitingForReply = stateForPid.waitingForReply -
sequenceNr,
+ latestSeqNr = sequenceNr,
+ usedTimestamp = System.currentTimeMillis())
handleUpdatedStateForPid(pid, newState)
}
} else {
@@ -238,20 +441,7 @@ private[pekko] object EventWriter {
if (!state.writeErrorHandlingInProgress &&
state.currentTransactionId == transactionId) {
perPidWriteState =
perPidWriteState.updated(pid,
state.copy(writeErrorHandlingInProgress = true))
- context.ask(
- journal,
- (replyTo: ActorRef[JournalProtocol.Response]) =>
- JournalProtocol.ReplayMessages(0L, 0L, 1L, pid,
replyTo.toClassic)) {
- case
Success(JournalProtocol.RecoverySuccess(highestSequenceNr)) =>
- MaxSeqNrForPid(pid, highestSequenceNr,
error.getMessage)
- case Success(unexpected) =>
- throw new IllegalArgumentException(
- s"Got unexpected reply from journal
${unexpected.getClass} for pid $pid")
- case Failure(exception) =>
- throw new RuntimeException(
- s"Error finding highest sequence number in
journal for pid $pid",
- exception)
- }
+ askMaxSeqNr(pid,
AskMaxSeqNrReason.WriteFailure(error.getMessage))
} else {
context.log.traceN(
"Ignoring failure for pid [{}], seq nr [{}], tx id
[{}], since write error handling already in progress or old tx id (current tx
id [{}])",
@@ -264,13 +454,170 @@ private[pekko] object EventWriter {
}
Behaviors.same
+ case JournalProtocol.WriteMessageRejected(message, cause,
transactionId) =>
+ val pid = message.persistenceId
+ val sequenceNr = message.sequenceNr
+ perPidWriteState.get(pid) match {
+ case Some(state)
+ if transactionId == state.currentTransactionId &&
state.waitingForReply.contains(sequenceNr) =>
+ // the journal rejected the write before storing it, e.g. it
could not serialize the event
+ // or a FilteredPayload gap filler, so all events of the
atomic write are rejected
+ context.log.warnN(
+ "Journal rejected write of event persistence id [{}],
sequence nr [{}]: {}",
+ pid,
+ sequenceNr,
+ cause.getMessage)
+ val newState = state.replyTo(sequenceNr,
StatusReply.error(journalWriteRejected(cause)))
+ handleUpdatedStateForPid(pid, newState.copy(waitingForReply
= newState.waitingForReply - sequenceNr))
+ case _ =>
+ context.log.debugN(
+ "Got write rejection for event with no waiting request,
ignoring (pid [{}], seq nr [{}], tx id [{}])",
+ pid,
+ sequenceNr,
+ transactionId)
+ }
+ Behaviors.same
+
case _ =>
// ignore all other journal protocol messages
Behaviors.same
}
+ def handleWrite(
+ repr: PersistentRepr,
+ replyTo: ActorRef[StatusReply[WriteAck]],
+ isSnapshotEvent: Boolean,
+ calledAfterMaxSeqNr: Boolean): Unit = {
+ val persistenceId = repr.persistenceId
+ val sequenceNumber = repr.sequenceNr
+ val newStateForPid =
+ perPidWriteState.get(persistenceId) match {
+ case None =>
+ if ((fillSequenceNumberGaps || isSnapshotEvent) &&
sequenceNumber != 1L) {
+ val reason = if (isSnapshotEvent)
AskMaxSeqNrReason.SnapshotEvent else AskMaxSeqNrReason.FillGaps
+ askMaxSeqNr(persistenceId, reason)
+ StateForPid(
+ waitingForReply = Map.empty,
+ waitingForSeqNrLookup = Vector(PendingLookup(repr,
replyTo, isSnapshotEvent)))
+ } else {
+ sendToJournal(1, Vector(repr))
+ StateForPid(Map((repr.sequenceNr, (repr, replyTo))),
currentTransactionId = 1)
+ }
+
+ case Some(state) =>
+ val expectedSeqNr =
+ if (fillSequenceNumberGaps || isSnapshotEvent)
state.nextExpectedSeqNr
+ else repr.sequenceNr
+
+ if (state.seqNrlookupInProgress) {
+ context.log.trace2(
+ "Seq nr lookup in progress for persistence id [{}], adding
sequence nr [{}] to pending",
+ persistenceId,
+ sequenceNumber)
+ state.copy(waitingForSeqNrLookup =
+ state.waitingForSeqNrLookup :+ PendingLookup(repr,
replyTo, isSnapshotEvent))
+ } else if (sequenceNumber == expectedSeqNr) {
+ if (state.idle) {
+ sendToJournal(state.currentTransactionId + 1, Vector(repr))
+ state.copy(
+ waitingForReply = Map((repr.sequenceNr, (repr,
replyTo))),
+ currentTransactionId = state.currentTransactionId + 1)
+ } else {
+ // write in progress for pid, add write to batch and
perform once current write completes
+ if
(state.waitingForWriteExceedingMaxBatchSize(settings.maxBatchSize)) {
+ replyMaxBatchReached(persistenceId, replyTo)
+ state
+ } else {
+ context.log.trace2(
+ "Writing event in progress for persistence id [{}],
adding sequence nr [{}] to batch",
+ persistenceId,
+ sequenceNumber)
+ state.copy(waitingForWrite = state.waitingForWrite :+
((repr, replyTo)))
+ }
+ }
+ } else if (sequenceNumber < expectedSeqNr) {
+ if (state.isPending(sequenceNumber)) {
+ // not durable yet, reply when the outcome of the pending
write is known
+ context.log.trace2(
+ "Duplicate of pending seq nr [{}] for persistence id
[{}]",
+ sequenceNumber,
+ persistenceId)
+ state.copy(duplicateWaiters =
state.duplicateWaiters.updated(
+ sequenceNumber,
+ state.duplicateWaiters.getOrElse(sequenceNumber,
Vector.empty) :+ replyTo))
+ } else {
+ context.log.trace2("Duplicate seq nr [{}] for persistence
id [{}]", sequenceNumber, persistenceId)
+ replyTo ! StatusReply.success(WriteAck(persistenceId,
sequenceNumber)) // duplicate
+ state
+ }
+ } else { // sequenceNumber > expectedSeqNr
+ require(
+ fillSequenceNumberGaps || isSnapshotEvent,
+ s"Unexpected sequence number gap, expected
[$expectedSeqNr], received [$sequenceNumber]. " +
+ "Enable
pekko.persistence.typed.event-writer.fill-sequence-number-gaps config if gaps
are " +
+ "expected and should be filled with FilteredPayload.")
+
+ if (!state.idle &&
state.waitingForWriteExceedingMaxBatchSize(settings.maxBatchSize)) {
+ // check before filling, the filled gaps don't count
towards the max batch size
+ replyMaxBatchReached(persistenceId, replyTo)
+ state
+ } else if (calledAfterMaxSeqNr || !state.idle) {
+ // here we can trust that the expectedSeqNr is correct and
don't have to lookup max
+ val fillRepr = (expectedSeqNr until sequenceNumber).map {
n =>
+ PersistentRepr(
+ FilteredPayload,
+ persistenceId = persistenceId,
+ sequenceNr = n,
+ manifest = "", // adapters would be on the producing
side, already applied
+ writerUuid = writerUuid,
+ sender = pekko.actor.ActorRef.noSender)
+ }.toVector
+ context.log.debugN(
+ "Detected sequence nr gap{} [{}-{}] for persistence id
[{}]. Filling with FilteredPayload.",
+ if (isSnapshotEvent) " for snapshot event" else "",
+ fillRepr.head.sequenceNr,
+ fillRepr.last.sequenceNr,
+ persistenceId)
+ val ignoreRef = context.system.ignoreRef
+ if (state.idle) {
+ sendToJournal(state.currentTransactionId + 1, fillRepr
:+ repr)
+ val newWaitingForReply =
+ (fillRepr.map(r => r.sequenceNr -> (r -> ignoreRef)) :+
+ (repr.sequenceNr -> (repr -> replyTo))).toMap
+ state.copy(
+ waitingForReply = newWaitingForReply,
+ currentTransactionId = state.currentTransactionId + 1)
+ } else {
+ if (context.log.isTraceEnabled)
+ context.log.traceN(
+ "Writing event in progress for persistence id [{}],
adding sequence nrs [{}-{}] to batch",
+ persistenceId,
+ fillRepr.head.sequenceNr,
+ sequenceNumber)
+ val newWaitingForWrite = fillRepr.map(_ -> ignoreRef) :+
(repr -> replyTo)
+ state.copy(waitingForWrite = state.waitingForWrite ++
newWaitingForWrite)
+ }
+ } else {
+ // No pending writes or we haven't just looked up latest
sequence nr.
+ // The reason we don't trust a previous state.latestSeqNr
is because another EventWriter
+ // on another node might have been taking writes since we
retrieved it.
+ val reason = if (isSnapshotEvent)
AskMaxSeqNrReason.SnapshotEvent else AskMaxSeqNrReason.FillGaps
+ askMaxSeqNr(persistenceId, reason)
+ context.log.trace2(
+ "Seq nr lookup needed for persistence id [{}], adding
sequence nr [{}] to pending",
+ persistenceId,
+ sequenceNumber)
+ state.copy(waitingForSeqNrLookup =
+ state.waitingForSeqNrLookup :+ PendingLookup(repr,
replyTo, isSnapshotEvent))
+ }
+
+ }
+ }
+ perPidWriteState = perPidWriteState.updated(persistenceId,
newStateForPid)
+ }
+
Behaviors.receiveMessage {
- case Write(persistenceId, sequenceNumber, event, metadata, tags,
replyTo) =>
+ case Write(persistenceId, sequenceNumber, event, isSnapshotEvent,
metadata, tags, replyTo) =>
val payload = if (tags.isEmpty) event else Tagged(event, tags)
val repr = PersistentRepr(
payload,
@@ -285,72 +632,41 @@ private[pekko] object EventWriter {
case _ => repr
}
- val newStateForPid =
- perPidWriteState.get(persistenceId) match {
- case None =>
- if (context.log.isTraceEnabled)
- context.log.traceN(
- "Writing event persistence id [{}], sequence nr [{}],
payload {}",
- persistenceId,
- sequenceNumber,
- event)
- sendToJournal(1, Vector(reprWithMeta))
- StateForPid(
- Map((reprWithMeta.sequenceNr, (reprWithMeta, replyTo))),
- emptyWaitingForWrite,
- currentTransactionId = 1)
- case Some(state) =>
- // write in progress for pid, add write to batch and perform
once current write completes
- if (state.waitingForWrite.size ==
eventWriterSettings.maxBatchSize) {
- replyTo ! StatusReply.error(
- s"Max batch reached for pid $persistenceId, at most
${eventWriterSettings.maxBatchSize} writes for " +
- "the same pid may be in flight at the same time")
- state
- } else {
- if (context.log.isTraceEnabled)
- context.log.traceN(
- "Writing event in progress for persistence id [{}],
adding sequence nr [{}], payload {} to batch",
- persistenceId,
- sequenceNumber,
- event)
- state.copy(waitingForWrite = state.waitingForWrite :+
((reprWithMeta, replyTo)))
- }
- }
- perPidWriteState = perPidWriteState.updated(persistenceId,
newStateForPid)
+ handleWrite(reprWithMeta, replyTo, isSnapshotEvent,
calledAfterMaxSeqNr = false)
Behaviors.same
- case MaxSeqNrForPid(pid, maxSeqNr, originalErrorDesc) =>
+ case MaxSeqNrForPid(pid, maxSeqNr,
AskMaxSeqNrReason.WriteFailure(errorDesc)) =>
+ bypassCircuitBreaker = true
// write failed, so we looked up the maxSeqNr to detect if it was
duplicate events, already in journal
perPidWriteState.get(pid) match {
case None =>
context.log.debug2(
"Got max seq nr with no waiting previous state for pid,
ignoring (pid [{}], original error desc: {})",
pid,
- originalErrorDesc)
+ errorDesc)
case Some(state) =>
val sortedSeqs = state.waitingForReply.keys.toSeq.sorted
val (alreadyInJournal, needsWrite) =
sortedSeqs.partition(seqNr => seqNr <= maxSeqNr)
if (alreadyInJournal.isEmpty) {
// error was not about duplicates
- state.waitingForReply.values.foreach {
- case (_, replyTo) =>
- replyTo ! StatusReply.error("Journal write failed")
+ val stateAfterReplies = sortedSeqs.foldLeft(state) { (state,
seqNr) =>
+ state.replyTo(seqNr, StatusReply.error("Journal write
failed"))
}
context.log.warnN(
"Failed writing event batch persistence id [{}], sequence
nr [{}-{}]: {}",
pid,
sortedSeqs.head,
sortedSeqs.last,
- originalErrorDesc)
- val newState = state.copy(waitingForReply = Map.empty,
writeErrorHandlingInProgress = false)
+ errorDesc)
+ val newState =
+ stateAfterReplies.copy(waitingForReply = Map.empty,
writeErrorHandlingInProgress = false)
handleUpdatedStateForPid(pid, newState)
} else {
// ack all already written
val stateAfterWritten = alreadyInJournal
.foldLeft(state) { (state, seqNr) =>
- val (_, replyTo) = state.waitingForReply(seqNr)
- replyTo ! StatusReply.success(WriteAck(pid, seqNr))
- state.copy(waitingForReply = state.waitingForReply -
seqNr)
+ val newState = state.replyTo(seqNr,
StatusReply.success(WriteAck(pid, seqNr)))
+ newState.copy(waitingForReply = newState.waitingForReply
- seqNr)
}
.copy(writeErrorHandlingInProgress = false)
if (needsWrite.isEmpty) {
@@ -378,6 +694,71 @@ private[pekko] object EventWriter {
}
Behaviors.same
+ case MaxSeqNrForPid(pid, maxSeqNr, AskMaxSeqNrReason.FillGaps |
AskMaxSeqNrReason.SnapshotEvent) =>
+ bypassCircuitBreaker = true
+ perPidWriteState.get(pid) match {
+ case None =>
+ context.log.debug("Got max seq nr with no waiting previous
state for pid, ignoring (pid [{}])", pid)
+ case Some(state) =>
+ context.log.debug("Retrieved max seq nr [{}] for pid [{}]",
maxSeqNr, pid)
+ val waiting = state.waitingForSeqNrLookup
+ perPidWriteState = perPidWriteState.updated(
+ pid,
+ state.copy(
+ latestSeqNr = maxSeqNr,
+ usedTimestamp = System.currentTimeMillis(),
+ waitingForSeqNrLookup = Vector.empty))
+ // each queued write keeps its own snapshot event flag, the
lookup reason only
+ // belongs to the write that triggered it
+ waiting.foreach { pending =>
+ handleWrite(pending.repr, pending.replyTo,
pending.isSnapshotEvent, calledAfterMaxSeqNr = true)
+ }
+ // all queued writes may have been duplicates, then nothing is
in flight and the
+ // idle state must go through the cache eviction like after a
completed write
+ perPidWriteState.get(pid).foreach { stateAfterWrites =>
+ if (stateAfterWrites.idle &&
!stateAfterWrites.seqNrlookupInProgress)
+ handleUpdatedStateForPid(pid, stateAfterWrites)
+ }
+ }
+ Behaviors.same
+
+ case MaxSeqNrForPidFailed(pid, reason, cause) =>
+ // journal seems to have real problems, let the circuit breaker
protect it until next successful write
+ bypassCircuitBreaker = false
+ def errorDescInLog = reason match {
+ case AskMaxSeqNrReason.WriteFailure(errorDesc) => s", original
error desc: $errorDesc"
+ case _ => ""
+ }
+ perPidWriteState.get(pid) match {
+ case None =>
+ context.log.debugN(
+ "Got max seq nr (failed) with no waiting previous state for
persistence id [{}]{}: {}",
+ pid,
+ errorDescInLog,
+ cause.getMessage)
+ case Some(state) =>
+ // fail the writes waiting for a journal reply and those
waiting for this lookup
+ val failedSeqNrs =
+ (state.waitingForReply.keys ++
state.waitingForSeqNrLookup.map(_.repr.sequenceNr)).toVector.sorted
+ val failure =
StatusReply.error[WriteAck](journalWriteFailed(cause))
+ val stateAfterReplies =
+ state.waitingForReply.keys.foldLeft(state)((state, seqNr) =>
state.replyTo(seqNr, failure))
+ state.waitingForSeqNrLookup.foreach(_.replyTo ! failure)
+ if (failedSeqNrs.nonEmpty)
+ context.log.warnN(
+ "Failed max seq nr, and therefore failed writing events
persistence id [{}], sequence nrs [{}]{}: {}",
+ pid,
+ failedSeqNrs.mkString(", "),
+ errorDescInLog,
+ cause.getMessage)
+ val newState = stateAfterReplies.copy(
+ waitingForReply = Map.empty,
+ waitingForSeqNrLookup = Vector.empty,
+ writeErrorHandlingInProgress = false)
+ handleUpdatedStateForPid(pid, newState)
+ }
+ Behaviors.same
+
case response: JournalProtocol.Response =>
handleJournalResponse(response)
case unexpected =>
diff --git
a/persistence/src/main/scala-2/org/apache/pekko/persistence/WriteMessagesCompanion.scala
b/persistence/src/main/scala-2/org/apache/pekko/persistence/WriteMessagesCompanion.scala
new file mode 100644
index 0000000000..269db25f8f
--- /dev/null
+++
b/persistence/src/main/scala-2/org/apache/pekko/persistence/WriteMessagesCompanion.scala
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence
+
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+
+/**
+ * INTERNAL API
+ *
+ * Scala 2 specific parent of the [[JournalProtocol.WriteMessages]] companion
object, keeping it binary
+ * compatible with the companion of the former 3 field case class (`Function3`
parent and an `unapply`
+ * returning an `Option` of a 3-tuple).
+ */
+@InternalApi
+private[persistence] abstract class WriteMessagesCompanion
+ extends scala.runtime.AbstractFunction3[Seq[PersistentEnvelope], ActorRef,
Int, JournalProtocol.WriteMessages]
+ with Serializable {
+
+ def unapply(w: JournalProtocol.WriteMessages):
Option[(Seq[PersistentEnvelope], ActorRef, Int)] =
+ Some((w.messages, w.persistentActor, w.actorInstanceId))
+
+ override def toString: String = "WriteMessages"
+}
diff --git
a/persistence/src/main/scala-3/org/apache/pekko/persistence/WriteMessagesCompanion.scala
b/persistence/src/main/scala-3/org/apache/pekko/persistence/WriteMessagesCompanion.scala
new file mode 100644
index 0000000000..1d67a382fb
--- /dev/null
+++
b/persistence/src/main/scala-3/org/apache/pekko/persistence/WriteMessagesCompanion.scala
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence
+
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.annotation.InternalApi
+
+/**
+ * INTERNAL API
+ *
+ * Scala 3 specific parent of the [[JournalProtocol.WriteMessages]] companion
object, keeping it binary
+ * compatible with the companion of the former 3 field case class
(`Mirror.Product` parent and an
+ * `unapply` returning the instance itself, with `_1`, `_2` and `_3` as the
extracted values).
+ */
+@InternalApi
+private[persistence] abstract class WriteMessagesCompanion extends
scala.deriving.Mirror.Product with Serializable {
+ type MirroredMonoType = JournalProtocol.WriteMessages
+
+ def unapply(w: JournalProtocol.WriteMessages): JournalProtocol.WriteMessages
= w
+
+ def fromProduct(p: scala.Product): JournalProtocol.WriteMessages =
+ new JournalProtocol.WriteMessages(
+ p.productElement(0).asInstanceOf[Seq[PersistentEnvelope]],
+ p.productElement(1).asInstanceOf[ActorRef],
+ p.productElement(2).asInstanceOf[Int])
+
+ override def toString: String = "WriteMessages"
+}
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/JournalProtocol.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/JournalProtocol.scala
index e61e6a1da7..76ad89bf07 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/JournalProtocol.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/JournalProtocol.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.persistence
@@ -39,18 +39,99 @@ private[persistence] object JournalProtocol {
final case class DeleteMessagesTo(persistenceId: String, toSequenceNr: Long,
persistentActor: ActorRef)
extends Request
+ object WriteMessages extends WriteMessagesCompanion {
+ def apply(messages: Seq[PersistentEnvelope], persistentActor: ActorRef,
actorInstanceId: Int): WriteMessages =
+ new WriteMessages(messages, persistentActor, actorInstanceId,
bypassCircuitBreaker = false)
+
+ def apply(
+ messages: Seq[PersistentEnvelope],
+ persistentActor: ActorRef,
+ actorInstanceId: Int,
+ bypassCircuitBreaker: Boolean): WriteMessages =
+ new WriteMessages(messages, persistentActor, actorInstanceId,
bypassCircuitBreaker)
+ }
+
/**
* Request to write messages.
*
+ * This used to be a case class with the first 3 fields. It is a regular
class (with the same
+ * `apply`, `unapply`, `copy` and product methods for the first 3 fields) to
stay binary compatible
+ * with journal plugins compiled against earlier versions. Pattern matching
extracts the first 3
+ * fields.
+ *
* @param messages messages to be written.
* @param persistentActor write requester.
+ * @param bypassCircuitBreaker when `true` the write is not protected by the
journal circuit breaker,
+ * used when write failures are expected, e.g.
writing duplicates.
*/
- final case class WriteMessages(
- messages: Seq[PersistentEnvelope],
- persistentActor: ActorRef,
- actorInstanceId: Int)
+ final class WriteMessages(
+ val messages: Seq[PersistentEnvelope],
+ val persistentActor: ActorRef,
+ val actorInstanceId: Int,
+ val bypassCircuitBreaker: Boolean)
extends Request
with NoSerializationVerificationNeeded
+ with Product
+ with Serializable {
+
+ def this(messages: Seq[PersistentEnvelope], persistentActor: ActorRef,
actorInstanceId: Int) =
+ this(messages, persistentActor, actorInstanceId, bypassCircuitBreaker =
false)
+
+ /** Keeps the value of `bypassCircuitBreaker` */
+ def copy(
+ messages: Seq[PersistentEnvelope] = messages,
+ persistentActor: ActorRef = persistentActor,
+ actorInstanceId: Int = actorInstanceId): WriteMessages =
+ new WriteMessages(messages, persistentActor, actorInstanceId,
bypassCircuitBreaker)
+
+ def withBypassCircuitBreaker(value: Boolean): WriteMessages =
+ new WriteMessages(messages, persistentActor, actorInstanceId, value)
+
+ def _1: Seq[PersistentEnvelope] = messages
+ def _2: ActorRef = persistentActor
+ def _3: Int = actorInstanceId
+
+ override def productArity: Int = 3
+
+ override def productElement(n: Int): Any = n match {
+ case 0 => messages
+ case 1 => persistentActor
+ case 2 => actorInstanceId
+ case _ => throw new IndexOutOfBoundsException(n.toString)
+ }
+
+ override def productElementName(n: Int): String = n match {
+ case 0 => "messages"
+ case 1 => "persistentActor"
+ case 2 => "actorInstanceId"
+ case _ => throw new IndexOutOfBoundsException(n.toString)
+ }
+
+ override def productIterator: Iterator[Any] = Iterator(messages,
persistentActor, actorInstanceId)
+
+ override def productPrefix: String = "WriteMessages"
+
+ override def canEqual(that: Any): Boolean =
that.isInstanceOf[WriteMessages]
+
+ override def equals(other: Any): Boolean = other match {
+ case that: WriteMessages =>
+ messages == that.messages && persistentActor == that.persistentActor &&
+ actorInstanceId == that.actorInstanceId && bypassCircuitBreaker ==
that.bypassCircuitBreaker
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ // persistentActor may be null (ActorRef.noSender)
+ var result = java.util.Objects.hashCode(messages)
+ result = 31 * result + java.util.Objects.hashCode(persistentActor)
+ result = 31 * result + actorInstanceId
+ result = 31 * result + java.lang.Boolean.hashCode(bypassCircuitBreaker)
+ result
+ }
+
+ override def toString: String =
+
s"WriteMessages($messages,$persistentActor,$actorInstanceId,$bypassCircuitBreaker)"
+ }
/**
* Reply message to a successful [[WriteMessages]] request. This reply is
sent to the requester
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/Persistent.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/Persistent.scala
index 61b86fc77b..c6ff3923c7 100644
--- a/persistence/src/main/scala/org/apache/pekko/persistence/Persistent.scala
+++ b/persistence/src/main/scala/org/apache/pekko/persistence/Persistent.scala
@@ -231,7 +231,8 @@ private[persistence] final case class PersistentImpl(
result = HashCode.hash(result, persistenceId)
result = HashCode.hash(result, manifest)
result = HashCode.hash(result, deleted)
- result = HashCode.hash(result, sender)
+ // sender is null for ActorRef.noSender, which HashCode.hash rejects; same
hash as before for other senders
+ result = HashCode.hash(result, if (sender eq null) 0 else sender.hashCode)
result = HashCode.hash(result, writerUuid)
// timestamp not included in equals for backwards compatibility
// meta not included in equals for backwards compatibility
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala
index 874422b74b..f2628927d4 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.persistence.journal
@@ -18,6 +18,8 @@ import java.util.concurrent.atomic.AtomicLong
import scala.collection.mutable
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
+import scala.concurrent.Promise
+import scala.concurrent.TimeoutException
import scala.concurrent.duration._
import scala.util.{ Failure, Success, Try }
import scala.util.control.NonFatal
@@ -77,6 +79,25 @@ trait AsyncWriteJournal extends Actor with WriteJournalBase
with AsyncRecovery {
val eventStream = context.system.eventStream // used from Future callbacks
implicit val ec: ExecutionContext = context.dispatcher
+ // writes that bypass the circuit breaker still get its call timeout, so
that a write that never completes
+ // can't hold back the (resequenced) responses of later writes
+ val writeCallTimeout: FiniteDuration =
config.getDuration("circuit-breaker.call-timeout", MILLISECONDS).millis
+
+ def withWriteCallTimeout[T](write: Future[T]): Future[T] =
+ if (write.isCompleted) write
+ else {
+ val promise = Promise[T]()
+ val timeout = context.system.scheduler.scheduleOnce(writeCallTimeout) {
+ promise.tryFailure(
+ new TimeoutException(s"Journal write bypassing the circuit breaker
timed out after [$writeCallTimeout]"))
+ }
+ write.onComplete { result =>
+ timeout.cancel()
+ promise.tryComplete(result)
+ }
+ promise.future
+ }
+
val batchingReplayRequesters = mutable.Map.empty[ActorRef, Int]
val replaySessions = mutable.Map.empty[Long, ReplaySession]
@@ -215,7 +236,10 @@ trait AsyncWriteJournal extends Actor with
WriteJournalBase with AsyncRecovery {
}
{
- case WriteMessages(messages, persistentActor, actorInstanceId) =>
+ case w: WriteMessages =>
+ val messages = w.messages
+ val persistentActor = w.persistentActor
+ val actorInstanceId = w.actorInstanceId
val cctr = resequencerCounter
resequencerCounter += messages.foldLeft(1)((acc, m) => acc + m.size)
@@ -229,8 +253,10 @@ trait AsyncWriteJournal extends Actor with
WriteJournalBase with AsyncRecovery {
Future.successful(Nil)
case Success(prep) =>
// try in case the asyncWriteMessages throws
- try breaker.withCircuitBreaker(asyncWriteMessages(prep))
- catch { case NonFatal(e) => Future.failed(e) }
+ try {
+ if (w.bypassCircuitBreaker)
withWriteCallTimeout(asyncWriteMessages(prep))
+ else breaker.withCircuitBreaker(asyncWriteMessages(prep))
+ } catch { case NonFatal(e) => Future.failed(e) }
case f @ Failure(_) =>
// exception from preparePersistentBatch => rejected
Future.successful(messages.collect { case _: AtomicWrite => f })
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
index f9228dc55c..10c6d4d279 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2015-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2015-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.persistence.journal
@@ -199,16 +199,17 @@ final class PersistencePluginProxy(config: Config)
extends Actor with Stash with
case req: JournalProtocol.Request =>
req match { // exhaustive match
- case WriteMessages(messages, persistentActor, actorInstanceId) =>
- val atomicWriteCount = messages.count(_.isInstanceOf[AtomicWrite])
- persistentActor ! WriteMessagesFailed(timeoutException(),
atomicWriteCount)
- messages.foreach {
+ case w: WriteMessages =>
+ // type pattern rather than extractor, WriteMessages is not a case
class
+ val atomicWriteCount = w.messages.count(_.isInstanceOf[AtomicWrite])
+ w.persistentActor ! WriteMessagesFailed(timeoutException(),
atomicWriteCount)
+ w.messages.foreach {
case a: AtomicWrite =>
a.payload.foreach { p =>
- persistentActor ! WriteMessageFailure(p, timeoutException(),
actorInstanceId)
+ w.persistentActor ! WriteMessageFailure(p, timeoutException(),
w.actorInstanceId)
}
case r: NonPersistentRepr =>
- persistentActor ! LoopMessageSuccess(r.payload, actorInstanceId)
+ w.persistentActor ! LoopMessageSuccess(r.payload,
w.actorInstanceId)
}
case ReplayMessages(_, _, _, _, persistentActor) =>
persistentActor ! ReplayMessagesFailure(timeoutException())
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/WriteMessagesSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/WriteMessagesSpec.scala
new file mode 100644
index 0000000000..9e6e442c1d
--- /dev/null
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/WriteMessagesSpec.scala
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence
+
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.persistence.JournalProtocol.WriteMessages
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class WriteMessagesSpec extends AnyWordSpec with Matchers {
+
+ private val writes = AtomicWrite(PersistentRepr("a", 1, "p1")) :: Nil
+
+ "PersistentRepr" must {
+ "have a hashCode without a sender" in {
+ val repr = PersistentRepr("a", 1, "p1")
+ repr.sender should ===(ActorRef.noSender)
+ repr.hashCode should ===(PersistentRepr("a", 1, "p1").hashCode)
+ }
+ }
+
+ "JournalProtocol.WriteMessages" must {
+ "not bypass the circuit breaker by default" in {
+ WriteMessages(writes, ActorRef.noSender, 1).bypassCircuitBreaker should
===(false)
+ new WriteMessages(writes, ActorRef.noSender, 1).bypassCircuitBreaker
should ===(false)
+ WriteMessages(writes, ActorRef.noSender, 1, bypassCircuitBreaker =
true).bypassCircuitBreaker should ===(true)
+ }
+
+ "keep bypassCircuitBreaker in copy" in {
+ val w = WriteMessages(writes, ActorRef.noSender, 1, bypassCircuitBreaker
= true)
+ val copied = w.copy(actorInstanceId = 2)
+ copied.actorInstanceId should ===(2)
+ copied.bypassCircuitBreaker should ===(true)
+ w.withBypassCircuitBreaker(false).bypassCircuitBreaker should ===(false)
+ }
+
+ "extract the first 3 fields in pattern matching" in {
+ val w: Any = WriteMessages(writes, ActorRef.noSender, 3,
bypassCircuitBreaker = true)
+ w match {
+ case WriteMessages(messages, persistentActor, actorInstanceId) =>
+ messages should ===(writes)
+ persistentActor should ===(ActorRef.noSender)
+ actorInstanceId should ===(3)
+ case other =>
+ fail(s"unexpected $other")
+ }
+ }
+
+ "include bypassCircuitBreaker in equality" in {
+ WriteMessages(writes, ActorRef.noSender, 1) should
===(WriteMessages(writes, ActorRef.noSender, 1))
+ WriteMessages(writes, ActorRef.noSender, 1).hashCode should ===(
+ WriteMessages(writes, ActorRef.noSender, 1).hashCode)
+ WriteMessages(writes, ActorRef.noSender, 1) should !==(
+ WriteMessages(writes, ActorRef.noSender, 1, bypassCircuitBreaker =
true))
+ }
+
+ "have product elements for the first 3 fields" in {
+ val w = WriteMessages(writes, ActorRef.noSender, 1, bypassCircuitBreaker
= true)
+ w.productArity should ===(3)
+ w.productIterator.toList should ===(List(writes, ActorRef.noSender, 1))
+ }
+ }
+}
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/journal/AsyncWriteJournalResponseOrderSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/journal/AsyncWriteJournalResponseOrderSpec.scala
index d1e3b3bd0d..ce03231447 100644
---
a/persistence/src/test/scala/org/apache/pekko/persistence/journal/AsyncWriteJournalResponseOrderSpec.scala
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/journal/AsyncWriteJournalResponseOrderSpec.scala
@@ -18,7 +18,7 @@
package org.apache.pekko.persistence.journal
import scala.collection.mutable
-import scala.concurrent.{ ExecutionContext, Future, Promise }
+import scala.concurrent.{ ExecutionContext, Future, Promise, TimeoutException }
import scala.util.Try
import org.apache.pekko
@@ -51,6 +51,12 @@ class AsyncWriteJournalResponseOrderSpec
| plugin-dispatcher = "pekko.actor.default-dispatcher"
| write-response-global-order = off
| }
+ | bypass-call-timeout {
+ | class = "${classOf[ControlledWriteCompletionPlugin].getName}"
+ | plugin-dispatcher = "pekko.actor.default-dispatcher"
+ | write-response-global-order = on
+ | circuit-breaker.call-timeout = 500ms
+ | }
|}
|""".stripMargin
))) with ImplicitSender {
@@ -93,14 +99,37 @@ class AsyncWriteJournalResponseOrderSpec
}
}
- private def mkWriteMessages(persistenceId: Int):
JournalProtocol.WriteMessages = JournalProtocol.WriteMessages(
+ "AsyncWriteJournal with writes bypassing the circuit breaker" must {
+ "time out a bypassed write that never completes and release the responses
of later writes" in {
+ val pluginRef =
+ extension.journalFor(journalPluginId =
s"${ControlledWriteCompletionPlugin.BaseId}.bypass-call-timeout")
+
+ // never completed
+ pluginRef ! mkWriteMessages(persistenceId = 1, bypassCircuitBreaker =
true)
+ pluginRef ! mkWriteMessages(persistenceId = 2)
+ pluginRef ! CompleteWriteOps(persistenceIdsInOrder = Vector(2))
+
+ // the bypassed write fails with the call timeout, then the held back
response of the later write follows
+ expectMsgType[JournalProtocol.WriteMessagesFailed]
+ val failure = expectMsgType[JournalProtocol.WriteMessageFailure]
+ failure.message.persistenceId shouldEqual "1"
+ failure.cause shouldBe a[TimeoutException]
+ expectMsg(JournalProtocol.WriteMessagesSuccessful)
+ getPersistenceIdsFromResponses(receiveN(1)) shouldEqual Vector(2)
+ }
+ }
+
+ private def mkWriteMessages(
+ persistenceId: Int,
+ bypassCircuitBreaker: Boolean = false): JournalProtocol.WriteMessages =
JournalProtocol.WriteMessages(
messages = Vector(AtomicWrite(PersistentRepr(
payload = "",
sequenceNr = 0L,
persistenceId = persistenceId.toString
))),
persistentActor = self,
- actorInstanceId = 1
+ actorInstanceId = 1,
+ bypassCircuitBreaker = bypassCircuitBreaker
)
private def getPersistenceIdsFromResponses(responses: Seq[AnyRef]):
Vector[Int] = responses.collect {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
