pjfanning commented on code in PR #2018:
URL: https://github.com/apache/pekko-connectors/pull/2018#discussion_r4371026926
##########
amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AbstractAmqpAsyncFlowStageLogic.scala:
##########
@@ -79,24 +82,23 @@ import scala.concurrent.Promise
private def onConfirmation(tag: DeliveryTag, multiple: Boolean): Unit = {
log.debug("Received confirmation for deliveryTag {} (multiple={}).", tag,
multiple)
-
- val dequeued: Iterable[AwaitingMessage[T]] = dequeueAwaitingMessages(tag,
multiple)
-
- dequeued.foreach(m => cancelTimer(m.tag))
-
- pushOrEnqueueResults(
- dequeued.map(m => (WriteResult.confirmed, m.passThrough)))
+ onResult(tag, multiple, WriteResult.confirmed)
}
private def onRejection(tag: DeliveryTag, multiple: Boolean): Unit = {
log.debug("Received rejection for deliveryTag {} (multiple={}).", tag,
multiple)
+ onResult(tag, multiple, WriteResult.rejected)
+ }
- val dequeued: Iterable[AwaitingMessage[T]] = dequeueAwaitingMessages(tag,
multiple)
+ private def onResult(tag: DeliveryTag, multiple: Boolean, result:
WriteResult): Unit = {
+ val dequeued: Iterable[AwaitingMessage[T]] = dequeueAwaitingMessages(tag,
multiple, result)
+ // Stop the timeout even if this result must wait for earlier messages
before it can be emitted.
+ cancelTimer(tag)
Review Comment:
Nothing in the tests would fail if this `cancelTimer(tag)` were removed,
because first-result-wins hides a stale timer firing. One option: confirm
message 2 while message 1 is still pending, wait past the confirmation timeout,
then confirm message 1 and expect both to be `confirmed`. To actually catch a
missing cancel, the test would also need to assert that nothing was emitted
early or that no timeout fired (e.g. via a log/debug hook), so this may not be
worth the effort. Mentioning it for completeness.
##########
amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpFlowSpec.scala:
##########
@@ -185,6 +186,52 @@ class AmqpFlowSpec extends AmqpSpec with AmqpMocking with
BeforeAndAfterEach {
messages should contain theSameElementsInOrderAs expectedResult
completion.futureValue shouldBe an[Done]
}
+
+ "preserve individual results when confirmations and rejections arrive out
of order" in assertAllStagesStopped {
+ val input = Vector("one", "two", "three", "four")
+ val (completion, probe, confirmCallback, rejectCallback) =
publishWithOrderedConfirm(input)
+
+ rejectCallback.handle(2L, false)
+ confirmCallback.handle(1L, false)
+ confirmCallback.handle(4L, false)
+ rejectCallback.handle(3L, false)
+
+ val expectedResult = Seq(
+ (WriteResult.confirmed, input(0)), (WriteResult.rejected, input(1)),
(WriteResult.rejected, input(2)),
+ (WriteResult.confirmed, input(3)))
+
+ probe.expectNextN(input.size) should contain theSameElementsInOrderAs
expectedResult
+ completion.futureValue shouldBe an[Done]
+ }
+
+ "preserve a rejection when a later batch confirmation includes the
message" in assertAllStagesStopped {
+ val input = Vector("one", "two", "three", "four")
+ val (completion, probe, confirmCallback, rejectCallback) =
publishWithOrderedConfirm(input)
+
+ rejectCallback.handle(2L, false)
+ // Message 4 is outside the batch but should be emitted once the earlier
messages have results.
+ confirmCallback.handle(4L, false)
+ confirmCallback.handle(3L, true)
+
+ val expectedResult = Seq(
+ (WriteResult.confirmed, input(0)), (WriteResult.rejected, input(1)),
(WriteResult.confirmed, input(2)),
+ (WriteResult.confirmed, input(3)))
+
+ probe.expectNextN(input.size) should contain theSameElementsInOrderAs
expectedResult
+ completion.futureValue shouldBe an[Done]
+ }
+
+ "preserve a confirmation when an earlier message times out" in
assertAllStagesStopped {
+ val input = Vector("one", "two")
+ val (completion, probe, confirmCallback, _) =
publishWithOrderedConfirm(input, confirmationTimeout = 200.millis)
Review Comment:
This 200ms timeout applies to both messages, and `confirmCallback.handle(2L,
false)` only runs after two polling Mockito `verify` calls. On a slow or
GC-paused CI runner that gap could exceed 200ms, so message 2 would time out
first and be emitted as `rejected`, making the test flaky. Could this be raised
to something like 1s? It adds little runtime and leaves a much larger margin.
--
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]
