pjfanning commented on code in PR #2762:
URL: https://github.com/apache/pekko/pull/2762#discussion_r2963107646
##########
persistence-typed/src/test/scala/org/apache/pekko/persistence/typed/delivery/ReliableDeliveryWithEventSourcedProducerQueueSpec.scala:
##########
@@ -178,6 +179,71 @@ class
ReliableDeliveryWithEventSourcedProducerQueueSpec(config: Config)
testKit.stop(consumerController)
}
+ "resume correctly after restart when all messages were confirmed" in {
+ val producerId = "p-restart-clean"
+ val producerProbe =
createTestProbe[ProducerController.RequestNext[String]]()
+ val consumerProbe =
createTestProbe[ConsumerController.Delivery[String]]()
+
+ // Phase 1: send one message and confirm it fully, then stop both
controllers cleanly.
+ val (pc1, cc1) = startProducerAndConsumer(producerId, producerProbe,
consumerProbe)
+ producerProbe.receiveMessage().sendNextTo ! "msg-1"
+ val del1 = consumerProbe.receiveMessage()
+ del1.confirmTo ! ConsumerController.Confirmed
+ // Wait for the ProducerController to process the confirmation and issue
a new RequestNext.
+ // The seqNr is captured dynamically: for non-chunked messages it will
be 2; for chunked
+ // messages each byte is a separate seqNr, so it will be higher (e.g. 6
for a 5-byte string).
+ // Note: StoreMessageConfirmed is intentionally write-behind
(fire-and-forget, no reply), so
+ // the confirmation may not yet be persisted to the journal when pc1 is
stopped below.
+ // If that happens, pc2 will re-deliver the earlier message — Phase 2
handles this case.
+ val nextSeqNr = producerProbe.receiveMessage().currentSeqNr
+ testKit.stop(pc1)
+ producerProbe.expectTerminated(pc1)
+ testKit.stop(cc1)
+ consumerProbe.expectTerminated(cc1)
Review Comment:
df1380f220c5ec739e7ea3913c72295e0b6f5375 attempts to address this
--
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]
