PG1204 commented on code in PR #8617: URL: https://github.com/apache/texera/pull/8617#discussion_r4138953882
########## common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/HuggingFaceParseBehaviorSpec.scala: ########## @@ -0,0 +1,196 @@ +/* + * 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.huggingFace.codegen + +import com.fasterxml.jackson.databind.ObjectMapper +import com.typesafe.config.ConfigFactory +import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path} +import java.util.concurrent.TimeUnit +import scala.jdk.CollectionConverters._ +import scala.util.Try + +/** + * Executes the *generated* response parser instead of matching its source text. + * + * Every other Hugging Face spec asserts on the emitted Python as a string, so a + * parser that raises at runtime still passes them (review feedback on #8617). + * This spec renders the operator, lifts `_parse_response` and its helpers out of + * the result, and runs them under a real interpreter against the response shapes + * providers actually return — valid, empty, missing, null and wrong-typed. + * + * The contract asserted here: `_parse_response` always returns a string and + * never raises, because an uncaught exception is caught by the per-row handler + * and replaces the cell with "Request failed" instead of the raw body. + */ +class HuggingFaceParseBehaviorSpec extends AnyFlatSpec with Matchers { + + private val mapper = new ObjectMapper() + + private def makeCtx(task: EncodableString): CodegenContext = + CodegenContext( + hfApiToken = "token", + modelId = "Qwen/Qwen2.5-72B-Instruct", + promptColumn = "prompt", + resultColumn = "hf_response", + task = task, + systemPrompt = "You are a helpful assistant.", + safeMaxTokens = 256, + safeTemp = 0.7 + ) + + /** Same resolution order as PythonCodeRawInvalidTextSpec: udf.conf, then PATH. */ + private lazy val pythonExe: Option[String] = { + def fromConfig: Option[String] = + Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption + .orElse(Try(ConfigFactory.load()).toOption) + .flatMap(c => Try(c.getConfig("python").getString("path")).toOption) + .map(_.trim) + .filter(_.nonEmpty) + + def isRunnable(exe: String): Boolean = + Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption + .exists { p => + if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + + (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(isRunnable) + } + + private lazy val probe: Path = { + val script = + scala.io.Source.fromInputStream( + getClass.getResourceAsStream("/python/hf_parse_probe.py"), + "UTF-8" + ) + val body = + try script.mkString + finally script.close() + val file = Files.createTempFile("hf_parse_probe", ".py") + Files.write(file, body.getBytes(StandardCharsets.UTF_8)) + file.toFile.deleteOnExit() + file + } + + /** Runs the generated parser for `task` over `bodies`; returns one outcome each. */ + private def parseAll(codegen: TaskCodegen, task: String, bodies: Seq[String]): Seq[String] = { + val exe = pythonExe.getOrElse(cancel("no python interpreter available to run the probe")) + val request = mapper.createObjectNode() + request.put("source", HuggingFaceCodegenBase.render(makeCtx(task), codegen)) + request.put("task", task) + val arr = request.putArray("bodies") + bodies.foreach(b => arr.add(mapper.readTree(b))) + + val process = new ProcessBuilder(exe, probe.toString).redirectErrorStream(false).start() + process.getOutputStream.write(mapper.writeValueAsBytes(request)) + process.getOutputStream.close() + val out = new String(process.getInputStream.readAllBytes(), StandardCharsets.UTF_8) + val err = new String(process.getErrorStream.readAllBytes(), StandardCharsets.UTF_8) + if (!process.waitFor(60, TimeUnit.SECONDS)) { + process.destroyForcibly(); fail("probe timed out") + } + withClue(s"probe stderr: $err\nprobe stdout: $out\n") { process.exitValue() shouldBe 0 } Review Comment: Confirmed, and it was a real hang rather than a theoretical one: with the probe made to sleep, the old code was still running after 90s and never reached its own 60s timeout. I went with redirecting stdout and stderr to temp files rather than moving the wait before the reads, that swaps this deadlock for the opposite one, since a child that writes past the pipe buffer would block on write and never exit, timing out a probe that was working fine. With files, neither side can block and the timeout is always reached. Verified with the same hung probe: it now fails in ~5s with "probe timed out". -- 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]
