Copilot commented on code in PR #7207:
URL: https://github.com/apache/texera/pull/7207#discussion_r3732477055
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala:
##########
@@ -49,8 +56,70 @@ final class PythonCodeRawInvalidTextSpec extends AnyFunSuite
{
private val MaxDepth: Int = 3
private val AcceptPackages: Seq[String] =
Seq("org.apache.texera.amber.operator")
+ /** Budget for one whole fanned-out pass over every descriptor. Deliberately
far
+ * above a real run (well under a second), so it only ever fires on a hang.
+ */
+ private val PassTimeout: FiniteDuration = 10.minutes
+
+ /** Runs the given work concurrently and returns the results in submission
+ * order, rethrowing the first failure so it fails the test. Sized to the
pool
+ * so the threads match the workers available to serve them.
+ */
+ private def awaitAll[T](work: Seq[() => T]): Seq[T] = {
+ val threads = Executors.newFixedThreadPool(PythonWorkerPool.maxWorkers)
+ try {
+ implicit val ec: ExecutionContext =
ExecutionContext.fromExecutorService(threads)
+ Await.result(Future.sequence(work.map(w => Future(w()))), PassTimeout)
+ } finally threads.shutdownNow()
+ }
Review Comment:
awaitAll uses a non-daemon fixed thread pool; if any task gets stuck past
PassTimeout (e.g., blocked on subprocess I/O), shutdownNow() may not terminate
the threads and the JVM can stay alive after the test has failed/timed out. Use
daemon threads (and optionally awaitTermination) so a wedged task can’t hang
the build process.
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala:
##########
@@ -249,18 +320,45 @@ final class PythonCodeRawInvalidTextSpec extends
AnyFunSuite {
val findings = checkResult.findings ++ pyCompileFindings
if (findings.isEmpty && checkResult.code.nonEmpty) {
- ok += 1
- println(s"[py-compile OK $ok/$total | checked $checked/$total]
${descriptorClass.getName}")
+ println(
+ s"[py-compile OK ${ok.incrementAndGet()}/$total | " +
+ s"checked ${checked.get()}/$total] ${descriptorClass.getName}"
+ )
}
findings
- }
+ }).flatten
- println(s"[py-compile SUMMARY] ok=$ok/$total")
+ println(s"[py-compile SUMMARY] ok=${ok.get()}/$total")
if (allFindings.nonEmpty) {
fail(PythonReflectionUtils.renderReport(allFindings, total = total))
}
}
+ /** py_compile above only parses the emitted code; running it needs the
packages
+ * it imports. Tagged, so only amber-integration — the job that installs
them —
+ * runs this. There a missing package is a defect; elsewhere it is a
local-setup
+ * fact, so cancel rather than fail.
+ */
+ test(
+ "the Python interpreter operator templates run in should import pandas and
plotly",
+ Tag(classOf[IntegrationTest].getName)
+ ) {
+ val provisioned =
sys.env.get("AMBER_TEST_FILTER").contains("integration-only")
+ def unavailable(message: String): Nothing =
+ if (provisioned) fail(message) else cancel(message)
+
+ val python = loadPythonExeFromUdfConf().getOrElse(unavailable("no runnable
python"))
+ val imported = Try {
+ val process = new ProcessBuilder(python, "-c", "import pandas, plotly")
+ .redirectErrorStream(true)
+ .start()
+ process.waitFor(60, TimeUnit.SECONDS) && process.exitValue() == 0
+ }
Review Comment:
If the import probe times out, the spawned Python process is left running
(waitFor returns false) and can leak into the rest of the test run. Destroy the
process on timeout before returning false.
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala:
##########
@@ -0,0 +1,435 @@
+/*
+ * 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.util.python
+
+import com.fasterxml.jackson.databind.node.ObjectNode
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+
+import java.io.{BufferedReader, BufferedWriter, InputStreamReader,
OutputStreamWriter}
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path, StandardCopyOption}
+import java.util.concurrent.{
+ Callable,
+ ConcurrentHashMap,
+ ExecutionException,
+ ExecutorService,
+ Executors,
+ LinkedBlockingQueue,
+ TimeUnit,
+ TimeoutException
+}
+import java.util.concurrent.atomic.AtomicInteger
+import scala.annotation.tailrec
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+/**
+ * Pools of persistent Python "worker" processes that eliminate the per-call
+ * interpreter-boot + import cost a test otherwise pays on every subprocess
+ * spawn. Testing operators one at a time does not scale when each one costs a
+ * spawn: a bare `-I -S` interpreter boots in ~25 ms, and once pandas and
plotly
+ * are imported a spawn costs ~260-310 ms — ~96% of a job whose real work is
+ * ~4 ms. A worker pays that once at startup, then serves many jobs over its
+ * lifetime, so N spawns become one.
+ *
+ * Lives in test scope here, rather than beside a single caller, because
tests in
+ * several modules run generated operator code and would otherwise each
+ * hand-roll a driver, a stdout protocol and a timeout. Other modules reach it
+ * through a `test->test` dependency on this one.
+ *
+ * Generic over the worker script — the pool never interprets the payload — so
+ * one implementation serves a syntax check, template execution and DataFrame
+ * comparison alike. Each distinct (resource, interpreterArgs, launchArgs,
+ * python, env) combination gets its own sub-pool.
+ *
+ * Protocol (line-delimited JSON, shared by all worker scripts):
+ * startup worker -> pool: {"ready": true}
+ * request pool -> worker: <caller-supplied JSON object>\n
+ * response worker -> pool: {"exit": <int>, "stdout": "...", "stderr":
"..."}\n
+ *
+ * Concurrency: callers submit from several threads at once — a spec run with
+ * ScalaTest's `-P4`, or one test fanning its cases out — so each sub-pool
holds
+ * up to [[maxWorkers]] workers, each serving one job at a time (borrow -> run
+ * -> return). A worker script may chdir per job, so a worker must never run
two
+ * jobs at once — the borrow/return discipline guarantees that.
+ *
+ * Robustness: an ordinary *job* failure comes back as an [[Outcome]] with
+ * `exit != 0` (worker keeps running). A hard interpreter crash ends a worker;
+ * the pool detects the EOF / broken pipe, discards it, and throws
+ * [[WorkerDiedException]] so the caller can fall back to a one-shot
subprocess
+ * — behavior is then never worse than the pre-pool path. A worker that stays
+ * alive but stops answering ends the same way, on the [[Timeouts]] below.
+ */
+object PythonWorkerPool extends LazyLogging {
+
+ /** Worker response: process-like exit code plus captured streams. */
+ final case class Outcome(exit: Int, stdout: String, stderr: String)
+
+ /** Thrown when a worker dies mid-job (hard crash / broken pipe). Callers
+ * catch this and fall back to a one-shot subprocess.
+ */
+ final class WorkerDiedException(message: String, cause: Throwable = null)
+ extends RuntimeException(message, cause)
+
+ /** Feature toggle. `TEXERA_TEST_PYTHON_WORKER=0` (or `false`/`off`) forces
the
+ * one-subprocess-per-call paths everywhere — an escape hatch for debugging
a
+ * suspected isolation leak. Default on.
+ */
+ val enabled: Boolean =
+ !sys.env
+ .get("TEXERA_TEST_PYTHON_WORKER")
+ .map(_.trim.toLowerCase)
+ .exists(Set("0", "false", "off"))
+
+ /** Max live workers per sub-pool. Defaults to 4 to match ScalaTest's `-P4`,
so
+ * the two concurrency bounds agree on how many interpreters may be live;
+ * override via `TEXERA_TEST_PYTHON_WORKERS`. Public so a caller fanning out
+ * jobs within one test can size that fan-out to the workers it will get.
+ */
+ val maxWorkers: Int =
+ sys.env
+ .get("TEXERA_TEST_PYTHON_WORKERS")
+ .flatMap(s => scala.util.Try(s.trim.toInt).toOption)
+ .filter(_ > 0)
+ .getOrElse(4)
+
+ /** How long a caller waits on a worker before the pool kills and discards
it.
+ * A read on a process pipe cannot be interrupted — a suite or executor
timeout
+ * leaves the reading thread stuck on it — so a worker that stays alive
without
+ * answering has to be bounded here. `response` keeps the 30 seconds the
+ * one-shot spawn this pool replaced allowed a job; `startup` is longer
because
+ * a worker imports its libraries before it reports ready, and a loaded CI
+ * machine makes that slow. Override in seconds via
+ * `TEXERA_TEST_PYTHON_WORKER_TIMEOUT` /
`TEXERA_TEST_PYTHON_WORKER_STARTUP_TIMEOUT`.
+ */
+ final case class Timeouts(responseMillis: Long, startupMillis: Long)
+
+ object Timeouts {
+ private def envSeconds(name: String, default: Long): Long =
+ sys.env
+ .get(name)
+ .flatMap(s => scala.util.Try(s.trim.toLong).toOption)
+ .filter(_ > 0)
+ .getOrElse(default) * 1000
+
+ val Default: Timeouts = Timeouts(
+ responseMillis = envSeconds("TEXERA_TEST_PYTHON_WORKER_TIMEOUT", 30),
+ startupMillis = envSeconds("TEXERA_TEST_PYTHON_WORKER_STARTUP_TIMEOUT",
60)
+ )
+ }
+
+ /**
+ * Run one job through a pooled worker for `resourcePath`, launched as
+ * `pythonExe <interpreterArgs> <script> <launchArgs>` with extra
environment
+ * `env`. `request` is the worker-specific JSON payload (the pool does not
+ * interpret it). Throws [[WorkerDiedException]] on a hard worker crash.
+ *
+ * `interpreterArgs` are the flags that must precede the script — a syntax
+ * checker wants `-I -S` so it validates under the same isolation a one-shot
+ * `python -I -S -m py_compile` gave it. `launchArgs` are the script's own
+ * (e.g. `--serve`), and `env` carries what a flag cannot (e.g. PYTHONPATH).
+ * All three are part of a worker's identity: one started differently is not
+ * interchangeable, so it gets its own sub-pool. `timeouts` is not — it
bounds
+ * this call, so a caller whose jobs are slower than most can raise it
without
+ * splitting the pool.
+ */
+ def run(
+ resourcePath: String,
+ launchArgs: Seq[String],
+ pythonExe: String,
+ request: ObjectNode,
+ env: Map[String, String] = Map.empty,
+ interpreterArgs: Seq[String] = Seq.empty,
+ timeouts: Timeouts = Timeouts.Default
+ ): Outcome = {
+ val envKey = env.toSeq.sorted.map { case (k, v) => s"$k=$v" }
+ val key =
+ (resourcePath +: pythonExe +: (interpreterArgs ++ launchArgs ++
envKey)).mkString(" ")
Review Comment:
Pool identity keys are built by joining parts with a single space. If any
part contains spaces (pythonExe path, env values like PYTHONPATH, etc.),
distinct configurations can collide and incorrectly share a worker pool. Use an
unambiguous delimiter (or a structured key) to avoid collisions.
--
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]
