aglinxinyuan commented on code in PR #5451: URL: https://github.com/apache/texera/pull/5451#discussion_r3368791257
########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManagerSpec.scala: ########## @@ -0,0 +1,222 @@ +/* + * 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.messaginglayer + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.Partitioning +import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement +import org.scalatest.flatspec.AnyFlatSpec + +import java.net.URI +import java.util.concurrent.LinkedBlockingQueue + +class InputManagerSpec extends AnyFlatSpec { + + // --------------------------------------------------------------------------- + // Fixtures + // --------------------------------------------------------------------------- + + private val actorId: ActorVirtualIdentity = ActorVirtualIdentity("worker-1") + + private def emptyQueue(): LinkedBlockingQueue[DPInputQueueElement] = + new LinkedBlockingQueue[DPInputQueueElement]() + + private def freshManager: InputManager = new InputManager(actorId, emptyQueue()) + + // A one-field schema is enough to build real Tuple instances for the batch tests. + private val intAttr = new Attribute("v", AttributeType.INTEGER) + private val schema: Schema = Schema().add(intAttr) + private def tuple(value: Int): Tuple = + Tuple.builder(schema).add(intAttr, Integer.valueOf(value)).build() + + private def channelId(from: String, to: String): ChannelIdentity = + ChannelIdentity(ActorVirtualIdentity(from), ActorVirtualIdentity(to), isControl = false) + + // --------------------------------------------------------------------------- + // getAllPorts / addPort / getPort + // --------------------------------------------------------------------------- + + "InputManager.getAllPorts (fresh)" should "be empty" in { + val mgr = freshManager + assert(mgr.getAllPorts.isEmpty) + } + + "InputManager.addPort" should + "register a fresh WorkerPort under the supplied portId (no reader threads when URI list is empty)" in { + val mgr = freshManager + val portId = PortIdentity(0) + mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty) + assert(mgr.getAllPorts == Set(portId)) + val port = mgr.getPort(portId) + assert(port.schema == schema) + assert(!port.completed) + assert(mgr.getInputPortReaderThreads.isEmpty) + } + + it should "be a no-op when called a second time for the same portId (first call wins)" in { + // Pin the "each port can only be added and initialized once" comment in + // the production code: the second addPort returns early WITHOUT + // overwriting the existing WorkerPort. We prove this by mutating the + // first port's `completed` flag, then re-adding, then checking the + // mutation survives — if the second call had replaced the WorkerPort, + // `completed` would revert to its case-class default `false`. + val mgr = freshManager + val portId = PortIdentity(0) + mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty) + val firstPort = mgr.getPort(portId) + firstPort.completed = true + mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty) + assert(mgr.getPort(portId) eq firstPort, "second addPort must not replace the WorkerPort") + assert(mgr.getPort(portId).completed, "second addPort must not reset the existing port's state") + } + + it should + "throw AssertionError when urisToRead.size does not match partitionings.size" in { + val mgr = freshManager + val portId = PortIdentity(0) + // 1 URI but 0 partitionings — sizes diverge. + assertThrows[AssertionError] { + mgr.addPort( + portId, + schema, + urisToRead = List(new URI("file:///nowhere")), + partitionings = List.empty[Partitioning] + ) + } + } + + "InputManager.getPort" should "return the WorkerPort previously registered by addPort" in { + val mgr = freshManager + val portId = PortIdentity(7) + mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty) + val a = mgr.getPort(portId) + val b = mgr.getPort(portId) + assert(a eq b, "getPort must be a pure lookup — no fresh WorkerPort each call") + } + + "InputManager.getInputPortReaderThreads" should + "be empty when no port has been added with URIs" in { + val mgr = freshManager + mgr.addPort(PortIdentity(0), schema, urisToRead = List.empty, partitionings = List.empty) + mgr.addPort(PortIdentity(1), schema, urisToRead = List.empty, partitionings = List.empty) + assert(mgr.getInputPortReaderThreads.isEmpty) + } + + // --------------------------------------------------------------------------- + // Batch cursor — initBatch, hasUnfinishedInput, getNextTuple, getCurrentTuple + // --------------------------------------------------------------------------- + + "InputManager.hasUnfinishedInput (fresh)" should "be false when no batch has been initialized" in { + val mgr = freshManager + assert(!mgr.hasUnfinishedInput) + } + + "InputManager.getCurrentTuple (fresh)" should "return null when no batch has been initialized" in { + val mgr = freshManager + assert(mgr.getCurrentTuple == null) + } + + "InputManager.initBatch" should + "swap in a new batch, reset the cursor, and surface the channel id" in { Review Comment: Fixed in 343e720855 — added a `getNextTuple` assertion that verifies the first element of the new batch is returned, directly proving `currentInputIdx` was reset. -- 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]
