aglinxinyuan commented on code in PR #6158: URL: https://github.com/apache/texera/pull/6158#discussion_r3531519569
########## 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: Good catch — updated to register two channels on the port and assert false-then-true (`chA` then `chB`), so the "wait until all port channels have delivered" branch is now actually exercised instead of aligning on first receipt. Fixed in ec86000927. ########## 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: Done. The scope now includes a `foreign` channel whose `toWorkerId != actorId` (registered in the gateway), so the `_.toWorkerId == actorId` filter in `getChannelsWithinScope` is genuinely asserted: alignment is reached on `inScope` alone precisely because `foreign` is filtered out. Fixed in ec86000927. -- 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]
