He-Pin commented on code in PR #3653:
URL: https://github.com/apache/pekko/pull/3653#discussion_r4355898953
##########
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:
Gap fillers can be rejected before storage, for example by a broadly bound
EventAdapter that does not accept FilteredPayload. AsyncWriteJournal reports
these as WriteMessageRejected, but handleJournalResponse currently handles only
WriteMessageSuccess and WriteMessageFailure and ignores the rejection. The
pending callers are then left unresolved. Please handle the rejection outcome
and cover a rejected atomic gap-fill batch followed by a later successful write.
##########
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)))
+ } 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 :+ ((repr, replyTo)))
+ } 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)) {
+ 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")
+ 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) {
+ context.log.trace2("Duplicate seq nr [{}] for persistence id
[{}]", sequenceNumber, persistenceId)
+ replyTo ! StatusReply.success(WriteAck(persistenceId,
sequenceNumber)) // duplicate
Review Comment:
`nextExpectedSeqNr` includes events in waitingForReply and waitingForWrite,
so a lower sequence number is not necessarily durable yet. With gap filling
enabled, send sequence 1 twice before the journal acknowledges the first write:
the second caller receives success here even if the pending write subsequently
fails. Please retain duplicate callers until the corresponding journal outcome
(or a highest-sequence-number lookup proving persistence), and assert that no
caller is acknowledged while that write is unresolved.
##########
persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala:
##########
@@ -378,6 +612,58 @@ private[pekko] object EventWriter {
}
Behaviors.same
+ case MaxSeqNrForPid(pid, maxSeqNr, reason @
(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, waitingForSeqNrLookup =
Vector.empty))
+ waiting.foreach {
+ case (repr, replyTo) =>
+ handleWrite(
+ repr,
+ replyTo,
+ isSnapshotEvent = reason ==
AskMaxSeqNrReason.SnapshotEvent,
Review Comment:
The lookup reason belongs to its initiating request, but this applies it to
every queued request. With general gap filling disabled, queue a normal event
behind a snapshot event while the snapshot's lookup is outstanding: the normal
event is replayed with `isSnapshotEvent = true` and its gap is filled too.
Please retain the flag per queued request and add a mixed
snapshot/ordinary-event regression.
##########
persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala:
##########
@@ -378,6 +612,58 @@ private[pekko] object EventWriter {
}
Behaviors.same
+ case MaxSeqNrForPid(pid, maxSeqNr, reason @
(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, waitingForSeqNrLookup =
Vector.empty))
+ waiting.foreach {
+ case (repr, replyTo) =>
+ handleWrite(
+ repr,
+ replyTo,
+ isSnapshotEvent = reason ==
AskMaxSeqNrReason.SnapshotEvent,
+ calledAfterMaxSeqNr = true)
+ }
+ }
+ Behaviors.same
+
+ case MaxSeqNrForPidFailed(pid, errorDescOpt) =>
+ // journal seems to have real problems, let the circuit breaker
protect it until next successful write
+ bypassCircuitBreaker = false
+ def errorDescInLog = errorDescOpt match {
+ case AskMaxSeqNrReason.WriteFailure(errorDesc) => s", original
error desc: $errorDesc"
+ case _ => ""
+ }
+ perPidWriteState.get(pid) match {
+ case None =>
+ context.log.debug2(
+ "Got max seq nr (failed) with no waiting previous state for
persistence id [{}]{}",
+ pid,
+ errorDescInLog)
+ case Some(state) =>
+ state.waitingForReply.values.foreach {
+ case (_, replyTo) =>
+ replyTo ! StatusReply.error("Journal write failed")
+ }
+ val sortedSeqs = state.waitingForReply.keys.toSeq.sorted
Review Comment:
For the initial FillGaps/SnapshotEvent lookup, line 433 puts the request in
`waitingForSeqNrLookup` and leaves `waitingForReply` empty. A
ReplayMessagesFailure or ask timeout therefore reaches `sortedSeqs.head/last`
on an empty collection and restarts the writer. Even guarding the log would
leave the lookup queue unanswered and uncleared. Please fail and clear every
affected waiter, preserve the failure cause, and test both initial lookup
failure and timeout followed by a successful later write.
##########
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:
This also removes the journal circuit-breaker call timeout. If a bypassed
write's Future never completes, its resequencer slot is never released; with
global response ordering enabled, later ordinary writes to the same journal can
lose their responses too. Please preserve a bounded call timeout independently
of breaker failure accounting, and test a hanging bypassed write followed by an
ordinary write whose response must eventually be released.
##########
persistence/src/test/scala/org/apache/pekko/persistence/WriteMessagesSpec.scala:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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
+
+ "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 ===(
Review Comment:
The current next CI failure is here: [job
log](https://github.com/apache/pekko/actions/runs/39882269597/job/120182587311).
Hashing `writes` reaches `PersistentImpl.hashCode(Persistent.scala:234)`,
which passes the default null sender to `HashCode.hash` and throws
`IllegalArgumentException: Unexpected hash parameter: null`. The outer
WriteMessages hash already handles a null persistentActor, so that is not the
failing value. Please fix the fixture or the underlying PersistentRepr
null-sender contract and rerun the focused spec plus the cancelled matrix jobs.
--
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]
