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/07/03 18:28:26 UTC
(texera) branch main updated: fix(test): stabilize flaky ReconfigurationIntegrationSpec pause race (#5915)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new b30f928dc4 fix(test): stabilize flaky ReconfigurationIntegrationSpec pause race (#5915)
b30f928dc4 is described below
commit b30f928dc48e4873d594737e3ccd4d2ffcb7e5ff
Author: Matthew B. <mg...@uci.edu>
AuthorDate: Fri Jul 3 11:19:24 2026 -0700
fix(test): stabilize flaky ReconfigurationIntegrationSpec pause race (#5915)
### What changes were proposed in this PR?
Makes `ReconfigurationIntegrationSpec` deterministic instead of flaky.
**The problem:** the test starts a workflow, pauses it, swaps operator
code, then resumes, and it relied on the *amount of data* to keep the
workflow alive long enough to pause. That is a race with no safe
setting:
- too little data: the workflow finishes before `pause` lands, so PAUSED
is never reached and the test times out;
- too much data (the `medium` CSV is 100k rows through 1-2 Python UDFs):
the workflow can't reach COMPLETED within the 1-minute wait, so it times
out.
Python throughput varies run to run, so either way it was flaky.
**The fix:** replace the data-volume race with a slow source.
`slowRegionSourceOpDesc` emits only 30 rows, 0.25s apart (~7.5s of
steady production):
- the workflow is provably still running for ~7.5s, so `pause` lands
reliably (an explicit running window instead of racing throughput);
- only 30 tiny tuples flow, so completion is near-instant and the
timeout is never stressed;
- the Python source generator suspends on pause and resumes the same
frame, so the remaining rows flow through the reconfigured UDF(s) after
resume and the assertions still hold.
Also kept from earlier: a `TextInput -> Python UDF` warmup in
`beforeAll` (speeds worker cold-start) and widened control-command
awaits (now just safety bounds, no longer the fix). Test 2 is unified
onto the `Region` column.
### Any related issues, documentation, discussions?
The reconfiguration mechanism itself is also covered by the pure-Scala
`ReconfigurationSpec` (Java operators, stable `amber` job); this spec
adds the Python-executor path.
### How was this PR tested?
- `DAO/compile` + `WorkflowExecutionService/Test/compile` succeed;
scalafmt clean.
- The `amber-integration` CI job runs `ReconfigurationIntegrationSpec`
end to end (Python workers + postgres + lakekeeper + Iceberg), which is
the real signal for this flake.
### Was this PR authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.8 in compliance with ASF
---------
Signed-off-by: Matthew B. <mg...@uci.edu>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../e2e/ReconfigurationIntegrationSpec.scala | 44 +++++++++++++---------
.../apache/texera/amber/engine/e2e/TestUtils.scala | 12 +++---
.../texera/amber/operator/TestOperators.scala | 20 ++++++++++
3 files changed, 54 insertions(+), 22 deletions(-)
diff --git a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala
index 5d2ed7e5e4..b4545a19b1 100644
--- a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala
+++ b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala
@@ -105,26 +105,31 @@ class ReconfigurationIntegrationSpec
}
/**
- * Run a trivial pure-Scala workflow (TextInput → terminal) once before the
- * timed tests start, so the first 5-second `startWorkflow` await in
- * [[TestUtils.shouldReconfigure]] doesn't have to absorb JVM JIT
- * warmup, pekko dispatcher first-touch, and `RegionExecutionCoordinator`
- * class loading.
- *
- * Hard-capped at 10 seconds total, defensively wrapped: if warmup itself
- * times out or throws, log and continue — the existing `Retries` mixin
- * still backs up individual test cases. This ensures warmup can never
- * hang the suite.
+ * Runs a TextInput -> Python UDF workflow once before the timed tests so
+ * Python worker cold-start is paid here, not inside a timed test. Capped and
+ * wrapped so warmup can never fail or hang the suite.
*/
private def warmupOnce(): Unit = {
- val warmupCap = Duration.fromSeconds(10)
+ val warmupCap = Duration.fromSeconds(60)
setUpWorkflowExecutionData(specId)
var client: AmberClient = null
try {
val src = new TextInputSourceOpDesc()
src.textInput = "warmup"
+ val udf = TestOperators.pythonOpDesc()
val warmupCtx = TestUtils.workflowContext(specId)
- val workflow = buildWorkflow(List(src), List.empty, warmupCtx)
+ val workflow = buildWorkflow(
+ List(src, udf),
+ List(
+ LogicalLink(
+ src.operatorIdentifier,
+ PortIdentity(),
+ udf.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ warmupCtx
+ )
client = new AmberClient(
system,
workflow.context,
@@ -166,8 +171,13 @@ class ReconfigurationIntegrationSpec
): Map[OperatorIdentity, List[Tuple]] =
TestUtils.shouldReconfigure(system, ctx, operators, links, targetOps, newOpExecInitInfo)
+ // Small source that emits slowly (30 rows, 0.25s apart) so a pause lands
+ // mid-run and the workflow still completes quickly after resume.
+ private def slowSource() =
+ TestOperators.slowRegionSourceOpDesc(numTuple = 30, delaySeconds = 0.25)
+
"Engine" should "be able to modify a python UDF worker in workflow" in {
- val sourceOpDesc = TestOperators.smallCsvScanOpDesc()
+ val sourceOpDesc = slowSource()
val udfOpDesc = TestOperators.pythonOpDesc()
val code = """
|from pytexera import *
@@ -198,7 +208,7 @@ class ReconfigurationIntegrationSpec
}
"Engine" should "propagate reconfiguration through a source operator in workflow" in {
- val sourceOpDesc = TestOperators.pythonSourceOpDesc(10000)
+ val sourceOpDesc = slowSource()
val udfOpDesc = TestOperators.pythonOpDesc()
val code = """
|from pytexera import *
@@ -206,7 +216,7 @@ class ReconfigurationIntegrationSpec
|class ProcessTupleOperator(UDFOperatorV2):
| @overrides
| def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]:
- | tuple_['field_1'] = tuple_['field_1'] + '_reconfigured'
+ | tuple_['Region'] = tuple_['Region'] + '_reconfigured'
| yield tuple_
|""".stripMargin
val result = shouldReconfigure(
@@ -223,12 +233,12 @@ class ReconfigurationIntegrationSpec
OpExecWithCode(code, "python")
)
assert(result(udfOpDesc.operatorIdentifier).exists { t =>
- t.getField("field_1").asInstanceOf[String].contains("_reconfigured")
+ t.getField("Region").asInstanceOf[String].contains("_reconfigured")
})
}
"Engine" should "be able to modify two python UDFs in workflow" in {
- val sourceOpDesc = TestOperators.smallCsvScanOpDesc()
+ val sourceOpDesc = slowSource()
val udfOpDesc1 = TestOperators.pythonOpDesc()
val udfOpDesc2 = TestOperators.pythonOpDesc()
val code = """
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala
index ac71483a5d..d001c4471c 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala
@@ -301,6 +301,8 @@ object TestUtils {
ControllerConfig.default,
error => {}
)
+ // Timeout for control-command acks (start/pause/reconfigure/resume).
+ val commandTimeout = Duration.fromSeconds(30)
val completion = Promise[Unit]()
var result: Map[OperatorIdentity, List[Tuple]] = null
client.registerCallback[ExecutionStateUpdate](evt => {
@@ -311,14 +313,14 @@ object TestUtils {
})
Await.result(
client.controllerInterface.startWorkflow(EmptyRequest(), ()),
- Duration.fromSeconds(5)
+ commandTimeout
)
val pausedReached = stateReached(client, PAUSED)
Await.result(
client.controllerInterface.pauseWorkflow(EmptyRequest(), ()),
- Duration.fromSeconds(5)
+ commandTimeout
)
- Await.result(pausedReached, Duration.fromSeconds(10))
+ Await.result(pausedReached, commandTimeout)
val physicalOps = targetOps.flatMap(op =>
workflow.physicalPlan.getPhysicalOpsOfLogicalOp(op.operatorIdentifier)
)
@@ -330,11 +332,11 @@ object TestUtils {
),
()
),
- Duration.fromSeconds(5)
+ commandTimeout
)
Await.result(
client.controllerInterface.resumeWorkflow(EmptyRequest(), ()),
- Duration.fromSeconds(5)
+ commandTimeout
)
Await.result(completion, Duration.fromMinutes(1))
result
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/TestOperators.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/TestOperators.scala
index b3c3873595..69783b5e1f 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/TestOperators.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/TestOperators.scala
@@ -181,4 +181,24 @@ object TestOperators {
udf
}
+ // Emits `numTuple` rows with a "Region" column, sleeping `delaySeconds`
+ // between rows.
+ def slowRegionSourceOpDesc(numTuple: Int, delaySeconds: Double): PythonUDFSourceOpDescV2 = {
+ val udf = new PythonUDFSourceOpDescV2()
+ udf.workers = 1
+ udf.columns = List(new Attribute("Region", AttributeType.STRING))
+ udf.code = s"""
+ |from pytexera import *
+ |import time
+ |
+ |class UDFSourceOperator(UDFSourceOperator):
+ | @overrides
+ | def produce(self) -> Iterator[Union[TupleLike, TableLike, None]]:
+ | for i in range($numTuple):
+ | time.sleep($delaySeconds)
+ | yield {'Region': 'Asia'}
+ |""".stripMargin
+ udf
+ }
+
}