pjfanning commented on code in PR #3653:
URL: https://github.com/apache/pekko/pull/3653#discussion_r4358370263
##########
persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala:
##########
@@ -229,8 +232,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) asyncWriteMessages(prep)
Review Comment:
Fixed in fd68ca81ad. A bypassed write is now bounded by
`circuit-breaker.call-timeout` through its own scheduled timeout, without
counting towards the breaker's failures. It fails with a `TimeoutException`,
which releases its resequencer slot. `AsyncWriteJournalResponseOrderSpec` has a
new plugin config with global response order and a 500ms call timeout. In the
test, a bypassed write never completes and is followed by an ordinary write.
The bypassed write gets `WriteMessageFailure` (timeout), and then the ordinary
write's success is released. Without the fix, the second response never arrives.
##########
persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala:
##########
@@ -269,8 +417,124 @@ private[pekko] object EventWriter {
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((repr, replyTo)))
Review Comment:
Fixed in fd68ca81ad. `handleJournalResponse` now handles
`WriteMessageRejected` for the current transaction: it fails the waiting write
and its duplicates with `Journal rejected write` (rejection as cause) and moves
on to the next batch. New test: a gap-fill batch `[1, 2, 3]` is rejected and
the caller gets the error. A later write of 4 then looks up the highest
sequence number again and writes `[1, 2, 3, 4]` successfully.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
