You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/14 09:41:14 UTC
(texera) branch gh-readonly-queue/main/pr-7207-c37f6ab7c68f15e7a1aa283d0c29b84e11f1ed6d created (now 2f77d5ab93)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7207-c37f6ab7c68f15e7a1aa283d0c29b84e11f1ed6d
in repository https://gitbox.apache.org/repos/asf/texera.git
at 2f77d5ab93 test(workflow-operator): route Python tests to amber-integration (#7207)
This branch includes the following new commits:
new 2f77d5ab93 test(workflow-operator): route Python tests to amber-integration (#7207)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: test(workflow-operator): route Python tests to amber-integration (#7207)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7207-c37f6ab7c68f15e7a1aa283d0c29b84e11f1ed6d
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 2f77d5ab93f94c8e9b33195f805a35e63c52f970
Author: Kary Zheng <15...@users.noreply.github.com>
AuthorDate: Fri Aug 14 09:40:52 2026 +0000
test(workflow-operator): route Python tests to amber-integration (#7207)
### What changes were proposed in this PR?
The `amber` job runs the `WorkflowOperator` test suite and installs no
Python packages. A test that *executes* an operator's generated
template, rather than only `py_compile`-ing it, therefore fails its
dependency probe and cancels — and a cancellation is neither a pass nor
a failure, so the suite still reports "All tests passed" and the gap
leaves no trace in the build.
This gives `WorkflowOperator` the split `amber` already has:
- **Tag** — `IntegrationTest` under `common/workflow-operator/src/test`.
amber's cannot be reused: it lives in `amber/src/test/integration`, and
amber depends on this module rather than the reverse.
- **Filter** — `common/workflow-operator/build.sbt` reads the same
`AMBER_TEST_FILTER`. The `amber` job already sets it on the step that
invokes `WorkflowOperator/jacoco`, so no workflow change is needed for
the exclusion to take effect.
- **Job** — `"WorkflowOperator/test"` added to `amber-integration`,
which already runs `integration-only` with `amber/requirements.txt` and
`amber/operator-requirements.txt` installed (pandas 2.2.3, plotly
5.24.1).
- **Shared logic** — the env-var-to-ScalaTest-args mapping moves to
`project/TestFilters.scala`, beside the build's other shared helpers
(`AddMetaInfLicenseFiles.scala`, `JdkOptions.scala`); each module passes
its own env var and tag, the two a caller has to match to the workflow
and to the annotation. One behavior changes, and amber inherits it: a
value that is neither of the two now fails at project load, where the
inline code fell through to running everything. Nothing sets a third
value, so what this catches is a typo that would otherwise leave a job
running the whole suite while reporting the subset it asked for.
- **First user** — `PythonCodeRawInvalidTextSpec` gains a tagged case
asserting that pandas and plotly import in the interpreter it already
resolves: `py_compile` only parses the emitted code, while running it
needs the packages it imports. Being tagged, it also exercises the
routing, and it turns a missing install in `amber-integration` into a
failure rather than the silent cancellation above. Elsewhere a bare
interpreter is a local-setup fact, so it cancels instead.
The second half addresses the scaling problem in the same issue: testing
operators one at a time does not scale when each one costs a spawn.
- **Batching** — the `py_compile` check spawned `python -I -S -B -m
py_compile` once per `PythonOperatorDescriptor`, 117 of them serially,
where the interpreter boot is the entire cost and the compile is under a
millisecond. It now goes through `PythonWorkerPool`: a worker launched
once with the same `-I -S` isolation, serving many sources over a
line-delimited JSON protocol. What the check accepts is unchanged —
`compile(source, path, "exec")` is what `py_compile` does before writing
a `.pyc`, and raises the same SyntaxError — and any worker the pool
cannot give out or keep falls back to the spawn, so behavior is never
worse than before: one that would not start, would not report ready, or
died mid-job all leave through the same exception, and the check counts
the descriptors that took the spawn into its summary so a run the pool
served none of does not read as a green pooled run.
`TEXERA_TEST_PYTHON_WORKER=0` selects the old path outright.
- **Parallelism** — the descriptors are fanned out across the pool's
workers, four at a time; the fan-out's executor is sized to the pool's
cap, so nothing runs past it. Together: 1119 ms to 310 ms. That cap is
per sub-pool, keyed by script, arguments and environment, and is
env-overridable, so it is the fan-out's own sizing that bounds this —
not a ScalaTest parameter, which no suite in the module currently
parallelizes under anyway.
- **Where the pool lives** — this module's test scope, not beside its
caller. The tests this PR unblocks execute generated templates against
pandas and plotly, where a spawn costs 260-310 ms, mostly the imports,
dwarfing the ~4 ms a job computes for. Without a shared seam each such
test hand-rolls a driver, a stdout protocol, a timeout and an
interpreter probe — the runtime test in #7149 already does. Other
modules reach it through a `test->test` dependency.
### Any related issues, documentation, discussions?
Fixes #7186. Found while reviewing #7149, whose runtime test is the case
cancelling today; tagging it is a one-line follow-up once this is in,
and converting its hand-rolled driver to the pool is the natural second
one. The pooled-worker design, and the measurements behind those
figures, are in #6975.
### How was this PR tested?
- `amber` is unchanged by the extraction: `show
WorkflowExecutionService/Test/testOptions` under all three env values
gives arguments byte-identical to its previous inline code — `-l <tag>`,
`-n <tag>`, nothing when unset. `WorkflowOperator` was checked the same
way and yields its own tag in those positions.
- The new case, run as `WorkflowOperator/testOnly
*PythonCodeRawInvalidTextSpec`: passes under `integration-only` (1 test
selected); is excluded under `skip-integration` while the spec's two
existing tests still run; and, pointed at a bare interpreter, fails
under `integration-only` and cancels with no filter set.
- The pooled check agrees with the spawn it replaces: 117/117
descriptors pass either way, and `TEXERA_TEST_PYTHON_WORKER=0` flips
between them. Timed with `-oD`: 1119 ms spawning, 310 ms pooled.
- Red-checked, since a green suite only proves the two paths agree on
valid code: appending a syntax error to every generated source takes the
pooled path to `ok=0/117` and fails the test, so it still detects what
it is there to detect.
- Each way a worker could escape the pool's borrow/return discipline was
reproduced before being fixed, and `PythonWorkerPoolSpec` keeps a case
on each: an interrupted caller used to leave a worker neither returned
nor discarded, and the case that refills every slot and then asks for
that many jobs again spins at the cap until its bound without the fix; a
first line that is not the protocol used to raise `JsonParseException`
past the caller's fallback, and the fixture now has a mode that writes
one; an interpreter that cannot be started arrives as the pool's own
exception rather than `ProcessBuilder`'s.
- The fallback is never worse than the spawn it replaces, checked end to
end: with the pooled path pointed at an interpreter that does not exist,
all 117 descriptors fall back, all 117 still pass, and the summary
reports 117 fallbacks.
- `WorkflowOperator/test` under `skip-integration`: 286 suites, 2163
tests, none failed.
- `sbt scalafmtCheckAll` is clean, and `WorkflowOperator`'s `Compile`
and `Test` scalafix checks pass.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
---------
Signed-off-by: Kary Zheng <15...@users.noreply.github.com>
Co-authored-by: Claude Opus 5 (1M context) <no...@anthropic.com>
Co-authored-by: Xinyuan Lin <xi...@uci.edu>
Co-authored-by: Yicong Huang <17...@users.noreply.github.com>
---
.github/workflows/build.yml | 13 +-
amber/build.sbt | 20 +-
common/workflow-operator/build.sbt | 13 +
.../amber/operator/tags/IntegrationTest.java | 51 +++
.../src/test/resources/python/hanging_worker.py | 87 ++++
.../src/test/resources/python/py_compile_worker.py | 77 ++++
.../amber/util/PythonCodeRawInvalidTextSpec.scala | 161 ++++++-
.../amber/util/python/PythonWorkerPool.scala | 503 +++++++++++++++++++++
.../amber/util/python/PythonWorkerPoolSpec.scala | 266 +++++++++++
project/TestFilters.scala | 48 ++
10 files changed, 1212 insertions(+), 27 deletions(-)
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index c28457b198..120dd1f2aa 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -284,9 +284,13 @@ jobs:
# Test config (sbt's `test` task does not transit dependsOn),
# so common modules' tests are listed explicitly here.
#
- # AMBER_TEST_FILTER=skip-integration tells amber/build.sbt to
- # exclude @org.apache.texera.amber.tags.IntegrationTest specs;
- # those run in the amber-integration job below.
+ # AMBER_TEST_FILTER=skip-integration excludes the integration-tagged
+ # tests of two modules, each under its own tag: amber/build.sbt drops
+ # @org.apache.texera.amber.tags.IntegrationTest, and
+ # common/workflow-operator/build.sbt drops
+ # @org.apache.texera.amber.operator.tags.IntegrationTest. Both run in the
+ # amber-integration job below. The second exclusion is what keeps the
+ # pandas and plotly probe out of this job, which installs neither.
env:
AMBER_TEST_FILTER: skip-integration
# unit job uses its provisioned postgres catalog; default (rest) needs a Lakekeeper not run here
@@ -631,6 +635,8 @@ jobs:
# specs. The Java @TagAnnotation makes the marker visible to
# ScalaTest's reflection, so `-n TAG` correctly narrows the
# run.
+ # WorkflowOperator/test is here for the same reason, with its own tag:
+ # its Python-dependent tests need packages the `amber` job lacks.
#
# scalafmtCheckAll + scalafixAll --check are run here as well
# because an integration-only PR fires only the
@@ -657,6 +663,7 @@ jobs:
run: |
sbt scalafmtCheckAll \
"scalafixAll --check" \
+ "WorkflowOperator/test" \
"WorkflowExecutionService/test"
- name: Build amber dist for the texera-web boot test
# Boot smoke test for texera-web (mirrors platform-integration for the
diff --git a/amber/build.sbt b/amber/build.sbt
index 06d1949692..b1f8a0d481 100644
--- a/amber/build.sbt
+++ b/amber/build.sbt
@@ -69,19 +69,13 @@ Test / unmanagedSourceDirectories += baseDirectory.value / "src" / "test" / "int
// scalafix still cover it and `sbt Test/runMain` can invoke benches.
Test / unmanagedSourceDirectories += baseDirectory.value / "src" / "bench" / "scala"
-// Test-filter switch driven by the AMBER_TEST_FILTER env var so the
-// amber and amber-integration CI jobs select disjoint subsets without
-// each invocation having to embed a `set Tests.Argument(...)` prefix.
-// skip-integration : exclude @IntegrationTest-tagged specs (amber job)
-// integration-only : include only @IntegrationTest-tagged specs (amber-integration job)
-// (unset) : run everything (default for local sbt)
-Test / testOptions ++= (sys.env.get("AMBER_TEST_FILTER") match {
- case Some("skip-integration") =>
- Seq(Tests.Argument(TestFrameworks.ScalaTest, "-l", "org.apache.texera.amber.tags.IntegrationTest"))
- case Some("integration-only") =>
- Seq(Tests.Argument(TestFrameworks.ScalaTest, "-n", "org.apache.texera.amber.tags.IntegrationTest"))
- case _ => Nil
-})
+// Lets the amber and amber-integration CI jobs select disjoint subsets without
+// each invocation having to embed a `set Tests.Argument(...)` prefix. See
+// project/TestFilters.scala.
+Test / testOptions ++= TestFilters.integrationSplit(
+ envVar = "AMBER_TEST_FILTER",
+ tag = "org.apache.texera.amber.tags.IntegrationTest"
+)
// Excluding some proto files:
PB.generate / excludeFilter := "scalapb.proto"
diff --git a/common/workflow-operator/build.sbt b/common/workflow-operator/build.sbt
index d1cd1d4cc3..2bb41754a5 100644
--- a/common/workflow-operator/build.sbt
+++ b/common/workflow-operator/build.sbt
@@ -35,6 +35,19 @@ ThisBuild / conflictManager := ConflictManager.latestRevision
// Restrict parallel execution of tests to avoid conflicts
Global / concurrentRestrictions += Tags.limit(Tags.Test, 1)
+// A test needing more than a bare Python interpreter is tagged, so the amber job
+// excludes it, and amber-integration, which installs amber's requirements files,
+// runs it. The amber job already sets this env var on the step that invokes
+// WorkflowOperator/jacoco, so no workflow change is needed for the exclusion.
+//
+// PythonCodeRawInvalidTextSpec reads the same env var and value directly, to tell
+// a missing package in amber-integration (a defect) from one on a developer's
+// machine (a local-setup fact). Changing the variable or the value here without
+// changing it there leaves that test cancelling in the job meant to fail it.
+Test / testOptions ++= TestFilters.integrationSplit(
+ envVar = "AMBER_TEST_FILTER",
+ tag = "org.apache.texera.amber.operator.tags.IntegrationTest"
+)
/////////////////////////////////////////////////////////////////////////////
// Compiler Options
diff --git a/common/workflow-operator/src/test/java/org/apache/texera/amber/operator/tags/IntegrationTest.java b/common/workflow-operator/src/test/java/org/apache/texera/amber/operator/tags/IntegrationTest.java
new file mode 100644
index 0000000000..1738a0e8ab
--- /dev/null
+++ b/common/workflow-operator/src/test/java/org/apache/texera/amber/operator/tags/IntegrationTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.operator.tags;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.scalatest.TagAnnotation;
+
+/**
+ * Marks a test in this module as needing more than a bare Python interpreter —
+ * pandas or plotly, which the {@code amber} job does not install. See the
+ * AMBER_TEST_FILTER block in {@code common/workflow-operator/build.sbt} for how
+ * it routes to {@code amber-integration}.
+ *
+ * <p>Apply it to a whole spec as an annotation, or — so that a spec's cheaper
+ * assertions stay in the unit job and its coverage report — to a single case:
+ * {@code test(name, Tag(classOf[IntegrationTest].getName))} in a FunSuite,
+ * {@code taggedAs} in a FlatSpec.
+ *
+ * <p>amber's own tag lives in {@code amber/src/test/integration} and is not
+ * reachable here, since amber depends on this module rather than the reverse.
+ *
+ * <p>Java, not Scala: ScalaTest finds tag annotations by
+ * {@code java.lang.annotation} reflection, and a Scala {@code StaticAnnotation}
+ * produces no JVM annotation interface for {@code @TagAnnotation} to mark.
+ */
+@TagAnnotation
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD, ElementType.TYPE})
+public @interface IntegrationTest {
+}
diff --git a/common/workflow-operator/src/test/resources/python/hanging_worker.py b/common/workflow-operator/src/test/resources/python/hanging_worker.py
new file mode 100644
index 0000000000..95bcdb9aeb
--- /dev/null
+++ b/common/workflow-operator/src/test/resources/python/hanging_worker.py
@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+#
+# 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.
+"""
+Test fixture: a worker that stays alive and stops talking, the failure a pool
+timeout exists for. A crash is not a substitute — that closes the pipe and the
+read returns on its own.
+
+ --hang-before-ready never announce readiness
+ --deaf announce readiness, then never read stdin at all, which
+ blocks the parent's write once the request outgrows the
+ pipe buffer
+ --babble announce a line that is not the protocol at all
+ (default) announce readiness, then answer every request except one
+ asking to hang: {"hang": true} is read and never answered.
+ {"drop-exit": true} is answered without an exit code
+
+Whether a request is answered is a property of the request, not of a flag,
+because the pool keys a sub-pool by script, args and env: a test that needs a
+hung worker and a healthy job in the *same* sub-pool cannot get them from two
+different launches.
+"""
+from __future__ import annotations
+
+import json
+import sys
+import time
+
+
+def _sleep_forever() -> None:
+ # Outlives any timeout a test sets, and the pool kills this process, so the
+ # sleep is what makes the worker unresponsive rather than slow.
+ while True:
+ time.sleep(3600)
+
+
+def main() -> None:
+ if "--hang-before-ready" in sys.argv:
+ _sleep_forever()
+
+ if "--babble" in sys.argv:
+ # Then stay alive: it is the parent that has to end this process, since
+ # nothing else would reap a worker rejected before it joined the pool.
+ sys.stdout.write("not a protocol line\n")
+ sys.stdout.flush()
+ _sleep_forever()
+
+ sys.stdout.write(json.dumps({"ready": True}) + "\n")
+ sys.stdout.flush()
+
+ if "--deaf" in sys.argv:
+ _sleep_forever()
+
+ for line in sys.stdin:
+ try:
+ request = json.loads(line)
+ except ValueError:
+ request = {}
+ if request.get("hang"):
+ _sleep_forever()
+ if request.get("drop-exit"):
+ # Parses as the protocol, but leaves out the one field the parent
+ # cannot supply for itself.
+ sys.stdout.write(json.dumps({"stdout": "", "stderr": ""}) + "\n")
+ sys.stdout.flush()
+ continue
+ sys.stdout.write(json.dumps({"exit": 0, "stdout": "", "stderr": ""}) + "\n")
+ sys.stdout.flush()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/common/workflow-operator/src/test/resources/python/py_compile_worker.py b/common/workflow-operator/src/test/resources/python/py_compile_worker.py
new file mode 100644
index 0000000000..f7f143b17d
--- /dev/null
+++ b/common/workflow-operator/src/test/resources/python/py_compile_worker.py
@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+#
+# 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.
+"""
+Persistent worker that syntax-checks generated operator code, replacing one
+`python -I -S -B -m py_compile <file>` spawn per operator descriptor.
+
+`compile(source, path, "exec", dont_inherit=True)` is what `py_compile` does
+before writing a `.pyc` and raises the same SyntaxError; skipping that write is
+why neither `-B` nor a temp file is needed here. `dont_inherit` is the half a
+pooled worker cannot leave out: without it this module's own `__future__`
+import applies to every source it checks, which on CPython 3.12 rejects a
+walrus inside an annotation that the spawn accepts.
+
+Protocol (line-delimited JSON, both directions):
+
+ startup worker -> parent: {"ready": true}
+ request parent -> worker: {"source": "<code>", "name": "<label>"}\n
+ response worker -> parent: {"exit": 0, "stdout": "...", "stderr": "..."}\n
+
+`exit` is 1 when the source does not compile, with the SyntaxError in `stderr`,
+mirroring the spawn it replaces so the parent's reporting is unchanged. That
+does not end the worker; only a hard interpreter crash does.
+"""
+from __future__ import annotations
+
+import json
+import sys
+import traceback
+
+
+def _compile_one(source: str, name: str) -> "dict[str, object]":
+ """Compile one generated module. `name` is the filename the traceback shows,
+ so a report names the descriptor rather than a temp path.
+ """
+ try:
+ compile(source, name, "exec", dont_inherit=True)
+ return {"exit": 0, "stdout": "", "stderr": ""}
+ except (SyntaxError, ValueError):
+ # ValueError: sources compile() rejects outright, e.g. an embedded NUL.
+ return {"exit": 1, "stdout": "", "stderr": traceback.format_exc()}
+
+
+def main() -> None:
+ sys.stdout.write(json.dumps({"ready": True}) + "\n")
+ sys.stdout.flush()
+
+ for line in sys.stdin:
+ line = line.strip()
+ if not line:
+ continue
+ try:
+ req = json.loads(line)
+ result = _compile_one(req["source"], req.get("name", "<generated>"))
+ except Exception: # malformed request — report, keep serving
+ result = {"exit": 1, "stdout": "", "stderr": traceback.format_exc()}
+ sys.stdout.write(json.dumps(result) + "\n")
+ sys.stdout.flush()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala
index 122b1dbae8..41b58c76f5 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala
@@ -21,15 +21,23 @@ package org.apache.texera.amber.util
import com.typesafe.config.ConfigFactory
import org.apache.texera.amber.operator.PythonOperatorDescriptor
+import org.apache.texera.amber.operator.tags.IntegrationTest
import org.apache.texera.amber.pybuilder.PythonReflectionTextUtils.truncateBlock
import org.apache.texera.amber.pybuilder.PythonReflectionUtils
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.apache.texera.amber.util.python.PythonWorkerPool
+import org.scalatest.Tag
import org.scalatest.funsuite.AnyFunSuite
import java.nio.charset.StandardCharsets
import java.nio.file.Files
import java.util.concurrent
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.{Executors, TimeUnit}
+import scala.concurrent.duration.{DurationInt, FiniteDuration}
+import scala.concurrent.{Await, ExecutionContext, Future}
import scala.util.Try
+import scala.util.control.NonFatal
/**
* Regression tests for validation pipeline used for PythonOperatorDescriptor codegen.
@@ -49,8 +57,99 @@ 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.
+ *
+ * Daemon threads: a task parked on a subprocess pipe answers no interrupt, so
+ * `shutdownNow` need not end it, and a non-daemon one left there would hold the
+ * JVM — and the build — open after [[PassTimeout]] has already failed the test.
+ */
+ private def awaitAll[T](work: Seq[() => T]): Seq[T] = {
+ val threads = Executors.newFixedThreadPool(
+ PythonWorkerPool.maxWorkers,
+ (r: Runnable) => {
+ val t = new Thread(r, "py-compile-check")
+ t.setDaemon(true)
+ t
+ }
+ )
+ try {
+ implicit val ec: ExecutionContext = ExecutionContext.fromExecutorService(threads)
+ Await.result(Future.sequence(work.map(w => Future(w()))), PassTimeout)
+ } finally threads.shutdownNow()
+ }
+
+ /** Count of checks the pooled path could not serve. Not a failure — the spawn each one
+ * fell back to is the pre-pool behavior — but a number the summary has to
+ * carry, since a check passing says nothing about which path answered it.
+ */
+ private val spawnFallbacks = new AtomicInteger(0)
+
+ /** Syntax-checks one generated module, through a pooled worker when available.
+ *
+ * The worker is launched with the same `-I -S` isolation the one-shot path
+ * uses, so what the check accepts is unchanged; it just stops paying an
+ * interpreter boot — the whole cost of a check whose real work is under a
+ * millisecond — once per descriptor. A worker the pool cannot give out, or
+ * loses mid-job, falls back to the spawn, so behavior is never worse than
+ * before the pool.
+ */
+ private def syntaxCheck(
+ pythonExecutable: String,
+ pythonSource: String,
+ descriptorName: String
+ ): Either[String, Unit] = {
+ def viaPool: Either[String, Unit] = {
+ val request = objectMapper.createObjectNode()
+ request.put("source", pythonSource)
+ request.put("name", s"$descriptorName.py")
+ val outcome = PythonWorkerPool.run(
+ resourcePath = "/python/py_compile_worker.py",
+ launchArgs = Seq.empty,
+ pythonExe = pythonExecutable,
+ request = request,
+ interpreterArgs = Seq("-I", "-S")
+ )
+ if (outcome.exit == 0) Right(())
+ else {
+ val output = if (outcome.stderr.trim.nonEmpty) outcome.stderr.trim else "(no output)"
+ Left(
+ s"py_compile failed (exit=${outcome.exit})\nOutput:\n" +
+ truncateBlock(output, maxLines = 40, maxChars = 8000)
+ )
+ }
+ }
+
+ if (PythonWorkerPool.enabled) {
+ try viaPool
+ catch {
+ // Anything the pooled path throws leaves the spawn as the answer, which is
+ // what makes it never worse than before: not only a worker that died
+ // mid-job, but equally one the pool could not hand out at all. Those
+ // arrive as WorkerDiedException; NonFatal also covers the steps outside
+ // that contract, such as materializing the worker script. Counted, so a
+ // run the pool served none of does not read as a green pooled run.
+ case NonFatal(thrown) =>
+ println(
+ s"[py-compile FALLBACK ${spawnFallbacks.incrementAndGet()}] $descriptorName: " +
+ s"pooled worker unavailable, spawning instead: " +
+ truncateBlock(thrown.toString, maxLines = 3, maxChars = 500)
+ )
+ pyCompile(pythonExecutable, pythonSource)
+ }
+ } else pyCompile(pythonExecutable, pythonSource)
+ }
+
/**
* Runs `python -m py_compile` on the provided source, using an isolated interpreter invocation.
+ * Retained as the pooled path's fallback and as the behavior selected by
+ * TEXERA_TEST_PYTHON_WORKER=0.
*
* Isolation flags:
* - -I : isolate (ignore user site-packages / env)
@@ -225,21 +324,23 @@ final class PythonCodeRawInvalidTextSpec extends AnyFunSuite {
}
val total = descriptorCandidates.size
- var ok = 0
- var checked = 0
-
- val allFindings = descriptorCandidates.flatMap { descriptorClass =>
- checked += 1
+ val ok = new AtomicInteger(0)
+ val checked = new AtomicInteger(0)
+ // Checked concurrently: the fan-out is what turns the pool's workers into
+ // parallel interpreters rather than a queue in front of one. The executor is
+ // sized to maxWorkers, so nothing runs past the cap.
+ val allFindings = awaitAll(descriptorCandidates.map { descriptorClass => () =>
val checkResult =
PythonReflectionUtils.checkDescriptorWithCode(
descriptorClass,
rawInvalidText = RawInvalid,
maxDepth = MaxDepth
)
+ checked.incrementAndGet()
val pyCompileFindings = checkResult.code.toSeq.flatMap { generatedCode =>
- pyCompile(pythonExecutable, generatedCode) match {
+ syntaxCheck(pythonExecutable, generatedCode, descriptorClass.getSimpleName) match {
case Left(errorMessage) =>
Seq(PythonReflectionUtils.Finding(descriptorClass.getName, "py-compile", errorMessage))
case Right(()) => Nil
@@ -249,18 +350,56 @@ 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, spawn fallbacks=${spawnFallbacks.get()}"
+ )
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)
+ ) {
+ // Same env var and value the build reads to select this subset, in
+ // common/workflow-operator/build.sbt; keep the two in step. Nothing enforces
+ // that from here, since TestFilters is build-scope and cannot be imported: if
+ // the selector is renamed and this string is not, the test keeps running in
+ // amber-integration but cancels instead of failing, which is the non-result
+ // the tag exists to remove.
+ 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()
+ // Killed on the way out: a probe that ran out of time is still running, and
+ // would otherwise leak into the rest of the run.
+ if (process.waitFor(60, TimeUnit.SECONDS)) process.exitValue() == 0
+ else { process.destroyForcibly(); false }
+ }
+ if (!imported.getOrElse(false)) {
+ unavailable(s"'$python' cannot import pandas and plotly")
+ }
+ }
+
}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala
new file mode 100644
index 0000000000..414ad40ebb
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala
@@ -0,0 +1,503 @@
+/*
+ * 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 — mostly the imports — dwarfing
+ * the ~4 ms of real work a job does. 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 — one test fanning
+ * its cases out, or suites running in parallel — 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 for a worker the pool could not give out or keep: one that would not
+ * start, one that never signalled ready, or one that died or fell silent
+ * mid-job. 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, so callers on distinct worker
+ * scripts add up rather than share this bound. Defaults to 4, 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. `responseMillis` keeps the 30 seconds the
+ * one-shot spawn this pool replaced allowed a job; `startupMillis` 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]] for every worker the pool could
+ * not give out or keep — one that would not start, would not report ready, or
+ * died mid-job — so one `catch` covers a caller's whole fallback.
+ *
+ * `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. `--deaf`), and `env` carries what a flag cannot — though not
+ * PYTHONPATH or any other PYTHON* var, which the `-I` above makes CPython ignore.
+ * 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 pool = pools.computeIfAbsent(
+ Key(resourcePath, pythonExe, interpreterArgs.toList, launchArgs.toList, env.toList.sorted),
+ _ => new Pool(resourcePath, launchArgs, pythonExe, env, interpreterArgs)
+ )
+ pool.run(request, timeouts)
+ }
+
+ /** What makes two launches the same worker. Compared field by field rather than
+ * as one joined string, so a value carrying whatever the separator was — a
+ * python path with a space in it, a PYTHONPATH — cannot make two different
+ * launches share a pool.
+ */
+ private final case class Key(
+ resourcePath: String,
+ pythonExe: String,
+ interpreterArgs: List[String],
+ launchArgs: List[String],
+ env: List[(String, String)]
+ )
+
+ /** How long a caller at the worker cap waits before re-examining it. Not a
+ * deadline — see [[Pool.borrow]].
+ */
+ private val CapRecheckMillis: Long = 250
+
+ private val pools = new ConcurrentHashMap[Key, Pool]()
+
+ Runtime.getRuntime.addShutdownHook(new Thread(() => shutdownAll()))
+
+ private def shutdownAll(): Unit =
+ pools.values().forEach(_.shutdown())
+
+ // A single sub-pool: up to `maxWorkers` live workers for one worker script.
+ private final class Pool(
+ resourcePath: String,
+ launchArgs: Seq[String],
+ pythonExe: String,
+ env: Map[String, String],
+ interpreterArgs: Seq[String]
+ ) {
+ private val idle = new LinkedBlockingQueue[Worker]()
+ private val liveCount = new AtomicInteger(0)
+ private val all = mutable.Set.empty[Worker] // guarded by `all`
+ @volatile private var script: Path = _
+
+ def run(request: ObjectNode, timeouts: Timeouts): Outcome = {
+ val w = borrow(timeouts)
+ try {
+ val outcome = w.run(request, timeouts.responseMillis)
+ idle.offer(w) // healthy — return to pool
+ outcome
+ } catch {
+ // Every throw, not only a WorkerDiedException: an interrupt on this
+ // thread — what an executor's `shutdownNow` sends — leaves the blocking
+ // read or write as an InterruptedException, which `NonFatal` excludes and
+ // [[Worker.run]] therefore does not wrap. A worker left neither returned
+ // nor discarded costs this sub-pool a slot for the life of the JVM, and
+ // its answer may still be in flight, so it cannot be reused either way.
+ case e: Throwable =>
+ discard(w)
+ throw e
+ }
+ }
+
+ @tailrec
+ private def borrow(timeouts: Timeouts): Worker = {
+ val existing = idle.poll()
+ if (existing != null) existing
+ else if (liveCount.getAndIncrement() < maxWorkers) {
+ try create(timeouts)
+ catch {
+ case e: Throwable =>
+ liveCount.decrementAndGet()
+ throw e
+ }
+ } else {
+ liveCount.decrementAndGet()
+ // At the cap. Waiting outright for a returned worker would strand this
+ // caller when the ones ahead are discarded instead: a discard frees a
+ // slot without putting anything back. So wait only briefly, then look at
+ // the cap again — the next pass starts a replacement. A long queue still
+ // waits as long as it takes; that is the caller's own backlog, not a hang.
+ val returned = idle.poll(CapRecheckMillis, TimeUnit.MILLISECONDS)
+ if (returned != null) returned else borrow(timeouts)
+ }
+ }
+
+ private def create(timeouts: Timeouts): Worker = {
+ val cmd =
+ (((pythonExe +: interpreterArgs) :+ ensureScript().toString) ++ launchArgs).asJava
+ val pb = new ProcessBuilder(cmd).redirectErrorStream(false)
+ env.foreach { case (k, v) => pb.environment().put(k, v) }
+ // A worker that cannot be started is a worker death like any other, so it
+ // leaves through the same exception: `start` throws a bare IOException — no
+ // interpreter at that path, or the OS out of processes under a fan-out — and
+ // a caller's fallback is written against [[WorkerDiedException]].
+ val process =
+ try pb.start()
+ catch {
+ case NonFatal(e) =>
+ throw new WorkerDiedException(
+ s"could not start python worker for $resourcePath: ${e.getMessage}",
+ e
+ )
+ }
+ val w = new Worker(process, s"$resourcePath ${launchArgs.mkString(" ")}".trim)
+ // Anything thrown before the worker joins `all` — a startup that timed out,
+ // an interrupt on this thread — leaves an interpreter nothing else would
+ // reap: not the caller, which never gets the handle, and not the shutdown
+ // hook, which walks `all`. `destroy` is idempotent, so a worker that already
+ // killed itself on the way out is no exception to that.
+ try w.awaitReady(timeouts.startupMillis)
+ catch {
+ case e: Throwable =>
+ w.destroy()
+ throw e
+ }
+ all.synchronized(all.add(w))
+ logger.debug(s"Started python worker for $resourcePath (live=${liveCount.get}/$maxWorkers)")
+ w
+ }
+
+ private def discard(w: Worker): Unit = {
+ all.synchronized(all.remove(w))
+ liveCount.decrementAndGet()
+ w.destroy()
+ }
+
+ private def ensureScript(): Path = {
+ if (script == null) synchronized {
+ if (script == null) {
+ val stream = getClass.getResourceAsStream(resourcePath)
+ require(stream != null, s"worker script not found on classpath at $resourcePath")
+ try {
+ val tmp = Files.createTempFile("py-worker-", ".py")
+ Files.copy(stream, tmp, StandardCopyOption.REPLACE_EXISTING)
+ tmp.toFile.deleteOnExit()
+ script = tmp
+ } finally stream.close()
+ }
+ }
+ script
+ }
+
+ def shutdown(): Unit =
+ all.synchronized {
+ all.foreach(_.destroy())
+ all.clear()
+ }
+ }
+
+ // One live worker process plus its framed-JSON stdin and background drains of
+ // its stdout (the protocol) and stderr (only non-empty on a hard crash).
+ private final class Worker(process: Process, label: String) {
+ private val stdin: BufferedWriter =
+ new BufferedWriter(new OutputStreamWriter(process.getOutputStream, StandardCharsets.UTF_8))
+ private val errBuf = new StringBuilder
+
+ // Protocol lines the worker has written, `None` marking end of stream. A
+ // dedicated thread owns the blocking read so a caller can wait with a
+ // timeout: `readLine` on a process pipe answers neither an interrupt nor a
+ // deadline, and only closing the pipe — killing the process — releases it.
+ private val lines = new LinkedBlockingQueue[Option[String]]()
+
+ // Owns the writing end for the same reason.
+ private val writer: ExecutorService = Executors.newSingleThreadExecutor { r =>
+ val t = new Thread(r, "python-worker-stdin")
+ t.setDaemon(true)
+ t
+ }
+
+ private val outThread: Thread = {
+ val t = new Thread(() => {
+ val r =
+ new BufferedReader(new InputStreamReader(process.getInputStream, StandardCharsets.UTF_8))
+ try {
+ var line = r.readLine()
+ while (line != null) {
+ lines.put(Some(line))
+ line = r.readLine()
+ }
+ } catch { case NonFatal(_) => () }
+ finally lines.put(None)
+ })
+ t.setDaemon(true)
+ t.setName("python-worker-stdout")
+ t.start()
+ t
+ }
+
+ private val errThread: Thread = {
+ val t = new Thread(() => {
+ val r =
+ new BufferedReader(new InputStreamReader(process.getErrorStream, StandardCharsets.UTF_8))
+ try {
+ var line = r.readLine()
+ while (line != null) {
+ errBuf.synchronized(errBuf.append(line).append('\n'))
+ line = r.readLine()
+ }
+ } catch { case NonFatal(_) => () }
+ })
+ t.setDaemon(true)
+ t.setName("python-worker-stderr")
+ t.start()
+ t
+ }
+
+ /** Wait for the worker's startup `{"ready": true}`; if it dies first (e.g. an
+ * import failed) or never gets there, surface its stderr. A worker that
+ * fails here is killed: it is not in the pool's set yet, so nothing else
+ * will reap it.
+ *
+ * A first line that is not the protocol at all counts as not ready, rather
+ * than throwing whatever the parser throws: the caller's contract here is
+ * [[WorkerDiedException]], and an escape past it would leave this
+ * interpreter with nothing to reap it. The line goes into the message —
+ * stderr is empty when a script writes its noise to stdout.
+ */
+ def awaitReady(timeoutMillis: Long): Unit = {
+ val line = nextLine(timeoutMillis, "signal ready")
+ val ready =
+ try objectMapper.readTree(line).path("ready").asBoolean(false)
+ catch { case NonFatal(_) => false }
+ if (!ready) {
+ destroy()
+ throw new WorkerDiedException(
+ s"python worker [$label] did not signal ready; it wrote: ${abbreviate(line)}." +
+ s" stderr:\n${drainErr()}"
+ )
+ }
+ }
+
+ /** An answer without an exit code is the worker breaking protocol, not a job
+ * that failed, so it leaves as a worker death — the same refusal to guess
+ * [[awaitReady]] makes about the startup line. Defaulting it would report the
+ * caller's own input as the thing that failed, with an empty stdout as the
+ * evidence, where a death routes to the caller's fallback instead. An absent
+ * stream is a different matter: empty is what it means.
+ */
+ def run(request: ObjectNode, timeoutMillis: Long): Outcome =
+ try {
+ send(request, timeoutMillis)
+ val line = nextLine(timeoutMillis, "answer")
+ val node = objectMapper.readTree(line)
+ val exit = node.path("exit")
+ if (!exit.isNumber) {
+ throw new WorkerDiedException(
+ s"python worker [$label] answered without an exit code;" +
+ s" it wrote: ${abbreviate(line)}"
+ )
+ }
+ Outcome(exit.asInt(), node.path("stdout").asText(""), node.path("stderr").asText(""))
+ } catch {
+ case e: WorkerDiedException => throw e
+ case NonFatal(e) =>
+ throw new WorkerDiedException(
+ s"I/O error talking to python worker [$label]: ${e.getMessage}",
+ e
+ )
+ }
+
+ /** Hand the request over, on the same deadline as the answer. A worker that
+ * has stopped reading its stdin blocks the write as soon as the payload
+ * outgrows the pipe buffer, and that write is no more interruptible than the
+ * read, so it too runs on a thread of its own — a daemon, so a lost one
+ * cannot keep the JVM alive. Killing the worker closes the pipe, which is
+ * what releases that thread.
+ */
+ private def send(request: ObjectNode, timeoutMillis: Long): Unit = {
+ val write = writer.submit(new Callable[Unit] {
+ override def call(): Unit = {
+ stdin.write(objectMapper.writeValueAsString(request))
+ stdin.write("\n")
+ stdin.flush()
+ }
+ })
+ try write.get(timeoutMillis, TimeUnit.MILLISECONDS)
+ catch {
+ case _: TimeoutException =>
+ destroy()
+ throw new WorkerDiedException(
+ s"python worker [$label] did not read its request within ${timeoutMillis}ms;" +
+ s" killed it. stderr:\n${drainErr()}"
+ )
+ case e: ExecutionException =>
+ throw new WorkerDiedException(
+ s"I/O error sending to python worker [$label]: ${e.getCause.getMessage}",
+ e.getCause
+ )
+ }
+ }
+
+ /** One protocol line, or a [[WorkerDiedException]]: the worker ended the
+ * stream, or it went `timeoutMillis` without writing. `what` names what was
+ * being waited for. A worker that timed out is killed here — that both frees
+ * the machine of a hung interpreter and unblocks [[outThread]] — and the
+ * pool discards it, so it never serves another job.
+ */
+ private def nextLine(timeoutMillis: Long, what: String): String =
+ lines.poll(timeoutMillis, TimeUnit.MILLISECONDS) match {
+ case null => // poll's own signal that the deadline passed with nothing written
+ destroy()
+ throw new WorkerDiedException(
+ s"python worker [$label] did not $what within ${timeoutMillis}ms; killed it." +
+ s" stderr:\n${drainErr()}"
+ )
+ case None => // end of stream: the interpreter is gone
+ throw new WorkerDiedException(s"python worker [$label] crashed. stderr:\n${drainErr()}")
+ case Some(line) => line
+ }
+
+ private def drainErr(): String = errBuf.synchronized(errBuf.toString)
+
+ /** Enough of a stray protocol line to recognize it, not a whole traceback. */
+ private def abbreviate(line: String): String =
+ if (line.length <= 200) line else s"${line.take(200)}... (${line.length} chars)"
+
+ /** Idempotent: called on a crash, a timeout, and again on pool shutdown.
+ *
+ * The process goes first and the streams after: a write blocked on a full
+ * pipe only comes back once the pipe has no reader, and closing the buffered
+ * writer would try to flush into that same pipe — so closing first is itself
+ * a way to hang. Nothing is flushed on the way out; a worker being destroyed
+ * has no use for the rest of a request.
+ */
+ def destroy(): Unit = {
+ writer.shutdownNow()
+ process.destroy()
+ try if (!process.waitFor(2, TimeUnit.SECONDS)) process.destroyForcibly()
+ catch { case NonFatal(_) => process.destroyForcibly() }
+ try process.getOutputStream.close()
+ catch { case NonFatal(_) => () }
+ }
+ }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPoolSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPoolSpec.scala
new file mode 100644
index 0000000000..6187cf8e24
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPoolSpec.scala
@@ -0,0 +1,266 @@
+/*
+ * 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 org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.concurrent.{Executors, TimeUnit}
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutorService, Future}
+import scala.util.Try
+
+/**
+ * What the pool owes a caller when a worker misbehaves. An ordinary job failure
+ * is the other suites' business; this one is about a worker that stays alive and
+ * stops taking part, which is the case that does not end by itself: a crash
+ * closes the pipe and the pending read returns, while silence would hold the
+ * caller forever — neither a read nor a write on a process pipe answers an
+ * interrupt or a deadline, so no suite-level timeout can release one.
+ *
+ * Every wait here is bounded and runs on daemon threads, so a regression fails
+ * these tests instead of wedging the run: a lost non-daemon thread parked on a
+ * pipe would keep the JVM, and the build, alive.
+ *
+ * The fixture worker is stdlib-only and runs under `-I -S`, so this needs an
+ * interpreter but none of the operator packages.
+ */
+final class PythonWorkerPoolSpec extends AnyFunSuite {
+
+ private val HangingWorker = "/python/hanging_worker.py"
+
+ /** Short enough to keep the suite quick, far enough above process startup to
+ * not be mistaken for one: a fixture asked to hang never answers at all, so
+ * these deadlines cannot race it.
+ */
+ private val ShortTimeouts: PythonWorkerPool.Timeouts =
+ PythonWorkerPool.Timeouts(responseMillis = 1500, startupMillis = 1500)
+
+ /** For the case that interrupts its own callers: far enough out that they are
+ * certainly still waiting on the worker, not already past their deadline.
+ */
+ private val PatientTimeouts: PythonWorkerPool.Timeouts =
+ PythonWorkerPool.Timeouts(responseMillis = 60000, startupMillis = 60000)
+
+ /** Ceiling on a whole case, well above the deadlines under test. Reaching it
+ * means something never gave up.
+ */
+ private val Bound: FiniteDuration = 25.seconds
+
+ /** Any interpreter serves — the fixture imports only `json` and `time` — so
+ * this deliberately skips the configured `python.path` the suites that need
+ * pandas resolve. A machine without one cancels rather than fails.
+ */
+ private lazy val python: String = {
+ def isRunnable(exe: String): Boolean =
+ Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption
+ .exists { p =>
+ if (p.waitFor(5, TimeUnit.SECONDS)) p.exitValue() == 0 else { p.destroyForcibly(); false }
+ }
+
+ List("python3", "python", "py").find(isRunnable).getOrElse(cancel("no runnable python"))
+ }
+
+ /** Resolves [[python]] here so that the cancel lands on the test thread. Forced
+ * inside a case, it would be forced inside a `Future`, and by the time it
+ * reached `intercept` a cancellation is a RuntimeException like any other: the
+ * cases would fail on a machine without an interpreter instead of cancelling,
+ * and the one at the cap would record it as the failure it asserts on and pass
+ * without ever reaching the cap.
+ */
+ override def withFixture(test: NoArgTest): org.scalatest.Outcome = {
+ val _ = python
+ super.withFixture(test)
+ }
+
+ private def onDaemonThreads[T](threads: Int)(body: ExecutionContext => T): T = {
+ val pool = Executors.newFixedThreadPool(
+ threads,
+ (r: Runnable) => {
+ val t = new Thread(r, "pool-spec-caller")
+ t.setDaemon(true)
+ t
+ }
+ )
+ val ec: ExecutionContextExecutorService = ExecutionContext.fromExecutorService(pool)
+ try body(ec)
+ finally pool.shutdownNow()
+ }
+
+ private def call(
+ launchArgs: Seq[String],
+ request: ObjectNode,
+ timeouts: PythonWorkerPool.Timeouts
+ ): PythonWorkerPool.Outcome =
+ PythonWorkerPool.run(
+ resourcePath = HangingWorker,
+ launchArgs = launchArgs,
+ pythonExe = python,
+ request = request,
+ interpreterArgs = Seq("-I", "-S"),
+ timeouts = timeouts
+ )
+
+ /** A job the fixture takes and never answers. `hang` travels in the request
+ * rather than in `launchArgs` so that [[healthyCall]] lands in the same
+ * sub-pool: `Key` covers the script, its arguments and the environment, and a
+ * job routed elsewhere would be served by a pool whose queue this suite never
+ * touched.
+ */
+ private def hangingCall(
+ launchArgs: Seq[String],
+ request: ObjectNode = objectMapper.createObjectNode(),
+ timeouts: PythonWorkerPool.Timeouts = ShortTimeouts
+ ): PythonWorkerPool.Outcome =
+ call(launchArgs, request.deepCopy().put("hang", true), timeouts)
+
+ /** A job in that same sub-pool which the fixture does answer. */
+ private def healthyCall(): PythonWorkerPool.Outcome =
+ call(Seq.empty, objectMapper.createObjectNode(), ShortTimeouts)
+
+ /** The call, on a daemon thread and under [[Bound]], expected to give up. */
+ private def interceptBounded(call: => Any): PythonWorkerPool.WorkerDiedException =
+ intercept[PythonWorkerPool.WorkerDiedException] {
+ onDaemonThreads(1)(ec => Await.result(Future(call)(ec), Bound))
+ }
+
+ test("a worker that takes the job and stops answering is killed and reported") {
+ val startedAt = System.nanoTime()
+ val thrown = interceptBounded(hangingCall(Seq.empty))
+ val elapsedMillis = (System.nanoTime() - startedAt) / 1000000
+
+ assert(thrown.getMessage.contains("did not answer"))
+ assert(thrown.getMessage.contains("killed it"))
+ // Well under the default response budget: what fired is the timeout passed in,
+ // not a wait that happened to end.
+ assert(elapsedMillis < PythonWorkerPool.Timeouts.Default.responseMillis / 2)
+ }
+
+ test("a worker that never signals ready is killed and reported") {
+ val startedAt = System.nanoTime()
+ val thrown = interceptBounded(hangingCall(Seq("--hang-before-ready")))
+ val elapsedMillis = (System.nanoTime() - startedAt) / 1000000
+
+ assert(thrown.getMessage.contains("did not signal ready"))
+ assert(elapsedMillis < PythonWorkerPool.Timeouts.Default.startupMillis / 2)
+ }
+
+ test("a worker that never reads its request is killed and reported") {
+ val request = objectMapper.createObjectNode()
+ // Past any pipe buffer, so the write cannot simply be handed to the kernel and
+ // left there: it is the blocked write itself that has to be given up on.
+ request.put("source", "x" * (4 * 1024 * 1024))
+
+ val thrown = interceptBounded(hangingCall(Seq("--deaf"), request))
+
+ assert(thrown.getMessage.contains("did not read its request"))
+ assert(thrown.getMessage.contains("killed it"))
+ }
+
+ test("a caller waiting at the worker cap is not stranded by a discarded worker") {
+ // One caller more than there are workers, all onto workers that go quiet:
+ // the callers that hold one time out, and their workers are discarded, which
+ // frees a slot without handing anything back, and the caller waiting at the
+ // cap has to notice that rather than wait for a hand-back that never comes.
+ // The discards it is waiting on happen on workers other than the one ahead
+ // of it.
+ val callers = PythonWorkerPool.maxWorkers + 1
+
+ val outcomes = onDaemonThreads(callers) { implicit ec =>
+ Await.result(Future.sequence(Seq.fill(callers)(Future(Try(hangingCall(Seq.empty))))), Bound)
+ }
+
+ assert(outcomes.length == callers)
+ assert(outcomes.forall(_.isFailure))
+ }
+
+ test("the pool still serves jobs after it has discarded a timed-out worker") {
+ interceptBounded(hangingCall(Seq.empty))
+
+ // Deliberately the sub-pool the discard happened in — see [[hangingCall]] —
+ // so what is asserted is that a pool short one worker starts a replacement,
+ // not that an untouched pool works.
+ assert(healthyCall().exit == 0)
+ }
+
+ test("an interpreter that cannot be started reaches the caller as a worker death") {
+ // Not the IOException ProcessBuilder raises: a caller's fallback is written
+ // against WorkerDiedException, and a pool that cannot hand out a worker at all
+ // is the case that fallback exists for.
+ val thrown = intercept[PythonWorkerPool.WorkerDiedException] {
+ PythonWorkerPool.run(
+ resourcePath = HangingWorker,
+ launchArgs = Seq.empty,
+ pythonExe = "no-such-python-on-this-machine",
+ request = objectMapper.createObjectNode(),
+ interpreterArgs = Seq("-I", "-S"),
+ timeouts = ShortTimeouts
+ )
+ }
+
+ assert(thrown.getMessage.contains("could not start python worker"))
+ }
+
+ test("a caller interrupted mid-job does not cost the pool a worker") {
+ val workers = PythonWorkerPool.maxWorkers
+
+ // Every slot taken by a job nobody will answer, and then the callers are
+ // interrupted — `shutdownNow` is what an executor does to a fan-out whose test
+ // has already failed. An interrupt is not a WorkerDiedException, so a worker
+ // left neither returned nor discarded would cost this sub-pool that slot for
+ // the rest of the JVM.
+ onDaemonThreads(workers) { implicit ec =>
+ Seq.fill(workers)(Future(Try(hangingCall(Seq.empty, timeouts = PatientTimeouts))))
+ // Enough for the callers to be waiting on a worker rather than starting one;
+ // the slot has to come back wherever the interrupt lands, so this only
+ // decides which of the two paths the case exercises.
+ Thread.sleep(500)
+ }
+
+ // Serving `workers` jobs again is the whole assertion: a leaked slot leaves
+ // these at the cap, rechecking it until [[Bound]] runs out.
+ val outcomes = onDaemonThreads(workers) { implicit ec =>
+ Await.result(Future.sequence(Seq.fill(workers)(Future(healthyCall()))), Bound)
+ }
+
+ assert(outcomes.forall(_.exit == 0))
+ }
+
+ test("a worker that answers without an exit code is reported as a worker death") {
+ // Not as a job that failed: a default would name the caller's own request as
+ // what went wrong, with an empty stdout as the evidence.
+ val request = objectMapper.createObjectNode().put("drop-exit", true)
+
+ val thrown = interceptBounded(call(Seq.empty, request, ShortTimeouts))
+
+ assert(thrown.getMessage.contains("without an exit code"))
+ assert(thrown.getMessage.contains("stdout"))
+ }
+
+ test("a worker whose first line is not the protocol is killed and reported") {
+ val thrown = interceptBounded(hangingCall(Seq("--babble")))
+
+ assert(thrown.getMessage.contains("did not signal ready"))
+ // Named in the message: stderr is empty when a script writes its noise to
+ // stdout, so the line itself is the only evidence of what went wrong.
+ assert(thrown.getMessage.contains("not a protocol line"))
+ }
+}
diff --git a/project/TestFilters.scala b/project/TestFilters.scala
new file mode 100644
index 0000000000..c948f1d45e
--- /dev/null
+++ b/project/TestFilters.scala
@@ -0,0 +1,48 @@
+/*
+ * 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.
+ */
+
+import sbt._
+
+/**
+ * Selects a module's tagged tests for the fast-unit job or the integration job:
+ * skip-integration excludes them, integration-only runs only them, unset runs
+ * everything, and any other value fails the build rather than quietly running
+ * everything in a job that selected a subset. Shared because the mapping is
+ * identical in every module, while the env var and the tag are not — the tag
+ * annotation has to live somewhere the module's own Test config can see.
+ */
+object TestFilters {
+
+ /** @param envVar the variable the two CI jobs set to opposite values; it has to
+ * be the one the workflow already sets on the step that invokes
+ * this module's tests, or neither subset is selected.
+ * @param tag fully-qualified name of the tag annotation, as
+ * `classOf[...].getName` gives it at the test site — ScalaTest
+ * matches these by string, so a rename that misses one side
+ * silently stops filtering.
+ */
+ def integrationSplit(envVar: String, tag: String): Seq[TestOption] =
+ sys.env.get(envVar) match {
+ case Some("skip-integration") =>
+ Seq(Tests.Argument(TestFrameworks.ScalaTest, "-l", tag))
+ case Some("integration-only") =>
+ Seq(Tests.Argument(TestFrameworks.ScalaTest, "-n", tag))
+ case Some(other) => sys.error(s"$envVar=$other: use skip-integration or integration-only")
+ case None => Nil
+ }
+}