Copilot commented on code in PR #8105: URL: https://github.com/apache/texera/pull/8105#discussion_r3886499969
########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/pythonworker/PythonWorkflowWorkerSpec.scala: ########## @@ -0,0 +1,302 @@ +/* + * 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.pythonworker + +import com.twitter.util.Promise +import org.apache.commons.lang3.SystemUtils +import org.apache.pekko.actor.ActorSystem +import org.apache.pekko.testkit.{ImplicitSender, TestActorRef, TestKit} +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.common.WorkflowActor +import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{NetworkAck, NetworkMessage} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + AsyncRPCContext, + ControlInvocation, + EmptyRequest +} +import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.apache.texera.amber.engine.common.ambermessage.WorkflowMessage.getInMemSize +import org.apache.texera.amber.engine.architecture.scheduling.config.WorkerConfig +import org.apache.texera.amber.engine.common.AmberRuntime +import org.apache.texera.common.config.{PythonUtils, UdfConfig} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.matchers.should.Matchers + +import java.nio.file.{Files, Path, Paths} +import java.util.concurrent.ExecutorService +import java.util.concurrent.atomic.AtomicBoolean +import scala.concurrent.duration._ +import scala.sys.process.Process + +/** + * Unit tests for the Python worker actor. + * + * The worker's `initState` starts a proxy server, spawns a real Python interpreter and starts a + * Flight client, none of which belongs in a unit test and none of which is guaranteed to exist on + * a CI runner. The class offers no seam for that, but `initState` is an ordinary override, so the + * suite subclasses the worker and makes it a no-op. Everything below then runs against a real + * actor instance with nothing spawned. + * + * Two members are read by reflection because they are `private`: `choosePythonBin`, which has no + * caller other than the process-spawning path, and the executors/port promise, which the teardown + * assertions need. Reflection into `private` members is an established pattern in this module's + * specs. `portNumberPromise` in particular has to be fulfilled before anything touches + * `pythonProxyClient`: that client blocks in its own constructor on `Await.result(promise)`, so a + * worker whose proxy server never started would otherwise hang the moment `postStop` ran. + * + * Out of scope, deliberately: `startPythonProcess`, which runs a real interpreter, and + * `startProxyServer`, which retries a socket bind until it succeeds. + */ +class PythonWorkflowWorkerSpec + extends TestKit(ActorSystem("PythonWorkflowWorkerSpec", AmberRuntime.pekkoConfig)) + with ImplicitSender + with AnyFlatSpecLike + with Matchers + with BeforeAndAfterAll { + + /** A cuid no other suite writes under, so the venv fixtures cannot collide. */ + private val FixtureCuid = 987654 + private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs") + + override def afterAll(): Unit = { + deleteRecursively(VenvRoot.resolve(FixtureCuid.toString)) + TestKit.shutdownActorSystem(system) + } + + /** The worker with its process/socket startup removed; see the class comment. */ + private class InertPythonWorkflowWorker(config: WorkerConfig) + extends PythonWorkflowWorker(config) { + override def initState(): Unit = () + } + + /** Stands in for the OS process the worker would have spawned. */ + private class RecordingProcess(failOnDestroy: Boolean = false) extends Process { + val destroyed = new AtomicBoolean(false) + override def isAlive(): Boolean = !destroyed.get() + override def exitValue(): Int = 0 + override def destroy(): Unit = { + destroyed.set(true) + if (failOnDestroy) throw new IllegalStateException("destroy failed") + } + } + + private def newWorker(config: WorkerConfig): TestActorRef[InertPythonWorkflowWorker] = + TestActorRef[InertPythonWorkflowWorker](new InertPythonWorkflowWorker(config)) + + private def config(pveName: String = "", cuid: Option[Int] = None): WorkerConfig = + WorkerConfig(ActorVirtualIdentity("python-worker-spec"), pveName = pveName, cuid = cuid) + + private def readPrivate[T](actor: PythonWorkflowWorker, name: String): T = { + val method = classOf[PythonWorkflowWorker].getDeclaredMethod(name) + method.setAccessible(true) + method.invoke(actor).asInstanceOf[T] + } + + private def chosenPythonBin(actor: PythonWorkflowWorker): String = + readPrivate[String](actor, "choosePythonBin") + + /** Releases the port the proxy client's constructor blocks on. */ + private def openPort(actor: PythonWorkflowWorker, port: Int = 65535): Unit = + readPrivate[Promise[Int]](actor, "portNumberPromise").setValue(port) + + private def setServerProcess(actor: PythonWorkflowWorker, process: Process): Unit = { + val setter = + classOf[PythonWorkflowWorker].getDeclaredMethod("pythonServerProcess_$eq", classOf[Process]) + setter.setAccessible(true) + setter.invoke(actor, process) + } + + /** The interpreter PveManager resolves for (cuid, name); POSIX and Windows differ. */ + private def venvPython(cuid: Int, pveName: String): Path = { + val venv = VenvRoot.resolve(cuid.toString).resolve(pveName).resolve("pve") + if (SystemUtils.IS_OS_WINDOWS) venv.resolve("Scripts").resolve("python.exe") + else venv.resolve("bin").resolve("python") + } + + private def deleteRecursively(path: Path): Unit = { + if (Files.exists(path)) { + Files + .walk(path) + .sorted(java.util.Comparator.reverseOrder[Path]()) + .forEach(p => Files.deleteIfExists(p)) + } + } Review Comment: `Files.walk(path)` returns a stream that should be closed; as written, this can leak file descriptors (especially on Windows) and intermittently fail deletions due to open handles. Wrap the walk in a try/finally (or `Using.resource`) and close the stream. ########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/pythonworker/PythonWorkflowWorkerSpec.scala: ########## @@ -0,0 +1,302 @@ +/* + * 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.pythonworker + +import com.twitter.util.Promise +import org.apache.commons.lang3.SystemUtils +import org.apache.pekko.actor.ActorSystem +import org.apache.pekko.testkit.{ImplicitSender, TestActorRef, TestKit} +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.common.WorkflowActor +import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{NetworkAck, NetworkMessage} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + AsyncRPCContext, + ControlInvocation, + EmptyRequest +} +import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.apache.texera.amber.engine.common.ambermessage.WorkflowMessage.getInMemSize +import org.apache.texera.amber.engine.architecture.scheduling.config.WorkerConfig +import org.apache.texera.amber.engine.common.AmberRuntime +import org.apache.texera.common.config.{PythonUtils, UdfConfig} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.matchers.should.Matchers + +import java.nio.file.{Files, Path, Paths} +import java.util.concurrent.ExecutorService +import java.util.concurrent.atomic.AtomicBoolean +import scala.concurrent.duration._ +import scala.sys.process.Process + +/** + * Unit tests for the Python worker actor. + * + * The worker's `initState` starts a proxy server, spawns a real Python interpreter and starts a + * Flight client, none of which belongs in a unit test and none of which is guaranteed to exist on + * a CI runner. The class offers no seam for that, but `initState` is an ordinary override, so the + * suite subclasses the worker and makes it a no-op. Everything below then runs against a real + * actor instance with nothing spawned. + * + * Two members are read by reflection because they are `private`: `choosePythonBin`, which has no + * caller other than the process-spawning path, and the executors/port promise, which the teardown + * assertions need. Reflection into `private` members is an established pattern in this module's + * specs. `portNumberPromise` in particular has to be fulfilled before anything touches + * `pythonProxyClient`: that client blocks in its own constructor on `Await.result(promise)`, so a + * worker whose proxy server never started would otherwise hang the moment `postStop` ran. + * + * Out of scope, deliberately: `startPythonProcess`, which runs a real interpreter, and + * `startProxyServer`, which retries a socket bind until it succeeds. + */ +class PythonWorkflowWorkerSpec + extends TestKit(ActorSystem("PythonWorkflowWorkerSpec", AmberRuntime.pekkoConfig)) + with ImplicitSender + with AnyFlatSpecLike + with Matchers + with BeforeAndAfterAll { + + /** A cuid no other suite writes under, so the venv fixtures cannot collide. */ + private val FixtureCuid = 987654 + private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs") + + override def afterAll(): Unit = { + deleteRecursively(VenvRoot.resolve(FixtureCuid.toString)) + TestKit.shutdownActorSystem(system) + } + + /** The worker with its process/socket startup removed; see the class comment. */ + private class InertPythonWorkflowWorker(config: WorkerConfig) + extends PythonWorkflowWorker(config) { + override def initState(): Unit = () + } + + /** Stands in for the OS process the worker would have spawned. */ + private class RecordingProcess(failOnDestroy: Boolean = false) extends Process { + val destroyed = new AtomicBoolean(false) + override def isAlive(): Boolean = !destroyed.get() + override def exitValue(): Int = 0 + override def destroy(): Unit = { + destroyed.set(true) + if (failOnDestroy) throw new IllegalStateException("destroy failed") + } + } + + private def newWorker(config: WorkerConfig): TestActorRef[InertPythonWorkflowWorker] = + TestActorRef[InertPythonWorkflowWorker](new InertPythonWorkflowWorker(config)) + + private def config(pveName: String = "", cuid: Option[Int] = None): WorkerConfig = + WorkerConfig(ActorVirtualIdentity("python-worker-spec"), pveName = pveName, cuid = cuid) + + private def readPrivate[T](actor: PythonWorkflowWorker, name: String): T = { + val method = classOf[PythonWorkflowWorker].getDeclaredMethod(name) + method.setAccessible(true) + method.invoke(actor).asInstanceOf[T] + } + + private def chosenPythonBin(actor: PythonWorkflowWorker): String = + readPrivate[String](actor, "choosePythonBin") + + /** Releases the port the proxy client's constructor blocks on. */ + private def openPort(actor: PythonWorkflowWorker, port: Int = 65535): Unit = + readPrivate[Promise[Int]](actor, "portNumberPromise").setValue(port) + + private def setServerProcess(actor: PythonWorkflowWorker, process: Process): Unit = { + val setter = + classOf[PythonWorkflowWorker].getDeclaredMethod("pythonServerProcess_$eq", classOf[Process]) + setter.setAccessible(true) + setter.invoke(actor, process) + } + + /** The interpreter PveManager resolves for (cuid, name); POSIX and Windows differ. */ + private def venvPython(cuid: Int, pveName: String): Path = { + val venv = VenvRoot.resolve(cuid.toString).resolve(pveName).resolve("pve") + if (SystemUtils.IS_OS_WINDOWS) venv.resolve("Scripts").resolve("python.exe") + else venv.resolve("bin").resolve("python") + } + + private def deleteRecursively(path: Path): Unit = { + if (Files.exists(path)) { + Files + .walk(path) + .sorted(java.util.Comparator.reverseOrder[Path]()) + .forEach(p => Files.deleteIfExists(p)) + } + } + + behavior of "PythonWorkflowWorker" + + it should "resolve the Python sources under the amber home" in { + val worker = newWorker(config()) + val path = worker.underlyingActor.pythonSrcDirectory + + // Asserted by tail rather than as an absolute string: the amber home moves with the checkout. + path.getFileName.toString shouldBe "python" + path.getParent.getFileName.toString shouldBe "main" + path.getParent.getParent.getFileName.toString shouldBe "src" + } + + it should "expose the configured R path with its surrounding whitespace removed" in { + val worker = newWorker(config()) + + worker.underlyingActor.RENVPath shouldBe UdfConfig.rPath.trim + worker.underlyingActor.RENVPath shouldBe worker.underlyingActor.RENVPath.trim + } + + it should "fall back to the default interpreter when the worker has no computing unit" in { + val worker = newWorker(config(pveName = "myenv", cuid = None)) + + chosenPythonBin(worker.underlyingActor) shouldBe PythonUtils.getPythonExecutable + } + + it should "fall back to the default interpreter when the environment name is blank" in { + // The name is trimmed before the emptiness check, so whitespace alone must not select a venv. + val worker = newWorker(config(pveName = " ", cuid = Some(FixtureCuid))) + + chosenPythonBin(worker.underlyingActor) shouldBe PythonUtils.getPythonExecutable + } + + it should "fall back to the default interpreter when the environment has no interpreter on disk" in { + val worker = newWorker(config(pveName = "absent-env", cuid = Some(FixtureCuid))) + + chosenPythonBin(worker.underlyingActor) shouldBe PythonUtils.getPythonExecutable + } + + it should "select the virtual environment's interpreter when one exists" in { + val pveName = "present-env" + val python = venvPython(FixtureCuid, pveName) + Files.createDirectories(python.getParent) + Files.deleteIfExists(python) + Files.createFile(python) + python.toFile.setExecutable(true) + // A filesystem that ignores the executable bit (or a root user) would make the assertion + // vacuous rather than failing, so skip instead of pretending to have covered the branch. + assume(Files.isExecutable(python), "the fixture interpreter is not executable here") + val worker = newWorker(config(pveName = pveName, cuid = Some(FixtureCuid))) + + chosenPythonBin(worker.underlyingActor) shouldBe python.toAbsolutePath.normalize().toString + } + + it should "close the proxy client, stop both executors and destroy the process on shutdown" in { + val worker = newWorker(config()) + val actor = worker.underlyingActor + openPort(actor) + val process = new RecordingProcess() + setServerProcess(actor, process) + val clientExecutor = readPrivate[ExecutorService](actor, "clientThreadExecutor") + val serverExecutor = readPrivate[ExecutorService](actor, "serverThreadExecutor") + + watch(worker) + system.stop(worker) + expectTerminated(worker, 10.seconds) + + clientExecutor.isShutdown shouldBe true + serverExecutor.isShutdown shouldBe true + process.destroyed.get() shouldBe true + } Review Comment: This test claims it verifies `pythonProxyClient.close()` is called, but it only asserts the executors are shutdown and the process is destroyed. If `pythonProxyClient.close()` is removed from `postStop`, this test would still pass, so the teardown coverage is incomplete. ########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/pythonworker/PythonWorkflowWorkerSpec.scala: ########## @@ -0,0 +1,302 @@ +/* + * 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.pythonworker + +import com.twitter.util.Promise +import org.apache.commons.lang3.SystemUtils +import org.apache.pekko.actor.ActorSystem +import org.apache.pekko.testkit.{ImplicitSender, TestActorRef, TestKit} +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.common.WorkflowActor +import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{NetworkAck, NetworkMessage} +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ + AsyncRPCContext, + ControlInvocation, + EmptyRequest +} +import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage +import org.apache.texera.amber.engine.common.ambermessage.WorkflowMessage.getInMemSize +import org.apache.texera.amber.engine.architecture.scheduling.config.WorkerConfig +import org.apache.texera.amber.engine.common.AmberRuntime +import org.apache.texera.common.config.{PythonUtils, UdfConfig} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.matchers.should.Matchers + +import java.nio.file.{Files, Path, Paths} +import java.util.concurrent.ExecutorService +import java.util.concurrent.atomic.AtomicBoolean +import scala.concurrent.duration._ +import scala.sys.process.Process + +/** + * Unit tests for the Python worker actor. + * + * The worker's `initState` starts a proxy server, spawns a real Python interpreter and starts a + * Flight client, none of which belongs in a unit test and none of which is guaranteed to exist on + * a CI runner. The class offers no seam for that, but `initState` is an ordinary override, so the + * suite subclasses the worker and makes it a no-op. Everything below then runs against a real + * actor instance with nothing spawned. + * + * Two members are read by reflection because they are `private`: `choosePythonBin`, which has no + * caller other than the process-spawning path, and the executors/port promise, which the teardown + * assertions need. Reflection into `private` members is an established pattern in this module's + * specs. `portNumberPromise` in particular has to be fulfilled before anything touches + * `pythonProxyClient`: that client blocks in its own constructor on `Await.result(promise)`, so a + * worker whose proxy server never started would otherwise hang the moment `postStop` ran. + * + * Out of scope, deliberately: `startPythonProcess`, which runs a real interpreter, and + * `startProxyServer`, which retries a socket bind until it succeeds. + */ +class PythonWorkflowWorkerSpec + extends TestKit(ActorSystem("PythonWorkflowWorkerSpec", AmberRuntime.pekkoConfig)) + with ImplicitSender + with AnyFlatSpecLike + with Matchers + with BeforeAndAfterAll { + + /** A cuid no other suite writes under, so the venv fixtures cannot collide. */ + private val FixtureCuid = 987654 + private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs") + + override def afterAll(): Unit = { + deleteRecursively(VenvRoot.resolve(FixtureCuid.toString)) + TestKit.shutdownActorSystem(system) + } + + /** The worker with its process/socket startup removed; see the class comment. */ + private class InertPythonWorkflowWorker(config: WorkerConfig) + extends PythonWorkflowWorker(config) { + override def initState(): Unit = () + } + + /** Stands in for the OS process the worker would have spawned. */ + private class RecordingProcess(failOnDestroy: Boolean = false) extends Process { + val destroyed = new AtomicBoolean(false) + override def isAlive(): Boolean = !destroyed.get() + override def exitValue(): Int = 0 + override def destroy(): Unit = { + destroyed.set(true) + if (failOnDestroy) throw new IllegalStateException("destroy failed") + } + } + + private def newWorker(config: WorkerConfig): TestActorRef[InertPythonWorkflowWorker] = + TestActorRef[InertPythonWorkflowWorker](new InertPythonWorkflowWorker(config)) + + private def config(pveName: String = "", cuid: Option[Int] = None): WorkerConfig = + WorkerConfig(ActorVirtualIdentity("python-worker-spec"), pveName = pveName, cuid = cuid) + + private def readPrivate[T](actor: PythonWorkflowWorker, name: String): T = { + val method = classOf[PythonWorkflowWorker].getDeclaredMethod(name) + method.setAccessible(true) + method.invoke(actor).asInstanceOf[T] + } + + private def chosenPythonBin(actor: PythonWorkflowWorker): String = + readPrivate[String](actor, "choosePythonBin") + + /** Releases the port the proxy client's constructor blocks on. */ + private def openPort(actor: PythonWorkflowWorker, port: Int = 65535): Unit = + readPrivate[Promise[Int]](actor, "portNumberPromise").setValue(port) + + private def setServerProcess(actor: PythonWorkflowWorker, process: Process): Unit = { + val setter = + classOf[PythonWorkflowWorker].getDeclaredMethod("pythonServerProcess_$eq", classOf[Process]) + setter.setAccessible(true) + setter.invoke(actor, process) + } + + /** The interpreter PveManager resolves for (cuid, name); POSIX and Windows differ. */ + private def venvPython(cuid: Int, pveName: String): Path = { + val venv = VenvRoot.resolve(cuid.toString).resolve(pveName).resolve("pve") + if (SystemUtils.IS_OS_WINDOWS) venv.resolve("Scripts").resolve("python.exe") + else venv.resolve("bin").resolve("python") + } + + private def deleteRecursively(path: Path): Unit = { + if (Files.exists(path)) { + Files + .walk(path) + .sorted(java.util.Comparator.reverseOrder[Path]()) + .forEach(p => Files.deleteIfExists(p)) + } + } + + behavior of "PythonWorkflowWorker" + + it should "resolve the Python sources under the amber home" in { + val worker = newWorker(config()) + val path = worker.underlyingActor.pythonSrcDirectory + + // Asserted by tail rather than as an absolute string: the amber home moves with the checkout. + path.getFileName.toString shouldBe "python" + path.getParent.getFileName.toString shouldBe "main" + path.getParent.getParent.getFileName.toString shouldBe "src" + } + + it should "expose the configured R path with its surrounding whitespace removed" in { + val worker = newWorker(config()) + + worker.underlyingActor.RENVPath shouldBe UdfConfig.rPath.trim + worker.underlyingActor.RENVPath shouldBe worker.underlyingActor.RENVPath.trim + } + + it should "fall back to the default interpreter when the worker has no computing unit" in { + val worker = newWorker(config(pveName = "myenv", cuid = None)) + + chosenPythonBin(worker.underlyingActor) shouldBe PythonUtils.getPythonExecutable + } + + it should "fall back to the default interpreter when the environment name is blank" in { + // The name is trimmed before the emptiness check, so whitespace alone must not select a venv. + val worker = newWorker(config(pveName = " ", cuid = Some(FixtureCuid))) + + chosenPythonBin(worker.underlyingActor) shouldBe PythonUtils.getPythonExecutable + } + + it should "fall back to the default interpreter when the environment has no interpreter on disk" in { + val worker = newWorker(config(pveName = "absent-env", cuid = Some(FixtureCuid))) + + chosenPythonBin(worker.underlyingActor) shouldBe PythonUtils.getPythonExecutable + } + + it should "select the virtual environment's interpreter when one exists" in { + val pveName = "present-env" + val python = venvPython(FixtureCuid, pveName) + Files.createDirectories(python.getParent) + Files.deleteIfExists(python) + Files.createFile(python) + python.toFile.setExecutable(true) + // A filesystem that ignores the executable bit (or a root user) would make the assertion + // vacuous rather than failing, so skip instead of pretending to have covered the branch. + assume(Files.isExecutable(python), "the fixture interpreter is not executable here") + val worker = newWorker(config(pveName = pveName, cuid = Some(FixtureCuid))) + + chosenPythonBin(worker.underlyingActor) shouldBe python.toAbsolutePath.normalize().toString + } + + it should "close the proxy client, stop both executors and destroy the process on shutdown" in { + val worker = newWorker(config()) + val actor = worker.underlyingActor + openPort(actor) + val process = new RecordingProcess() + setServerProcess(actor, process) + val clientExecutor = readPrivate[ExecutorService](actor, "clientThreadExecutor") + val serverExecutor = readPrivate[ExecutorService](actor, "serverThreadExecutor") + + watch(worker) + system.stop(worker) + expectTerminated(worker, 10.seconds) + + clientExecutor.isShutdown shouldBe true + serverExecutor.isShutdown shouldBe true + process.destroyed.get() shouldBe true + } + + it should "not let a failing teardown step escape postStop" in { + val worker = newWorker(config()) + val actor = worker.underlyingActor + openPort(actor) + val process = new RecordingProcess(failOnDestroy = true) + setServerProcess(actor, process) + val clientExecutor = readPrivate[ExecutorService](actor, "clientThreadExecutor") + val serverExecutor = readPrivate[ExecutorService](actor, "serverThreadExecutor") + + watch(worker) + system.stop(worker) + + // The actor still terminates, and the steps that ran before the failure still took effect. + expectTerminated(worker, 10.seconds) + clientExecutor.isShutdown shouldBe true + serverExecutor.isShutdown shouldBe true + process.destroyed.get() shouldBe true + } + + it should "answer a credit request with the credit currently queued for that channel" in { + val worker = newWorker(config()) + openPort(worker.underlyingActor) + val channel = + ChannelIdentity( + ActorVirtualIdentity("upstream"), + ActorVirtualIdentity("downstream"), + isControl = true + ) + + worker ! WorkflowActor.CreditRequest(channel) + + val response = expectMsgType[WorkflowActor.CreditResponse](10.seconds) + response.channelId shouldBe channel + // Nothing has been handed to the Python side, so it is holding nothing. + response.credit shouldBe 0L + } + + it should "stay responsive after backpressure is pushed to the Python side" in { + val worker = newWorker(config()) + openPort(worker.underlyingActor) + val channel = + ChannelIdentity( + ActorVirtualIdentity("upstream"), + ActorVirtualIdentity("downstream"), + isControl = true + ) + + worker.underlyingActor.handleBackpressure(true) + worker.underlyingActor.handleBackpressure(false) + + // The command is queued for the Python process rather than answered here, so the observable + // effect is that the actor is still serving its own protocol afterwards. + worker ! WorkflowActor.CreditRequest(channel) + expectMsgType[WorkflowActor.CreditResponse](10.seconds).channelId shouldBe channel + } + + it should "hand a control payload to the Python side and acknowledge the message" in { + val worker = newWorker(config()) + openPort(worker.underlyingActor) + val channel = + ChannelIdentity( + ActorVirtualIdentity("upstream"), + ActorVirtualIdentity("python-worker-spec"), + isControl = true + ) + val payload = + ControlInvocation( + "method", + EmptyRequest(), + AsyncRPCContext(channel.fromWorkerId, channel.toWorkerId), + 0 + ) + val message = WorkflowFIFOMessage(channel, 0, payload) + + worker ! NetworkMessage(7L, message) + + val ack = expectMsgType[NetworkAck](10.seconds) + ack.messageId shouldBe 7L + ack.ackedCredit shouldBe getInMemSize(message) + // The payload is queued for the Python process, which holds no data of its own here. + ack.queuedCredit shouldBe 0L + } Review Comment: This spec only checks the `NetworkAck`, but does not verify that the `ControlInvocation` was actually forwarded/enqueued to the Python side (via `pythonProxyClient.enqueueCommand`). That means the test would still pass if the worker stopped forwarding control payloads. -- 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]
