pjfanning commented on code in PR #3653:
URL: https://github.com/apache/pekko/pull/3653#discussion_r4359837325
##########
persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala:
##########
@@ -378,6 +687,62 @@ 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, 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)
+ }
Review Comment:
Fixed in 9ddc41a42c. After the writes queued for a lookup are replayed, an
idle state (nothing in flight, no lookup in progress) now goes through
`handleUpdatedStateForPid`, so the cache eviction applies. The lookup also
refreshes `usedTimestamp`. New `EventWriterFillGapsSpec` test, as you
described: cache capacity 2, sequence 5 for six pids, each lookup answered with
highest 5. Sending sequence 6 for all six then gives 2 direct `WriteMessages`
and 4 lookups. Against fd68ca81ad it gives 6 writes.
##########
persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala:
##########
@@ -264,13 +449,168 @@ 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)) {
+ 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) {
+ 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 (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)
Review Comment:
Fixed in 9ddc41a42c. In the gap branch, a write added to a pending batch now
goes through the same max-batch check as an ordinary write before the fillers
are created. Filled gaps still do not count towards the limit. New
`EventWriterFillGapsSpec` test with max-batch-size 2: 1 is in flight, then 3, 5
and 7 are sent. 7 is rejected with the max batch error, and the next batch is
`[2, 3, 4, 5]`. This fails against fd68ca81ad.
--
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]
