You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zozo123 (via GitHub)" <gi...@apache.org> on 2026/10/04 13:49:20 UTC
[PR] [WIP][SPARK-55278][UDF] Add an external command that runs a direct worker inside sbx [spark]
zozo123 opened a new pull request, #59232:
URL: https://github.com/apache/spark/pull/59232
### What changes were proposed in this pull request?
Add `bin/sbx-udf-worker`. It is an external worker command. Spark does not call it, and Spark gains no sandbox configuration. An operator puts the command in the direct worker spec. Spark appends `--id` and `--connection` as it already does for every direct worker.
The command starts the real worker with `sbx exec` and binds the host socket named by `--connection`. A socket created inside the sandbox belongs to the guest kernel, so the host executor cannot connect to that file even when the path is mounted.
This is a draft. It is not part of https://github.com/apache/spark/pull/59155.
### Why are the changes needed?
A direct worker has to bind the Unix socket on the executor host. Running that worker inside Docker Sandboxes needs a host process that owns the host socket and forwards bytes. The existing worker command is that extension point.
### Does this PR introduce _any_ user-facing change?
Yes, for an operator who chooses this command. The command is:
```
bin/sbx-udf-worker --sandbox NAME -- GUEST_COMMAND
```
Spark then appends `--id` and `--connection`. Executors that do not use the command are unchanged. Spark does not require `sbx`.
### How was this patch tested?
The same forwarding program returned rows `(0, 0)` through `(5, 5)` from `PlanExternalUDFsSuite` through two guest workers. Each worker logged `Linux aarch64` and `transport=epoll`, accepted one connection, and stopped on SIGTERM. That run is not repeated here because the suite change is still in https://github.com/apache/spark/pull/59155, and Spark CI has no `sbx` login.
`python3 -m py_compile bin/sbx-udf-worker` passed. The command rejects a missing `--` and a command that has no `--id` or `--connection`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.7
Made with [Cursor](https://cursor.com)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "zozo123 (via GitHub) via reviews" <re...@spark.apache.org>.
zozo123 commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6557270436
@haiyangsun-db gentle ping, this is green and mergeable. Happy to adjust anything.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "zozo123 (via GitHub)" <gi...@apache.org>.
zozo123 commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6145750687
Independent verification: **PASS+NOTES** at head `fee9de6ef9828167ec5654cd7c0a830e043acc96`, base `9b452e002fb9985afd5971e67ee7a646c8c2b207` (the merge base). The `git patch-id --stable` of the base-to-head diff is `b4c32997f587f785a7649fab1713cb78bdf3e088`. I did not write this change.
The diff touches 4 files, all test sources plus one `worker_spec.proto` comment. Engine code does not change. All six review threads are resolved, and each requested fix is present at this head. The PR is `MERGEABLE` and `CLEAN`.
Evidence. Each run used `./build/mvn -B -pl udf/worker/grpc -am -Dtest=none -Dsurefire.failIfNoSpecifiedTests=false -DwildcardSuites=org.apache.spark.udf.worker.grpc.DirectGrpcDispatcherIntegrationSuite test`.
- macOS arm64 from a clean checkout at this head. 6/6 passed, including both launcher tests.
- Linux arm64, `eclipse-temurin:17-jdk` (17.0.20.1), sources byte-identical to this head. PID 1 was `sleep infinity`, so orphans are never reaped. 6/6 passed in 4 of 4 runs.
- After each Linux run, `ps` showed exactly one `[java] <defunct>` with PPID 1. That is the inner worker of the SIGKILL test, which exited after its launcher was SIGKILLed. `ProcessHandle.of(pid).isAlive()` returned `true` for it, and `/proc/<pid>/stat` showed state `Z`. So the zombie check in `assertWorkerStopped` is what lets the test pass on non-reaping hosts. No `ForwardingWorkerMain` or `EchoGrpcWorkerMain` process was left running on either host.
The tests cover these lifecycle paths:
- SIGTERM: session close.
- Forced SIGKILL and parent death: `destroyForcibly` on the launcher stops the inner worker through its `onExit` parent watch.
- Zombies: Linux state `Z` counts as exited.
- Cleanup: launcher and worker are both gone after close.
Notes, none blocking:
- CI did not run these tests. The Maven job uses `-DskipTests`, and the `udf-worker` module in `dev/sparktestsupport/modules.py` only runs `udf-worker-core/test`. This gap predates this PR. The test evidence above is from local runs.
- If the launcher dies before `EchoGrpcWorkerMain` reads `ProcessHandle.current().parent()`, the watch targets the reparent target. The window is test-only and narrow.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4231761207
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/ForwardingWorkerMain.scala:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.spark.udf.worker.grpc.testing
+
+import java.io.IOException
+import java.net.{StandardProtocolFamily, UnixDomainSocketAddress}
+import java.nio.ByteBuffer
+import java.nio.channels.{ServerSocketChannel, SocketChannel}
+import java.nio.file.{Files, Paths}
+import java.util.concurrent.TimeUnit
+
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+/**
+ * Test-only launcher that stands in for a sandbox or container launcher. It
+ * receives the engine's `--id` and `--connection`, starts the real worker on a
+ * different socket, and forwards each engine connection to it. This is the
+ * shape a launcher has when the worker cannot bind the engine's socket itself:
+ *
+ * java -cp <classpath> ...ForwardingWorkerMain <worker command...>
+ * --id <id> --connection <uds-path>
+ *
+ * The inner worker exits when this launcher exits, including by `SIGKILL`.
Review Comment:
**Nit (P3):** Nothing in `ForwardingWorkerMain` makes this true by itself. A SIGKILLed JVM runs no shutdown hook, and the worker is started with a plain `ProcessBuilder`, so after the launcher is killed the inner worker stops only because `EchoGrpcWorkerMain` watches its parent (`EchoGrpcWorkerMain.scala:133-135`). If this launcher wraps any other worker command, that worker is orphaned on SIGKILL. The next line also leaves out that the SIGTERM wait is capped at 2 seconds and then followed by `destroyForcibly()`.
Since this file is meant to stand in for sandbox and container launchers, readers may copy it as the way to meet the new `ProcessCallable` clause. The earlier thread asked for the scaladoc to say how the SIGKILL clause is met. Something like:
```
* On `SIGTERM`, the shutdown hook sends `SIGTERM` to the inner worker, waits up
* to 2 seconds, and then kills it with `SIGKILL`. A `SIGKILL` of this launcher
* runs no hook, so the inner worker must watch its parent and exit on its own;
* `EchoGrpcWorkerMain` does this.
```
**Verification:**
- **Inspection:** Reading the scaladoc against the shutdown hook and EchoGrpcWorkerMain's parent watch, every lifecycle claim matches the code: the 2 s bounded SIGTERM wait with SIGKILL fallback, and that the SIGKILL guarantee comes from the inner worker.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,65 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got $workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input), emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should terminate the launcher")
+ workers.foreach { worker =>
+ worker.onExit().get(10, TimeUnit.SECONDS)
+ assert(!worker.isAlive, s"session close should stop the launched worker ${worker.pid}")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got $workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ worker.onExit().get(10, TimeUnit.SECONDS)
Review Comment:
**Nit (P3):** Once the launcher is SIGKILLed, this inner worker is no longer a child of anything that waits for it. It is reparented to PID 1 (or a subreaper), and after it exits it stays a zombie until that process reaps it. JDK `ProcessHandle` counts a non-child zombie as alive. I checked on OpenJDK 11.0.32: for a zombie that is not a child of the JVM, `ProcessHandle.of(pid).isAlive()` returned `true` and `onExit().get(2, SECONDS)` timed out. So on a host whose PID 1 does not reap orphans, this line fails with `TimeoutException` even though the parent watch worked and the worker exited. Examples are a container started without `--init` with the build JVM as PID 1, or a GitHub Actions job container. The session-close test hits the same thing at line 195 whenever the inner worker takes longer than the launcher's 2 s grace and gets SIGKILLed.
A small suite helper for both waits would avoid this. It would wait a bounded time until the handle is no longer alive or is a zombie (on Linux, state `Z` in `/proc/<pid>/stat`), and destroy the handle before failing if it is still running. The destroy-before-failing part also covers the leaked-worker finding on this test.
See **Shared repair plan 1** in the review body.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/ForwardingWorkerMain.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.spark.udf.worker.grpc.testing
+
+import java.io.IOException
+import java.net.{StandardProtocolFamily, UnixDomainSocketAddress}
+import java.nio.ByteBuffer
+import java.nio.channels.{ServerSocketChannel, SocketChannel}
+import java.nio.file.{Files, Paths}
+import java.util.concurrent.TimeUnit
+
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+/**
+ * Test-only launcher that stands in for a sandbox or container launcher. It
+ * receives the engine's `--id` and `--connection`, starts the real worker on a
+ * different socket, and forwards each engine connection to it. This is the
+ * shape a launcher has when the worker cannot bind the engine's socket itself:
+ *
+ * java -cp <classpath> ...ForwardingWorkerMain <worker command...>
+ * --id <id> --connection <uds-path>
+ *
+ * `SIGTERM` stops the inner worker with `SIGTERM` and waits for it.
+ */
+object ForwardingWorkerMain {
+
+ def main(args: Array[String]): Unit = {
+ val connection = args.indexOf("--connection")
+ require(connection >= 0 && connection + 1 < args.length, "--connection is required")
+ val socketPath = args(connection + 1)
+ val innerPath = socketPath + ".inner"
+ val workerCommand =
+ args.take(connection) ++ args.drop(connection + 2) ++ Seq("--connection", innerPath)
+
+ Files.deleteIfExists(Paths.get(innerPath))
+ val worker = new ProcessBuilder(workerCommand.toSeq.asJava).inheritIO().start()
+ // This standalone test-worker main has no Spark on its classpath, so it
+ // cannot use Spark's ShutdownHookManager and registers directly.
+ // scalastyle:off runtimeaddshutdownhook
+ Runtime.getRuntime.addShutdownHook(new Thread(() => {
Review Comment:
Thanks, I checked acb3771b. EchoGrpcWorkerMain now registers the parent onExit watch before it parks. ForwardingWorkerMain's 2 s SIGTERM wait now finishes before the suite's 10 s graceful timeout. The new test SIGKILLs the launcher with no SIGTERM and checks that the inner worker exits. That fixes the orphaned inner worker this thread was about.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4226908183","thread_id":"inline:4226908183","verdict_sha256":"115e561e8987ebf66c34b52efa7fa104d464522610163613e77351f7e1863a2b"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6328285449
cc @haiyangsun-db
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "zozo123 (via GitHub)" <gi...@apache.org>.
zozo123 commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4233891915
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala:
##########
@@ -125,7 +128,11 @@ object EchoGrpcWorkerMain {
server.start()
stderrln(s"listening on $socketPath (transport=${transport.name})")
- // Park until the shutdown hook fires.
+ // SIGKILL skips JVM shutdown hooks. Register before parking so a parent
+ // that has already exited still runs cleanup and unblocks the wait below.
+ if (launcher.isPresent) {
+ launcher.get().onExit().thenRun(() => cleanup())
Review Comment:
Done in `fee9de6e`. The class doc now says the worker blocks until the JVM receives SIGTERM or its parent process exits.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,91 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got $workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input), emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should terminate the launcher")
+ workers.foreach { worker =>
+ assertWorkerStopped(worker, "session close should stop the launched worker")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got $workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ assertWorkerStopped(worker, "SIGKILL of the launcher should stop worker")
+ } finally {
+ session.close(emptyCancel)
+ }
+ }
+
+ // An orphaned worker is reparented, and an unreaped zombie stays isAlive
+ // until PID 1 reaps it. State Z in /proc means the worker has already exited.
+ private def assertWorkerStopped(worker: ProcessHandle, what: String): Unit = {
Review Comment:
Done in `fee9de6e`. `assertWorkerStopped`, `hasExited`, and `linuxState` sit with the other private helpers, above the first test.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,91 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got $workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input), emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should terminate the launcher")
+ workers.foreach { worker =>
+ assertWorkerStopped(worker, "session close should stop the launched worker")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got $workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ assertWorkerStopped(worker, "SIGKILL of the launcher should stop worker")
+ } finally {
+ session.close(emptyCancel)
+ }
+ }
+
+ // An orphaned worker is reparented, and an unreaped zombie stays isAlive
+ // until PID 1 reaps it. State Z in /proc means the worker has already exited.
+ private def assertWorkerStopped(worker: ProcessHandle, what: String): Unit = {
+ val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10)
+ while (System.nanoTime() < deadline && !hasExited(worker)) {
+ Thread.sleep(50)
+ }
+ if (!hasExited(worker)) {
+ worker.destroyForcibly()
+ fail(s"$what ${worker.pid}")
+ }
+ }
+
+ private def hasExited(worker: ProcessHandle): Boolean = {
+ !worker.isAlive || linuxState(worker.pid).contains('Z')
+ }
+
+ private def linuxState(pid: Long): Option[Char] = {
+ val stat = Paths.get(s"/proc/$pid/stat")
+ if (!Files.isRegularFile(stat)) {
+ None
+ } else {
+ val text = Files.readString(stat)
Review Comment:
Done in `fee9de6e`. A failed `/proc` read returns no state, so a reap between the checks does not fail the test. Both launcher tests passed locally.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,65 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got $workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input), emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should terminate the launcher")
+ workers.foreach { worker =>
+ worker.onExit().get(10, TimeUnit.SECONDS)
+ assert(!worker.isAlive, s"session close should stop the launched worker ${worker.pid}")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got $workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ worker.onExit().get(10, TimeUnit.SECONDS)
Review Comment:
Done in `fee9de6e`. `linuxState` treats an `IOException` from that read as no state.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "spark-test (via GitHub)" <gi...@apache.org>.
spark-test commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6260356039
The AI review requested by @HyukjinKwon finished.
*Generated by Omnigent on Databricks*
<!-- SPARK_OSS_REVIEW_REQUEST 33610049161 -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "zozo123 (via GitHub)" <gi...@apache.org>.
zozo123 commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4232086637
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/ForwardingWorkerMain.scala:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.spark.udf.worker.grpc.testing
+
+import java.io.IOException
+import java.net.{StandardProtocolFamily, UnixDomainSocketAddress}
+import java.nio.ByteBuffer
+import java.nio.channels.{ServerSocketChannel, SocketChannel}
+import java.nio.file.{Files, Paths}
+import java.util.concurrent.TimeUnit
+
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+/**
+ * Test-only launcher that stands in for a sandbox or container launcher. It
+ * receives the engine's `--id` and `--connection`, starts the real worker on a
+ * different socket, and forwards each engine connection to it. This is the
+ * shape a launcher has when the worker cannot bind the engine's socket itself:
+ *
+ * java -cp <classpath> ...ForwardingWorkerMain <worker command...>
+ * --id <id> --connection <uds-path>
+ *
+ * The inner worker exits when this launcher exits, including by `SIGKILL`.
Review Comment:
Done in `c30ddb05`. The scaladoc now says the 2 second SIGTERM wait ends in SIGKILL, and that a SIGKILL of the launcher runs no hook. The inner worker stops because `EchoGrpcWorkerMain` watches its parent.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,65 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got $workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input), emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should terminate the launcher")
+ workers.foreach { worker =>
+ worker.onExit().get(10, TimeUnit.SECONDS)
+ assert(!worker.isAlive, s"session close should stop the launched worker ${worker.pid}")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got $workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ worker.onExit().get(10, TimeUnit.SECONDS)
Review Comment:
Done in `c30ddb05`. Both launcher waits treat Linux state `Z` as exited, so an unreaped zombie does not time out the test. If the worker is still running when the wait ends, the test calls `destroyForcibly` before failing. Both launcher tests passed locally.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "zozo123 (via GitHub)" <gi...@apache.org>.
zozo123 commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6258679945
@HyukjinKwon rebased and CI is green on `b4d34b63585e`. The earlier notes are addressed. Ready for your review.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "spark-test (via GitHub)" <gi...@apache.org>.
spark-test commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6260351644
I checked the earlier fixes at b4d34b63585e: the lifecycle docs, parent watch, bounded launcher wait, zombie-aware exit checks, IOException handling, and helper placement are present. This check was from the pinned source; I did not rerun the tests or observe CI.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:6258679945","thread_id":"conversation:6258679945","verdict_sha256":"08b630e06bebe39c5a24d15a0d21fa7c4e3ba93aca8b2a1325547182ce5dd6aa"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4233007629
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,91 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got $workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input), emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should terminate the launcher")
+ workers.foreach { worker =>
+ assertWorkerStopped(worker, "session close should stop the launched worker")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got $workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ assertWorkerStopped(worker, "SIGKILL of the launcher should stop worker")
+ } finally {
+ session.close(emptyCancel)
+ }
+ }
+
+ // An orphaned worker is reparented, and an unreaped zombie stays isAlive
+ // until PID 1 reaps it. State Z in /proc means the worker has already exited.
+ private def assertWorkerStopped(worker: ProcessHandle, what: String): Unit = {
Review Comment:
**Nit (P3):** `assertWorkerStopped`, `hasExited`, and `linuxState` land between the SIGKILL test and `the channel overrides the authority derived from the socket path`. Every other private helper in this suite (`echoRunner` through `grpcChannel`, lines 73-122) sits above the first test. The repository's AGENTS.md asks new members to follow the file's existing sectioning rather than be placed at their first use. Moving these three helpers, with their comment, up next to `grpcChannel` keeps the test section contiguous.
**Verification:**
- **Inspection:** The suite's member order is fields and private helpers first, then test cases, with no helper definition between tests.
- **Behavior:** Both launcher tests and the existing dispatcher integration tests still pass after the relocation.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala:
##########
@@ -125,7 +128,11 @@ object EchoGrpcWorkerMain {
server.start()
stderrln(s"listening on $socketPath (transport=${transport.name})")
- // Park until the shutdown hook fires.
+ // SIGKILL skips JVM shutdown hooks. Register before parking so a parent
+ // that has already exited still runs cleanup and unblocks the wait below.
+ if (launcher.isPresent) {
+ launcher.get().onExit().thenRun(() => cleanup())
Review Comment:
**Nit (P3):** With this registration the worker also shuts down when its parent process exits, but the object scaladoc at lines 40-41 still says it "blocks until the JVM receives `SIGTERM` (then shuts down gracefully)". `ForwardingWorkerMain`'s scaladoc now sends readers here for the parent watch that covers a launcher SIGKILL. A direct spawn also exits now when the spawning JVM exits. The class doc should name both stop triggers, for example: "blocks until the JVM receives `SIGTERM` or its parent process exits (then shuts down gracefully)".
**Verification:**
- **Inspection:** Reading the object scaladoc against main() shows that every documented stop path (the SIGTERM shutdown hook and the parent onExit cleanup) exists in the code, and every stop path in the code is documented.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,91 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got $workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input), emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should terminate the launcher")
+ workers.foreach { worker =>
+ assertWorkerStopped(worker, "session close should stop the launched worker")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got $workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ assertWorkerStopped(worker, "SIGKILL of the launcher should stop worker")
+ } finally {
+ session.close(emptyCancel)
+ }
+ }
+
+ // An orphaned worker is reparented, and an unreaped zombie stays isAlive
+ // until PID 1 reaps it. State Z in /proc means the worker has already exited.
+ private def assertWorkerStopped(worker: ProcessHandle, what: String): Unit = {
+ val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10)
+ while (System.nanoTime() < deadline && !hasExited(worker)) {
+ Thread.sleep(50)
+ }
+ if (!hasExited(worker)) {
+ worker.destroyForcibly()
+ fail(s"$what ${worker.pid}")
+ }
+ }
+
+ private def hasExited(worker: ProcessHandle): Boolean = {
+ !worker.isAlive || linuxState(worker.pid).contains('Z')
+ }
+
+ private def linuxState(pid: Long): Option[Char] = {
+ val stat = Paths.get(s"/proc/$pid/stat")
+ if (!Files.isRegularFile(stat)) {
+ None
+ } else {
+ val text = Files.readString(stat)
Review Comment:
**Nit (P3):** If PID 1 reaps the inner worker between `Files.isRegularFile(stat)` and this read, `/proc/<pid>` disappears and `readString` throws `NoSuchFileException` (or an ESRCH `IOException` if the file was opened just before). Nothing catches it, so `assertWorkerStopped` errors out exactly when the worker has stopped. I checked on OpenJDK 11.0.32 that once a pid is reaped, `isRegularFile` on its stat file returns false and `readString` throws `NoSuchFileException`. The window is small for each 50 ms poll, but both launcher tests go through this helper, so it can flake on a busy Linux CI host. Treating a failed read as "no state" lets the next poll's `isAlive` check end the wait, for example:
```scala
val text = try Files.readString(stat) catch { case _: IOException => "" }
```
The existing `close < 0` check then returns `None`. This needs `java.io.IOException` imported in the suite.
**Verification:**
- **Regression:** Calling the state helper for a pid whose /proc entry has disappeared, such as a process that has just been reaped, returns without throwing, and the exit check then reports the process as exited.
- **Behavior:** Both launcher tests still pass on Linux, and a still-running inner worker still makes the wait fail after it is force-killed.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/ForwardingWorkerMain.scala:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.spark.udf.worker.grpc.testing
+
+import java.io.IOException
+import java.net.{StandardProtocolFamily, UnixDomainSocketAddress}
+import java.nio.ByteBuffer
+import java.nio.channels.{ServerSocketChannel, SocketChannel}
+import java.nio.file.{Files, Paths}
+import java.util.concurrent.TimeUnit
+
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+/**
+ * Test-only launcher that stands in for a sandbox or container launcher. It
+ * receives the engine's `--id` and `--connection`, starts the real worker on a
+ * different socket, and forwards each engine connection to it. This is the
+ * shape a launcher has when the worker cannot bind the engine's socket itself:
+ *
+ * java -cp <classpath> ...ForwardingWorkerMain <worker command...>
+ * --id <id> --connection <uds-path>
+ *
+ * The inner worker exits when this launcher exits, including by `SIGKILL`.
Review Comment:
Thanks, I checked c30ddb05. The scaladoc now says the shutdown hook sends SIGTERM to the inner worker, waits up to 2 seconds, and then uses SIGKILL. It also says a SIGKILL of the launcher runs no hook, so the inner worker stops because EchoGrpcWorkerMain watches its parent. That matches the hook and the parent watch, so this thread is resolved.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4231761207","thread_id":"inline:4231761207","verdict_sha256":"58d0930295ab8ce0a8053cf959aba9caa5eb64d8fd8eb1b0a3567d4389bbb291"} -->
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala:
##########
@@ -160,6 +162,65 @@ class DirectGrpcDispatcherIntegrationSuite
assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
}
+ test("a launcher command that forwards the connection serves sessions") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val sessions = Seq.fill(2)(dispatcher.createSession(None))
+ val processes = sessions.map(workerProcess)
+ val workers = processes.flatMap(_.process.descendants().iterator().asScala)
+ val payloads = Seq(
+ ByteString.copyFrom(Array.fill[Byte](5 * 1024 * 1024)(3)),
+ ByteString.copyFromUtf8("small"))
+
+ try {
+ assert(workers.size == 2, s"each launcher should start one worker, got $workers")
+ sessions.zip(payloads).foreach { case (session, payload) =>
+ session.init(basicInit)
+ val input = DataRequest.newBuilder().setData(payload).build()
+ assert(session.process(Iterator.single(input), emptyFinish).map(_.getData).toList ==
+ List(payload))
+ }
+ } finally {
+ sessions.foreach(_.close(emptyCancel))
+ }
+
+ assert(processes.forall(!_.process.isAlive), "session close should terminate the launcher")
+ workers.foreach { worker =>
+ worker.onExit().get(10, TimeUnit.SECONDS)
+ assert(!worker.isAlive, s"session close should stop the launched worker ${worker.pid}")
+ }
+ }
+
+ test("SIGKILL of the forwarding launcher stops the inner worker") {
+ val launcher = ProcessCallable.newBuilder()
+ .addCommand(javaExecutable)
+ .addCommand("-cp")
+ .addCommand(javaClasspath)
+ .addCommand(classOf[ForwardingWorkerMain.type].getName.stripSuffix("$"))
+ .addAllCommand(echoRunner.getCommandList)
+ .build()
+ dispatcher = new DirectGrpcDispatcher(workerSpec(launcher))
+ val session = dispatcher.createSession(None)
+ val process = workerProcess(session)
+ val workers = process.process.descendants().iterator().asScala.toList
+ assert(workers.size == 1, s"the launcher should start one worker, got $workers")
+ val worker = workers.head
+ try {
+ session.init(basicInit)
+ process.process.destroyForcibly()
+ worker.onExit().get(10, TimeUnit.SECONDS)
Review Comment:
Thanks, I checked c30ddb05. assertWorkerStopped now treats state Z as exited and calls destroyForcibly before failing, which fixes the zombie timeout and the leaked worker. One small gap remains in linuxState: if PID 1 reaps the worker between Files.isRegularFile and Files.readString, readString throws NoSuchFileException (I confirmed this on OpenJDK 11 for a reaped pid), and the test fails just as the worker stops. Treating an IOException from that read as exited, or as None followed by another isAlive check, closes it.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4231761219","thread_id":"inline:4231761219","verdict_sha256":"58d0930295ab8ce0a8053cf959aba9caa5eb64d8fd8eb1b0a3567d4389bbb291"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4226908183
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/ForwardingWorkerMain.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.spark.udf.worker.grpc.testing
+
+import java.io.IOException
+import java.net.{StandardProtocolFamily, UnixDomainSocketAddress}
+import java.nio.ByteBuffer
+import java.nio.channels.{ServerSocketChannel, SocketChannel}
+import java.nio.file.{Files, Paths}
+import java.util.concurrent.TimeUnit
+
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+/**
+ * Test-only launcher that stands in for a sandbox or container launcher. It
+ * receives the engine's `--id` and `--connection`, starts the real worker on a
+ * different socket, and forwards each engine connection to it. This is the
+ * shape a launcher has when the worker cannot bind the engine's socket itself:
+ *
+ * java -cp <classpath> ...ForwardingWorkerMain <worker command...>
+ * --id <id> --connection <uds-path>
+ *
+ * `SIGTERM` stops the inner worker with `SIGTERM` and waits for it.
+ */
+object ForwardingWorkerMain {
+
+ def main(args: Array[String]): Unit = {
+ val connection = args.indexOf("--connection")
+ require(connection >= 0 && connection + 1 < args.length, "--connection is required")
+ val socketPath = args(connection + 1)
+ val innerPath = socketPath + ".inner"
+ val workerCommand =
+ args.take(connection) ++ args.drop(connection + 2) ++ Seq("--connection", innerPath)
+
+ Files.deleteIfExists(Paths.get(innerPath))
+ val worker = new ProcessBuilder(workerCommand.toSeq.asJava).inheritIO().start()
+ // This standalone test-worker main has no Spark on its classpath, so it
+ // cannot use Spark's ShutdownHookManager and registers directly.
+ // scalastyle:off runtimeaddshutdownhook
+ Runtime.getRuntime.addShutdownHook(new Thread(() => {
Review Comment:
**Nit (P3):** This shutdown hook is the only thing that stops the inner worker, and JVM shutdown hooks do not run when the launcher is SIGKILLed. The engine SIGKILLs the launcher pid directly on two paths:
- `DirectWorkerDispatcher.cleanupRawSpawn`, with no SIGTERM, after any `connectWorker`/readiness failure.
- `WorkerArtifacts.close`, once `gracefulTimeoutMs` expires in `DirectWorkerProcess.close`. The suite's 10 s graceful timeout is the same as this hook's `waitFor(10, SECONDS)`, and the engine's timer starts first. So the engine's SIGKILL arrives before the `destroyForcibly()` fallback here can run.
In both cases the `EchoGrpcWorkerMain` JVM is reparented and stays parked in `server.awaitTermination()`, because nothing ties its lifetime to the launcher. For example, if the second `createSession` in the new test misses gRPC READY within 30 s on a loaded CI host, `cleanupRawSpawn` kills that launcher and its echo JVM keeps running on the host. It also means the in-tree example launcher does not meet the clause this PR adds to `worker_spec.proto`: "It must not leave the worker running after it exits for any reason, because the engine sends SIGKILL to the launcher only." The effect is limited to test runs that fail or are slow.
**Recommended change:** Make the launched test worker stop when its launcher exits, however the launcher exits. For example, the test worker watches its parent process (ProcessHandle.current().parent() onExit) and runs its normal shutdown path when the parent disappears. Keep ForwardingWorkerMain's own SIGTERM wait for the inner worker below the suite's graceful termination timeout, so its forced-kill fallback runs before the engine escalates. Update the ForwardingWorkerMain scaladoc to say how the helper satisfies the SIGKILL clause. Add a forwarding launcher regression test to the dispatcher integration tests that kills the launcher outright and checks that its worker exits.
**Why this works:** A worker-side parent-liveness watch fires whether the launcher exits normally, through its SIGTERM hook, or by SIGKILL. That makes the inner worker's lifetime bounded by the launcher's own, which is exactly the contract's 'must not leave the worker running after it exits for any reason'. A hook wait shorter than the graceful timeout keeps the existing SIGTERM path deterministic instead of racing the engine's SIGKILL.
**Scope:** Test-only change to the UDF gRPC worker test harness: lifetime binding between the forwarding launcher and its worker, plus one regression test. No production or proto change.
**Compatibility:** The engine-facing behavior of the test binaries stays the same: --id/--connection handling, readiness on the engine socket, and SIGTERM-driven graceful stop.
**Risks:** A parent-exit watch must not fire spuriously. If the parent is already gone at startup, the process is reparented, so the watch has to be registered against the original parent before blocking.
**Constraints:** Keep the change inside test sources. The accepted frame says engine behavior and the proto contract stay unchanged.
**Success:** When the engine SIGKILLs the forwarding launcher, with or without a prior SIGTERM, its inner worker exits within a bounded time. The existing launcher test still passes, and session close still stops both the launcher and its worker. Direct (non-launcher) echo worker tests behave as before.
**Verification:**
- **Regression:** Starting a session through the forwarding launcher and then killing the launcher outright, with no SIGTERM, leaves no inner worker alive after a bounded wait.
- **Behavior:** Launcher-served sessions still round-trip large and small payloads, and graceful session close stops both processes.
- **Compatibility:** Direct echo worker spawn, close, and readiness tests are unaffected by the worker's parent watch.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "zozo123 (via GitHub)" <gi...@apache.org>.
zozo123 commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4231135658
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/ForwardingWorkerMain.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.spark.udf.worker.grpc.testing
+
+import java.io.IOException
+import java.net.{StandardProtocolFamily, UnixDomainSocketAddress}
+import java.nio.ByteBuffer
+import java.nio.channels.{ServerSocketChannel, SocketChannel}
+import java.nio.file.{Files, Paths}
+import java.util.concurrent.TimeUnit
+
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+/**
+ * Test-only launcher that stands in for a sandbox or container launcher. It
+ * receives the engine's `--id` and `--connection`, starts the real worker on a
+ * different socket, and forwards each engine connection to it. This is the
+ * shape a launcher has when the worker cannot bind the engine's socket itself:
+ *
+ * java -cp <classpath> ...ForwardingWorkerMain <worker command...>
+ * --id <id> --connection <uds-path>
+ *
+ * `SIGTERM` stops the inner worker with `SIGTERM` and waits for it.
+ */
+object ForwardingWorkerMain {
+
+ def main(args: Array[String]): Unit = {
+ val connection = args.indexOf("--connection")
+ require(connection >= 0 && connection + 1 < args.length, "--connection is required")
+ val socketPath = args(connection + 1)
+ val innerPath = socketPath + ".inner"
+ val workerCommand =
+ args.take(connection) ++ args.drop(connection + 2) ++ Seq("--connection", innerPath)
+
+ Files.deleteIfExists(Paths.get(innerPath))
+ val worker = new ProcessBuilder(workerCommand.toSeq.asJava).inheritIO().start()
+ // This standalone test-worker main has no Spark on its classpath, so it
+ // cannot use Spark's ShutdownHookManager and registers directly.
+ // scalastyle:off runtimeaddshutdownhook
+ Runtime.getRuntime.addShutdownHook(new Thread(() => {
Review Comment:
Done in `acb3771b`, and that Build is green. `EchoGrpcWorkerMain` records its parent before `awaitTermination` and runs cleanup when that parent exits. `ForwardingWorkerMain` waits 2 seconds on SIGTERM before `destroyForcibly`. The new test records the inner pid, SIGKILLs the launcher with no SIGTERM, and checks the inner worker is gone.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "zozo123 (via GitHub) via reviews" <re...@spark.apache.org>.
zozo123 commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4483941475
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala:
##########
@@ -54,6 +55,9 @@ object EchoGrpcWorkerMain {
throw new IllegalArgumentException("--connection <uds-path> is required"))
val workerId = parsed.getOrElse("id", "<unknown>")
+ // Captured before any blocking work. A launcher that dies during startup is
+ // still this handle, so its onExit callback runs even if it is already gone.
+ val launcher = ProcessHandle.current().parent()
Review Comment:
Agreed this is a real but narrow window. Parent is captured before Netty setup, and this helper is test-only.
I will leave the current capture as-is for this PR. If we tighten it later, passing the launcher pid into the child and exiting when ProcessHandle.current().parent() differs is the right follow-up. Happy to do that in a follow-up if you want it before merge.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "zozo123 (via GitHub)" <gi...@apache.org>.
zozo123 commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6309539489
@HyukjinKwon rebased and CI is green on `edbb847c6c1`. The earlier notes are addressed. Ready for your review.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "haiyangsun-db (via GitHub) via reviews" <re...@spark.apache.org>.
haiyangsun-db commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6663274301
Looks good at a glance, taking a deeper look and will stamp later.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "haiyangsun-db (via GitHub) via reviews" <re...@spark.apache.org>.
haiyangsun-db commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4483666914
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala:
##########
@@ -54,6 +55,9 @@ object EchoGrpcWorkerMain {
throw new IllegalArgumentException("--connection <uds-path> is required"))
val workerId = parsed.getOrElse("id", "<unknown>")
+ // Captured before any blocking work. A launcher that dies during startup is
+ // still this handle, so its onExit callback runs even if it is already gone.
+ val launcher = ProcessHandle.current().parent()
Review Comment:
one non-blocking P3:
If the launcher is SIGKILLed after spawning the child but before this line executes, the child may already be reparented. It would then watch PID 1 or another reaper instead of the launcher, potentially leaving the echo-worker JVM running after startup fails.
This should be rare because parent capture happens before Netty initialization. Maybe pass the launcher’s PID into the child and exit if the captured parent differs. Given the narrow startup window and test-only scope, this doesn’t block approval.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "haiyangsun-db (via GitHub) via reviews" <re...@spark.apache.org>.
haiyangsun-db commented on code in PR #59232:
URL: https://github.com/apache/spark/pull/59232#discussion_r4484637668
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala:
##########
@@ -54,6 +55,9 @@ object EchoGrpcWorkerMain {
throw new IllegalArgumentException("--connection <uds-path> is required"))
val workerId = parsed.getOrElse("id", "<unknown>")
+ // Captured before any blocking work. A launcher that dies during startup is
+ // still this handle, so its onExit callback runs even if it is already gone.
+ val launcher = ProcessHandle.current().parent()
Review Comment:
this can be up to you, don't want to over-complicate it for a test-only launcher.
##########
udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala:
##########
@@ -54,6 +55,9 @@ object EchoGrpcWorkerMain {
throw new IllegalArgumentException("--connection <uds-path> is required"))
val workerId = parsed.getOrElse("id", "<unknown>")
+ // Captured before any blocking work. A launcher that dies during startup is
+ // still this handle, so its onExit callback runs even if it is already gone.
+ val launcher = ProcessHandle.current().parent()
Review Comment:
up to you, don't want to over-complicate it for a test-only launcher.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55278][UDF] Document and test direct worker launcher commands [spark]
Posted by "haiyangsun-db (via GitHub) via reviews" <re...@spark.apache.org>.
haiyangsun-db commented on PR #59232:
URL: https://github.com/apache/spark/pull/59232#issuecomment-6781608658
@HyukjinKwon this testing-only touch should be safe to merge
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org