You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/10/02 20:45:54 UTC
(texera) 01/01: fix(amber): surface region launch failures instead of hanging (#8668)
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 <12...@users.noreply.github.com>
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 <su...@suryaacharans-MacBook-Pro.local>
Co-authored-by: Claude Opus 5.5 (1M context) <no...@anthropic.com>
---
.../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")