Copilot commented on code in PR #6158: URL: https://github.com/apache/texera/pull/6158#discussion_r3524610869
########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/EmbeddedControlMessageManagerSpec.scala: ########## @@ -0,0 +1,164 @@ +/* + * 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.texera.amber.engine.architecture.worker + +import org.apache.texera.amber.core.tuple.Schema +import org.apache.texera.amber.core.virtualidentity.{ + ActorVirtualIdentity, + ChannelIdentity, + EmbeddedControlMessageIdentity, + OperatorIdentity, + PhysicalOpIdentity +} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.core.workflow.WorkflowContext.DEFAULT_WORKFLOW_ID +import org.apache.texera.amber.engine.architecture.messaginglayer.{ + InputManager, + NetworkInputGateway +} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + EmbeddedControlMessage, + EmbeddedControlMessageType +} +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement +import org.apache.texera.amber.util.VirtualIdentityUtils +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.util.concurrent.LinkedBlockingQueue + +class EmbeddedControlMessageManagerSpec extends AnyFlatSpec with Matchers { + + private val testOpId = PhysicalOpIdentity(OperatorIdentity("testop"), "main") + private val upstreamOpIdA = PhysicalOpIdentity(OperatorIdentity("senderA"), "main") + private val upstreamOpIdB = PhysicalOpIdentity(OperatorIdentity("senderB"), "main") + + private val actorId: ActorVirtualIdentity = + VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, testOpId, 0) + private val senderA: ActorVirtualIdentity = + VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdA, 0) + private val senderB: ActorVirtualIdentity = + VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdB, 0) + + private def mkGateway: NetworkInputGateway = new NetworkInputGateway(actorId) + + private def mkInputManager: InputManager = + new InputManager(actorId, new LinkedBlockingQueue[DPInputQueueElement]()) + + private def mkEcm( + ecmType: EmbeddedControlMessageType, + scope: Seq[ChannelIdentity] + ): EmbeddedControlMessage = + EmbeddedControlMessage( + EmbeddedControlMessageIdentity("ecm"), + ecmType, + scope, + Map.empty + ) + + "EmbeddedControlMessageManager" should "align ALL_ALIGNMENT only after every data channel is received" in { + val gateway = mkGateway + val inputManager = mkInputManager + val portId = PortIdentity() + + val chA = ChannelIdentity(senderA, actorId, isControl = false) + val chB = ChannelIdentity(senderB, actorId, isControl = false) + gateway.getChannel(chA).setPortId(portId) + gateway.getChannel(chB).setPortId(portId) + + val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager) + val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq.empty) + + // first data channel: not yet received from all channels in scope + mgr.isECMAligned(chA, ecm) shouldBe false + // second (last) data channel: now aligned + mgr.isECMAligned(chB, ecm) shouldBe true + } + + it should "align PORT_ALIGNMENT once all channels registered on the port are received" in { + val gateway = mkGateway + val inputManager = mkInputManager + val portId = PortIdentity() + + val chId = ChannelIdentity(senderA, actorId, isControl = false) + gateway.getChannel(chId).setPortId(portId) + + inputManager.addPort(portId, Schema(), List.empty, List.empty) + inputManager.getPort(portId).channels.add(chId) + + val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager) + val ecm = mkEcm(EmbeddedControlMessageType.PORT_ALIGNMENT, Seq(chId)) + + mgr.isECMAligned(chId, ecm) shouldBe true + } Review Comment: The PORT_ALIGNMENT test currently only has a single channel registered on the port, so it can’t exercise the “wait until *all* port channels have delivered the ECM” behavior (it will always align on the first receipt). Adding a second channel on the same port and asserting false-then-true would better cover the intended branch semantics. ########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/EmbeddedControlMessageManagerSpec.scala: ########## @@ -0,0 +1,164 @@ +/* + * 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.texera.amber.engine.architecture.worker + +import org.apache.texera.amber.core.tuple.Schema +import org.apache.texera.amber.core.virtualidentity.{ + ActorVirtualIdentity, + ChannelIdentity, + EmbeddedControlMessageIdentity, + OperatorIdentity, + PhysicalOpIdentity +} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.core.workflow.WorkflowContext.DEFAULT_WORKFLOW_ID +import org.apache.texera.amber.engine.architecture.messaginglayer.{ + InputManager, + NetworkInputGateway +} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + EmbeddedControlMessage, + EmbeddedControlMessageType +} +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement +import org.apache.texera.amber.util.VirtualIdentityUtils +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.util.concurrent.LinkedBlockingQueue + +class EmbeddedControlMessageManagerSpec extends AnyFlatSpec with Matchers { + + private val testOpId = PhysicalOpIdentity(OperatorIdentity("testop"), "main") + private val upstreamOpIdA = PhysicalOpIdentity(OperatorIdentity("senderA"), "main") + private val upstreamOpIdB = PhysicalOpIdentity(OperatorIdentity("senderB"), "main") + + private val actorId: ActorVirtualIdentity = + VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, testOpId, 0) + private val senderA: ActorVirtualIdentity = + VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdA, 0) + private val senderB: ActorVirtualIdentity = + VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdB, 0) + + private def mkGateway: NetworkInputGateway = new NetworkInputGateway(actorId) + + private def mkInputManager: InputManager = + new InputManager(actorId, new LinkedBlockingQueue[DPInputQueueElement]()) + + private def mkEcm( + ecmType: EmbeddedControlMessageType, + scope: Seq[ChannelIdentity] + ): EmbeddedControlMessage = + EmbeddedControlMessage( + EmbeddedControlMessageIdentity("ecm"), + ecmType, + scope, + Map.empty + ) + + "EmbeddedControlMessageManager" should "align ALL_ALIGNMENT only after every data channel is received" in { + val gateway = mkGateway + val inputManager = mkInputManager + val portId = PortIdentity() + + val chA = ChannelIdentity(senderA, actorId, isControl = false) + val chB = ChannelIdentity(senderB, actorId, isControl = false) + gateway.getChannel(chA).setPortId(portId) + gateway.getChannel(chB).setPortId(portId) + + val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager) + val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq.empty) + + // first data channel: not yet received from all channels in scope + mgr.isECMAligned(chA, ecm) shouldBe false + // second (last) data channel: now aligned + mgr.isECMAligned(chB, ecm) shouldBe true + } + + it should "align PORT_ALIGNMENT once all channels registered on the port are received" in { + val gateway = mkGateway + val inputManager = mkInputManager + val portId = PortIdentity() + + val chId = ChannelIdentity(senderA, actorId, isControl = false) + gateway.getChannel(chId).setPortId(portId) + + inputManager.addPort(portId, Schema(), List.empty, List.empty) + inputManager.getPort(portId).channels.add(chId) + + val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager) + val ecm = mkEcm(EmbeddedControlMessageType.PORT_ALIGNMENT, Seq(chId)) + + mgr.isECMAligned(chId, ecm) shouldBe true + } + + it should "align NO_ALIGNMENT only on the first received ECM" in { + val gateway = mkGateway + val inputManager = mkInputManager + val portId = PortIdentity() + + val chA = ChannelIdentity(senderA, actorId, isControl = false) + val chB = ChannelIdentity(senderB, actorId, isControl = false) + gateway.getChannel(chA).setPortId(portId) + gateway.getChannel(chB).setPortId(portId) + + val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager) + val ecm = mkEcm(EmbeddedControlMessageType.NO_ALIGNMENT, Seq.empty) + + // first ECM triggers + mgr.isECMAligned(chA, ecm) shouldBe true + // subsequent ECM does not trigger again + mgr.isECMAligned(chB, ecm) shouldBe false + } + + it should "throw IllegalArgumentException for an unsupported ECM type" in { + val gateway = mkGateway + val inputManager = mkInputManager + val portId = PortIdentity() + + val chId = ChannelIdentity(senderA, actorId, isControl = false) + gateway.getChannel(chId).setPortId(portId) + + val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager) + val ecm = mkEcm(EmbeddedControlMessageType.Unrecognized(-1), Seq.empty) + + assertThrows[IllegalArgumentException] { + mgr.isECMAligned(chId, ecm) + } + } + + it should "restrict the scope to channels targeting this worker when scope is non-empty" in { + val gateway = mkGateway + val inputManager = mkInputManager + val portId = PortIdentity() + + // an in-scope channel targeting this worker, plus an out-of-scope extra channel + val inScope = ChannelIdentity(senderA, actorId, isControl = false) + val extra = ChannelIdentity(senderB, actorId, isControl = false) + gateway.getChannel(inScope).setPortId(portId) + gateway.getChannel(extra).setPortId(portId) + + val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager) + // scope names only the in-scope channel (toWorkerId == actorId), so extra is ignored + val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq(inScope)) + + mgr.isECMAligned(inScope, ecm) shouldBe true + } Review Comment: The non-empty-scope test enters the `ecm.scope.nonEmpty` branch, but it doesn’t validate the key behavior of filtering out scope entries that *don’t* target this worker (`toWorkerId != actorId`). Including such a channel in the scope (and registering it in the gateway) would make the test actually assert the filter semantics. -- 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]
