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-8668-e5c85111e1b026d0f6d726f99a895e1784f5c9d5 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 3ae15e6428442ae77ed8b113392f122d746a87b5 Author: Suryaa Charan <[email protected]> AuthorDate: Fri Oct 2 20:45:29 2026 +0000 fix(amber): surface region launch failures instead of hanging (#8668) ### What changes were proposed in this PR? A worker failing to initialize in a region's second (non-dependee) phase, e.g. a Python UDF after a HashJoin, used to leave the execution `RUNNING` forever with no error. See the issue for the full chain. - `WorkflowExecutionManager.advanceRegionExecutions`: keep the `syncStatusAndTransitionRegionExecutionPhase()` futures instead of dropping them in a `foreach`. The advance future now fails if any sync fails, so the existing `.onFailure → FatalError` in `AdvanceRegionExecutionsHandler` reports it. The rest of the method moved unchanged into `advanceToNextRegions`. - `RegionExecutionManager`: a failed launch RPC (`initializeExecutor`, `assignPort`, `openExecutor`, `startWorker`) is wrapped as `WorkflowRuntimeException` with the target worker ID, so the `FatalError` names the failing worker and operator (review feedback). - `AsyncRPCClient.logControlReply`: log the `ControlError` message, details and stack trace. It isn't a `Throwable`, so the logger silently dropped it. | | Before | After | |---|---|---| | UDF missing `from pytexera import *` after a join | hangs, all 0 tuples, no error | fails in ~1s with `name 'UDFTableOperator' is not defined`, attributed to the UDF worker | | Log line | `received error from …` | `received error from … (controlID: N): <message>` | Behavior note: the future returned by `advanceRegionExecutions` now also waits on, and fails with, the syncs it triggers. `AdvanceRegionExecutionsHandler` doesn't wait on it. `StartWorkflowHandler` has no unfinished regions at start, so nothing changes there. ### Any related issues, documentation, discussions? Closes #8667 ### How was this PR tested? Added tests. Each new failing-case test fails before the fix and passes after it. - `WorkflowExecutionManagerSpec`, 2 new tests: a region whose probe op has a dependee port, followed by a UDF op. - The UDF's `initializeExecutor` returns a `ControlError` → the advance fails with that message. - The UDF initializes fine → it is started. - `DataProcessingSpec`, 1 new e2e test on csv, csv → join → UDF: a Java UDF whose code doesn't compile, so the worker fails in `initializeExecutor`, the same step as a Python UDF with an import error. - Before the fix it times out after 1 min; after, it fails in ~0.5s, naming the UDF worker. - It uses a Java UDF because CI's amber job has no Python environment. - `RegionExecutionManagerTestSupport`: `CoordinatorRpcProbe` takes an optional `initializeExecutorResponse`. ``` sbt 'WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.scheduling.WorkflowExecutionManagerSpec' sbt 'WorkflowExecutionService/testOnly org.apache.texera.amber.engine.e2e.DataProcessingSpec -- -z "fail, not hang"' ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: suryaacharan <[email protected]> Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]> --- .../scheduling/RegionExecutionManager.scala | 81 ++++++++------ .../scheduling/WorkflowExecutionManager.scala | 17 ++- .../amber/engine/common/rpc/AsyncRPCClient.scala | 7 +- .../RegionExecutionManagerTestSupport.scala | 13 ++- .../scheduling/WorkflowExecutionManagerSpec.scala | 123 ++++++++++++++++++++- .../amber/engine/e2e/DataProcessingSpec.scala | 53 +++++++++ 6 files changed, 255 insertions(+), 39 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala index c7371f8c47..e884a6ba30 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala @@ -21,6 +21,7 @@ package org.apache.texera.amber.engine.architecture.scheduling import org.apache.pekko.pattern.gracefulStop import com.twitter.util.{Future, JavaTimer, Promise, Return, Throw, Timer} +import org.apache.texera.amber.core.WorkflowRuntimeException import org.apache.texera.amber.core.state.State import org.apache.texera.amber.core.storage.{DocumentFactory, RepositoryMountManager, VFSURIFactory} import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity @@ -460,6 +461,17 @@ class RegionExecutionManager( ) } + /** + * Tags a failed worker RPC with the worker it targeted, so the `FatalError` surfaced for a + * region that cannot be launched can point at the failing operator/worker. + */ + private def attributeFailureTo[T](workerId: ActorVirtualIdentity)(rpc: Future[T]): Future[T] = + rpc.rescue { + case err: WorkflowRuntimeException if err.relatedWorkerId.isDefined => Future.exception(err) + case err => + Future.exception(new WorkflowRuntimeException(err.getMessage, err, Some(workerId))) + } + private def initExecutors( operators: Set[PhysicalOp], resourceConfig: ResourceConfig @@ -471,14 +483,16 @@ class RegionExecutionManager( val workerConfigs = resourceConfig.operatorConfigs(physicalOp.id).workerConfigs val opExecInitInfo = physicalOp.executableOpExecInitInfo workerConfigs.map(_.workerId).map { workerId => - asyncRPCClient.workerInterface.initializeExecutor( - InitializeExecutorRequest( - workerConfigs.length, - opExecInitInfo, - physicalOp.isSourceOperator, - loopStartPortUris - ), - asyncRPCClient.mkContext(workerId) + attributeFailureTo(workerId)( + asyncRPCClient.workerInterface.initializeExecutor( + InitializeExecutorRequest( + workerConfigs.length, + opExecInitInfo, + physicalOp.isSourceOperator, + loopStartPortUris + ), + asyncRPCClient.mkContext(workerId) + ) ) } }) @@ -557,15 +571,17 @@ class RegionExecutionManager( case (globalPortId, (storageUris, partitionings, schema)) => resourceConfig.operatorConfigs(globalPortId.opId).workerConfigs.map(_.workerId).map { workerId => - asyncRPCClient.workerInterface.assignPort( - AssignPortRequest( - globalPortId.portId, - globalPortId.input, - schema.toRawSchema, - storageUris, - partitionings - ), - asyncRPCClient.mkContext(workerId) + attributeFailureTo(workerId)( + asyncRPCClient.workerInterface.assignPort( + AssignPortRequest( + globalPortId.portId, + globalPortId.input, + schema.toRawSchema, + storageUris, + partitionings + ), + asyncRPCClient.mkContext(workerId) + ) ) } } @@ -593,8 +609,10 @@ class RegionExecutionManager( workflowExecution.getRegionExecution(region.id).getOperatorExecution(opId).getWorkerIds ) .map { workerId => - asyncRPCClient.workerInterface - .openExecutor(EmptyRequest(), asyncRPCClient.mkContext(workerId)) + attributeFailureTo(workerId)( + asyncRPCClient.workerInterface + .openExecutor(EmptyRequest(), asyncRPCClient.mkContext(workerId)) + ) } .toSeq ) @@ -620,18 +638,19 @@ class RegionExecutionManager( .getOperatorExecution(opId) .getWorkerIds .map { workerId => - asyncRPCClient.workerInterface - .startWorker(EmptyRequest(), asyncRPCClient.mkContext(workerId)) - .map(resp => - // Update worker state, ordered by the worker's logical state version - // (not arrival time) so this RUNNING snapshot cannot clobber a later - // COMPLETED if the response arrives after the worker has finished. - workflowExecution - .getRegionExecution(region.id) - .getOperatorExecution(opId) - .getWorkerExecution(workerId) - .updateState(resp.stateVersion, resp.state) - ) + attributeFailureTo(workerId)( + asyncRPCClient.workerInterface + .startWorker(EmptyRequest(), asyncRPCClient.mkContext(workerId)) + ).map(resp => + // Update worker state, ordered by the worker's logical state version + // (not arrival time) so this RUNNING snapshot cannot clobber a later + // COMPLETED if the response arrives after the worker has finished. + workflowExecution + .getRegionExecution(region.id) + .getOperatorExecution(opId) + .getWorkerExecution(workerId) + .updateState(resp.stateVersion, resp.state) + ) } } .toSeq diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManager.scala index b736cd171c..aabfaba2fb 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManager.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManager.scala @@ -110,9 +110,22 @@ class WorkflowExecutionManager( val unfinishedRegionManagers = regionExecutionManagers.values.filter(!_.isCompleted).toSeq - // Trigger sync for each unfinished region. - unfinishedRegionManagers.foreach(_.syncStatusAndTransitionRegionExecutionPhase()) + // Trigger sync for each unfinished region. The returned futures must not be dropped: a sync + // may launch the next phase of a region, and a failure there (e.g., a worker failing to + // initialize its executor) would otherwise be lost, leaving the execution RUNNING forever + // with no error reported to the client. + val syncFutures = + unfinishedRegionManagers.map(_.syncStatusAndTransitionRegionExecutionPhase()) + Future + .collect(syncFutures :+ advanceToNextRegions(actorService, unfinishedRegionManagers)) + .unit + } + + private def advanceToNextRegions( + actorService: PekkoActorService, + unfinishedRegionManagers: Seq[RegionExecutionManager] + ): Future[Unit] = { // Wait only for region termination futures (kill path), then re-run the advance loop. val terminationFutures = unfinishedRegionManagers.flatMap(_.getTerminationFutureOpt) if (terminationFutures.nonEmpty) { diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/common/rpc/AsyncRPCClient.scala b/amber/src/main/scala/org/apache/texera/amber/engine/common/rpc/AsyncRPCClient.scala index 88f6f7b350..2e70f25ce0 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/common/rpc/AsyncRPCClient.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/common/rpc/AsyncRPCClient.scala @@ -205,7 +205,12 @@ class AsyncRPCClient( ) ret.returnValue match { case err: ControlError => - logger.error(s"received error from $channelId", err) + // ControlError is not a Throwable, so pass its contents explicitly; otherwise the + // logger drops it and only "received error from ..." is printed. + logger.error( + s"received error from $channelId (controlID: ${ret.commandId}): " + + s"${err.errorMessage}\n${err.errorDetails}\n${err.stackTrace}" + ) case _ => } } else { diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala index 6840eaf9b7..7eb9a480b8 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala @@ -88,9 +88,14 @@ object RegionExecutionManagerTestSupport { * * Non-termination RPCs are completed immediately because these tests focus on termination * ordering. `endWorker` responses are controlled by `endWorkerResponse`, allowing each test to - * hold termination pending, fail an attempt, or allow it to succeed. + * hold termination pending, fail an attempt, or allow it to succeed. `initializeExecutor` + * responses are controlled by `initializeExecutorResponse`, allowing a test to simulate a worker + * whose executor fails to initialize (e.g., a Python UDF with an error at import time). */ - class CoordinatorRpcProbe(endWorkerResponse: WorkerRpcCall => Option[ControlReturn]) { + class CoordinatorRpcProbe( + endWorkerResponse: WorkerRpcCall => Option[ControlReturn], + initializeExecutorResponse: WorkerRpcCall => ControlReturn = _ => EmptyReturn() + ) { val calls: mutable.ArrayBuffer[WorkerRpcCall] = mutable.ArrayBuffer() val inputGateway = new NetworkInputGateway(COORDINATOR) val outputGateway = new NetworkOutputGateway(COORDINATOR, handleOutput) @@ -137,7 +142,9 @@ object RegionExecutionManagerTestSupport { private def immediateReturn(call: WorkerRpcCall): Option[ControlReturn] = { call.methodName match { - case InitializeExecutor | OpenExecutor => + case InitializeExecutor => + Some(initializeExecutorResponse(call)) + case OpenExecutor => Some(EmptyReturn()) case StartWorker => // RUNNING is the worker's 2nd transition (UNINITIALIZED -> READY -> RUNNING). diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManagerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManagerSpec.scala index 7ddc7d2967..141de24460 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManagerSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionManagerSpec.scala @@ -19,20 +19,32 @@ package org.apache.texera.amber.engine.architecture.scheduling +import com.twitter.util.Future import org.apache.pekko.actor.ActorSystem +import org.apache.texera.amber.core.WorkflowRuntimeException import org.apache.pekko.testkit.TestKit import org.apache.texera.amber.core.executor.OpExecInitInfo import org.apache.texera.amber.core.virtualidentity.{ + ActorVirtualIdentity, ExecutionIdentity, OperatorIdentity, PhysicalOpIdentity, WorkflowIdentity } -import org.apache.texera.amber.core.workflow.PhysicalOp +import org.apache.texera.amber.core.workflow.{InputPort, PhysicalOp, PortIdentity} import org.apache.texera.amber.engine.architecture.coordinator.CoordinatorConfig import org.apache.texera.amber.engine.architecture.coordinator.execution.WorkflowExecution -import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.{ + ControlError, + EmptyReturn, + ErrorLanguage +} import org.apache.texera.amber.engine.architecture.scheduling.RegionExecutionManagerTestSupport._ +import org.apache.texera.amber.engine.architecture.scheduling.config.{ + OperatorConfig, + ResourceConfig, + WorkerConfig +} import org.apache.texera.amber.engine.common.AmberRuntime import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpecLike @@ -150,6 +162,113 @@ class WorkflowExecutionManagerSpec assert(rpcProbe.startedWorkers.contains(secondWorkerId)) } + // Same failure as a Python UDF whose code is missing `from pytexera import *`. + private val udfInitError = "NameError: name 'UDFTableOperator' is not defined" + + /** + * Mirrors a region like HashJoin-probe + Python UDF: the probe op has a dependee input port, + * so the region first runs a dependee phase with only the probe, and launches the remaining + * operators (here, the UDF) only once the dependee port completes. That second launch is + * triggered from a later advance round. + * + * Runs round 1, marks the probe's dependee port completed, and returns the round-2 future + * together with the probe and the UDF worker id. + */ + private def runTwoPhaseRegion( + udfInitFails: Boolean + ): (Future[Unit], CoordinatorRpcProbe, ActorVirtualIdentity) = { + val probeOp = createSourceOp("probe-op").withInputPorts( + List( + InputPort(PortIdentity(0)), + InputPort(PortIdentity(1), dependencies = List(PortIdentity(0))) + ) + ) + val probeWorkerId = createWorkerId(probeOp) + val udfOp = createSourceOp("udf-op") + val udfWorkerId = createWorkerId(udfOp) + + val region = Region( + RegionIdentity(1), + physicalOps = Set(probeOp, udfOp), + physicalLinks = Set.empty, + resourceConfig = Some( + ResourceConfig( + operatorConfigs = Map( + probeOp.id -> OperatorConfig(List(WorkerConfig(probeWorkerId))), + udfOp.id -> OperatorConfig(List(WorkerConfig(udfWorkerId))) + ) + ) + ) + ) + + val workflowExecution = WorkflowExecution() + seedReusableWorkerExecution(workflowExecution, seedRegionId = 101, probeOp, probeWorkerId) + seedReusableWorkerExecution(workflowExecution, seedRegionId = 102, udfOp, udfWorkerId) + + val rpcProbe = new CoordinatorRpcProbe( + endWorkerResponse = _ => Some(EmptyReturn()), + initializeExecutorResponse = call => + if (udfInitFails && call.receiver == udfWorkerId) + ControlError( + errorMessage = udfInitError, + errorDetails = "", + stackTrace = "", + language = ErrorLanguage.PYTHON + ) + else EmptyReturn() + ) + val coordinator = createCoordinatorHarness() + registerLiveWorker(coordinator.actorRefService, probeWorkerId) + registerLiveWorker(coordinator.actorRefService, udfWorkerId) + + val workflowManager = new WorkflowExecutionManager( + workflowExecution, + CoordinatorConfig(None, None, None, None), + rpcProbe.asyncRPCClient + ) + workflowManager.schedule = Schedule(Map(0 -> Set(region))) + workflowManager.setupActorRefService(coordinator.actorRefService) + + // Round 1: only the dependee phase (the probe op) is launched. + await(workflowManager.advanceRegionExecutions(coordinator.actorService)) + assert(rpcProbe.initializedWorkers == Seq(probeWorkerId)) + assert(rpcProbe.startedWorkers == Seq(probeWorkerId)) + + // The probe's dependee port completes (e.g., the hash-join build side is fully read). + workflowExecution + .getRegionExecution(region.id) + .getOperatorExecution(probeOp.id) + .getWorkerExecution(probeWorkerId) + .getInputPortExecution(PortIdentity(0)) + .setCompleted() + + // Round 2 launches the non-dependee phase (the UDF). + ( + workflowManager.advanceRegionExecutions(coordinator.actorService), + rpcProbe, + udfWorkerId + ) + } + + it should "fail coordination when a worker fails to initialize in a later region phase" in { + val (round2, rpcProbe, udfWorkerId) = runTwoPhaseRegion(udfInitFails = true) + + val error = intercept[WorkflowRuntimeException](await(round2)) + assert(error.getMessage.contains(udfInitError)) + // The failing worker is carried so the surfaced FatalError can point at it. + assert(error.relatedWorkerId.contains(udfWorkerId)) + assert(rpcProbe.initializedWorkers.contains(udfWorkerId)) + assert(!rpcProbe.startedWorkers.contains(udfWorkerId)) + } + + it should "launch the later region phase when its workers initialize successfully" in { + val (round2, rpcProbe, udfWorkerId) = runTwoPhaseRegion(udfInitFails = false) + + await(round2) + assert(rpcProbe.initializedWorkers.contains(udfWorkerId)) + assert(rpcProbe.startedWorkers.contains(udfWorkerId)) + } + "Jumping to an operator's region" should "make the next scheduled region contain the target operator's region" in { val (first, second, _, schedule) = threeLevelSchedule() diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala index b6b815a120..43e92f7592 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala @@ -23,6 +23,7 @@ import org.apache.pekko.actor.{ActorSystem, Props} import org.apache.pekko.testkit.{ImplicitSender, TestKit} import org.apache.pekko.util.Timeout import org.apache.texera.amber.clustering.SingleNodeListener +import org.apache.texera.amber.core.WorkflowRuntimeException import org.apache.texera.amber.core.tuple.{AttributeType, Tuple} import org.apache.texera.amber.core.virtualidentity.OperatorIdentity import org.apache.texera.amber.core.workflow.{ @@ -41,6 +42,7 @@ import org.apache.texera.amber.engine.e2e.TestUtils.{ setUpWorkflowExecutionData } import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.amber.operator.udf.java.JavaUDFOpDesc import org.apache.texera.amber.operator.aggregate.AggregationFunction import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.flatspec.AnyFlatSpecLike @@ -302,6 +304,57 @@ class DataProcessingSpec executeWorkflow(workflow) } + /** + * csv -> join(build), csv -> join(probe) -> Java UDF. The UDF shares a region with the join's + * probe side, which has a dependee input port, so the UDF is only launched in the region's + * second (non-dependee) phase. A Java UDF compiles its code when the worker initializes its + * executor, so code that does not compile fails at the same point as a Python UDF with an + * import error, without needing a Python environment. + */ + private def joinThenJavaUdfWorkflow(udfCode: String): Workflow = { + val headerlessCsvOpDesc1 = TestOperators.headerlessSmallCsvScanOpDesc() + val headerlessCsvOpDesc2 = TestOperators.headerlessSmallCsvScanOpDesc() + val joinOpDesc = TestOperators.joinOpDesc("column-1", "column-1") + val javaUdfOpDesc = new JavaUDFOpDesc() + javaUdfOpDesc.retainInputColumns = true + javaUdfOpDesc.code = udfCode + buildWorkflow( + List(headerlessCsvOpDesc1, headerlessCsvOpDesc2, joinOpDesc, javaUdfOpDesc), + List( + LogicalLink( + headerlessCsvOpDesc1.operatorIdentifier, + PortIdentity(), + joinOpDesc.operatorIdentifier, + PortIdentity() + ), + LogicalLink( + headerlessCsvOpDesc2.operatorIdentifier, + PortIdentity(), + joinOpDesc.operatorIdentifier, + PortIdentity(1) + ), + LogicalLink( + joinOpDesc.operatorIdentifier, + PortIdentity(), + javaUdfOpDesc.operatorIdentifier, + PortIdentity() + ) + ), + workflowContext + ) + } + + "Engine" should "fail, not hang, when a udf after a join fails to initialize" in { + // The UDF code does not compile, so the worker fails to initialize its executor. The + // execution must fail with the compile error instead of hanging. + val error = intercept[WorkflowRuntimeException] { + executeWorkflow(joinThenJavaUdfWorkflow("public class JavaUDFOpExec { not valid java }")) + } + assert(error.getMessage.contains("Error at line")) + // The failing UDF worker is carried so the UI can attribute the error to it. + assert(error.relatedWorkerId.exists(_.name.contains("JavaUDFOpDesc"))) + } + "Engine" should "execute headerlessCsv->keyword workflow with MATERIALIZED mode" in { val headerlessCsvOpDesc = TestOperators.headerlessSmallCsvScanOpDesc() val keywordOpDesc = TestOperators.keywordSearchOpDesc("column-1", "Asia")
