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:53 UTC
(texera) branch gh-readonly-queue/main/pr-8668-e5c85111e1b026d0f6d726f99a895e1784f5c9d5 created (now 3ae15e6428)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-8668-e5c85111e1b026d0f6d726f99a895e1784f5c9d5
in repository https://gitbox.apache.org/repos/asf/texera.git
at 3ae15e6428 fix(amber): surface region launch failures instead of hanging (#8668)
This branch includes the following new commits:
new 3ae15e6428 fix(amber): surface region launch failures instead of hanging (#8668)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: fix(amber): surface region launch failures instead of hanging (#8668)
Posted by gi...@apache.org.
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")