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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
