This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8002-e750074eb92f57a6a945b2a09d07b8e742c0b82f in repository https://gitbox.apache.org/repos/asf/texera.git
commit e54ecd2970ac8d2099c1e8f061cf053d32dd5e85 Author: anthonychengit <[email protected]> AuthorDate: Fri Aug 28 03:37:51 2026 +0000 fix(amber): guard resends by destination (#8002) ### What changes were proposed in this PR? Use the destination worker, rather than the sender, when deciding whether an unacknowledged network message can be resent. ```text Before: timeout -> sender exists -> resend toward a removed destination After: timeout -> destination exists -> resend; otherwise keep the message pending ``` Regression tests cover both directions: a registered destination with an unknown sender resends successfully, while a registered sender with a missing destination does not resend. ### Any related issues, documentation, discussions? Closes #6921. ### How was this PR tested? The two regression tests were added first and both failed against the old sender guard. After the one-line fix: ```text sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.common.PekkoMessageTransferServiceSpec" ``` Result: 5 tests passed. ```text sbt "scalafixAll --check" "scalafmtCheckAll" git diff --check ``` Result: all lint, formatting, and diff checks passed. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Codex was used for implementation and verification assistance. I reviewed the final code and test output before submission. --- .../common/PekkoMessageTransferService.scala | 2 +- .../common/PekkoMessageTransferServiceSpec.scala | 63 ++++++++++++++++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferService.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferService.scala index 16c9e394ca..6af9d6cdcf 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferService.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferService.scala @@ -172,7 +172,7 @@ class PekkoMessageTransferService( if (msgsNeedResend.nonEmpty) { logger.debug(s"output for $channel: ${cc.getStatusReport}") } - if (refService.hasActorRef(channel.fromWorkerId)) { + if (refService.hasActorRef(channel.toWorkerId)) { msgsNeedResend.foreach { msg => refService.forwardToActor(msg) } diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferServiceSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferServiceSpec.scala index a3858c8a9a..0653cd6348 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/common/PekkoMessageTransferServiceSpec.scala @@ -80,6 +80,13 @@ class PekkoMessageTransferServiceSpec private def net(id: Long, chan: ChannelIdentity, seq: Long): NetworkMessage = NetworkMessage(id, fifo(chan, seq)) + private def backdateSentTime(cc: CongestionControl, id: Long): Unit = { + val field = classOf[CongestionControl].getDeclaredField("sentTime") + field.setAccessible(true) + val sentTime = field.get(cc).asInstanceOf[mutable.LongMap[Long]] + sentTime(id) = System.currentTimeMillis() - cc.resendTimeLimit - 1 + } + // Pin the assumed payload size so this test fails loudly if the size accounting // changes in a way that would invalidate the credit math below. assert(WorkflowMessage.getInMemSize(fifo(dataChannel(), 0L)) == 200L) @@ -189,6 +196,46 @@ class PekkoMessageTransferServiceSpec service.stop() // cancels the (already-cancelled) captured handles; must not throw assert(service.channelToCC.contains(chan)) } + + "checkResend" should "resend a timed-out message when its destination is registered" in { + val actorService = new CapturingActorService(actorId, freshContext()) + val sender = ActorVirtualIdentity("known-sender") + val destination = ActorVirtualIdentity("known-destination") + val refService = new RecordingRefService(actorService, Set(destination)) + val service = new PekkoMessageTransferService(actorService, refService, _ => ()) + val channel = ChannelIdentity(sender, destination, isControl = false) + val message = net(10L, channel, 0L) + val cc = new CongestionControl() + cc.markMessageInTransit(message) + backdateSentTime(cc, message.messageId) + service.channelToCC(channel) = cc + + service.initialize() + actorService.capturedCallables.head() + + assert(refService.queriedIds == Seq(destination)) + assert(refService.forwardedMessages == Seq(message)) + } + + it should "not resend a timed-out message when only its sender is registered" in { + val actorService = new CapturingActorService(actorId, freshContext()) + val sender = ActorVirtualIdentity("known-sender") + val destination = ActorVirtualIdentity("removed-destination") + val refService = new RecordingRefService(actorService, Set(sender)) + val service = new PekkoMessageTransferService(actorService, refService, _ => ()) + val channel = ChannelIdentity(sender, destination, isControl = false) + val message = net(11L, channel, 0L) + val cc = new CongestionControl() + cc.markMessageInTransit(message) + backdateSentTime(cc, message.messageId) + service.channelToCC(channel) = cc + + service.initialize() + actorService.capturedCallables.head() + + assert(refService.queriedIds == Seq(destination)) + assert(refService.forwardedMessages.isEmpty) + } } /** Minimal actor used only to obtain a real `ActorContext` from Pekko TestKit. */ @@ -215,3 +262,19 @@ class CapturingActorService(vid: ActorVirtualIdentity, ac: ActorContext) Cancellable.alreadyCancelled } } + +class RecordingRefService( + actorService: PekkoActorService, + knownIds: Set[ActorVirtualIdentity] +) extends PekkoActorRefMappingService(actorService) { + + val queriedIds: mutable.ArrayBuffer[ActorVirtualIdentity] = mutable.ArrayBuffer() + val forwardedMessages: mutable.ArrayBuffer[NetworkMessage] = mutable.ArrayBuffer() + + override def hasActorRef(id: ActorVirtualIdentity): Boolean = { + queriedIds += id + knownIds.contains(id) + } + + override def forwardToActor(msg: NetworkMessage): Unit = forwardedMessages += msg +}
