Skip to content

Commit 8803d08

Browse files
Ma77BallMa77Ball
authored
fix(test): isolate e2e suites by unique workflow/execution id (apache#5888)
### What changes were proposed in this PR? - Add `TestUtils.workflowContext(id, settings)` that sets both `workflowId` and `executionId` to `id`, and make the DB fixtures plus `setUp`/`cleanupWorkflowExecutionData` take an `id` (the user email is derived from `id` to avoid the unique-email collision). - Give each materializing e2e suite a distinct id so concurrent suites no longer share an Iceberg result keyspace or DB rows: DataProcessingSpec=1, PauseSpec=2, ReconfigurationSpec=3, ReconfigurationIntegrationSpec=4. - Test-only change, no production code; BatchSizePropagationSpec and CheckpointSpec are untouched because they do not materialize results. ### Any related issues, documentation, discussions? Closes: apache#5887 ### How was this PR tested? - `sbt "WorkflowExecutionService/Test/compile"` compiles clean on Java 17. - Run the previously-flaky suite against the integration services (Postgres test DB + MinIO/S3 + Iceberg catalog, as in the `build / amber` CI job): `sbt "WorkflowExecutionService/testOnly *DataProcessingSpec"`; expect all DataProcessingSpec tests green with no CommitFailedException flake. - The timing-dependent flake could not be reproduced locally (no MinIO/Iceberg env), so final verification is the `build / amber` CI job staying green across re-runs. ### Was this PR authored or co-authored using generative AI tooling? Co-authored with Claude Opus 4.8 in compliance with ASF
1 parent 5a3ddcc commit 8803d08

5 files changed

Lines changed: 74 additions & 45 deletions

File tree

‎amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ import org.apache.texera.amber.clustering.SingleNodeListener
2828
import org.apache.texera.amber.core.executor.{OpExecInitInfo, OpExecWithCode}
2929
import org.apache.texera.amber.core.tuple.Tuple
3030
import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
31-
import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
31+
import org.apache.texera.amber.core.workflow.PortIdentity
3232
import org.apache.texera.amber.engine.architecture.controller.{
3333
ControllerConfig,
3434
ExecutionStateUpdate
@@ -80,14 +80,15 @@ class ReconfigurationIntegrationSpec
8080
implicit val timeout: Timeout = Timeout(5.seconds)
8181

8282
val logger = Logger("ReconfigurationIntegrationSpecLogger")
83-
val ctx = new WorkflowContext()
83+
private val specId = 4
84+
val ctx = TestUtils.workflowContext(specId)
8485

8586
override protected def beforeEach(): Unit = {
86-
setUpWorkflowExecutionData()
87+
setUpWorkflowExecutionData(specId)
8788
}
8889

8990
override protected def afterEach(): Unit = {
90-
cleanupWorkflowExecutionData()
91+
cleanupWorkflowExecutionData(specId)
9192
}
9293

9394
override def beforeAll(): Unit = {
@@ -117,12 +118,12 @@ class ReconfigurationIntegrationSpec
117118
*/
118119
private def warmupOnce(): Unit = {
119120
val warmupCap = Duration.fromSeconds(10)
120-
setUpWorkflowExecutionData()
121+
setUpWorkflowExecutionData(specId)
121122
var client: AmberClient = null
122123
try {
123124
val src = new TextInputSourceOpDesc()
124125
src.textInput = "warmup"
125-
val warmupCtx = new WorkflowContext()
126+
val warmupCtx = TestUtils.workflowContext(specId)
126127
val workflow = buildWorkflow(List(src), List.empty, warmupCtx)
127128
client = new AmberClient(
128129
system,
@@ -150,7 +151,7 @@ class ReconfigurationIntegrationSpec
150151
try client.shutdown()
151152
catch { case _: Throwable => () }
152153
}
153-
cleanupWorkflowExecutionData()
154+
cleanupWorkflowExecutionData(specId)
154155
}
155156
}
156157

‎amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -66,21 +66,24 @@ class DataProcessingSpec
6666

6767
implicit val timeout: Timeout = Timeout(5.seconds)
6868

69-
val workflowContext: WorkflowContext = new WorkflowContext()
69+
private val specId = 1
7070

71-
val materializedWorkflowContext: WorkflowContext = new WorkflowContext(
72-
workflowSettings = WorkflowSettings(
71+
val workflowContext: WorkflowContext = TestUtils.workflowContext(specId)
72+
73+
val materializedWorkflowContext: WorkflowContext = TestUtils.workflowContext(
74+
specId,
75+
WorkflowSettings(
7376
dataTransferBatchSize = 400,
7477
executionMode = ExecutionMode.MATERIALIZED
7578
)
7679
)
7780

7881
override protected def beforeEach(): Unit = {
79-
setUpWorkflowExecutionData()
82+
setUpWorkflowExecutionData(specId)
8083
}
8184

8285
override protected def afterEach(): Unit = {
83-
cleanupWorkflowExecutionData()
86+
cleanupWorkflowExecutionData(specId)
8487
}
8588

8689
override def beforeAll(): Unit = {

‎amber/src/test/scala/org/apache/texera/amber/engine/e2e/PauseSpec.scala‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ import org.apache.pekko.util.Timeout
2525
import com.twitter.util.{Await, Duration, Promise}
2626
import com.typesafe.scalalogging.Logger
2727
import org.apache.texera.amber.clustering.SingleNodeListener
28-
import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
28+
import org.apache.texera.amber.core.workflow.PortIdentity
2929
import org.apache.texera.amber.engine.architecture.controller.{
3030
ControllerConfig,
3131
ExecutionStateUpdate
@@ -70,12 +70,14 @@ class PauseSpec
7070

7171
val logger = Logger("PauseSpecLogger")
7272

73+
private val specId = 2
74+
7375
override protected def beforeEach(): Unit = {
74-
setUpWorkflowExecutionData()
76+
setUpWorkflowExecutionData(specId)
7577
}
7678

7779
override protected def afterEach(): Unit = {
78-
cleanupWorkflowExecutionData()
80+
cleanupWorkflowExecutionData(specId)
7981
}
8082

8183
override def beforeAll(): Unit = {
@@ -95,7 +97,7 @@ class PauseSpec
9597
links: List[LogicalLink]
9698
): Unit = {
9799
val workflow =
98-
TestUtils.buildWorkflow(operators, links, new WorkflowContext())
100+
TestUtils.buildWorkflow(operators, links, TestUtils.workflowContext(specId))
99101
val client =
100102
new AmberClient(
101103
system,

‎amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import org.apache.texera.amber.clustering.SingleNodeListener
2727
import org.apache.texera.amber.core.executor.OpExecInitInfo
2828
import org.apache.texera.amber.core.tuple.Tuple
2929
import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
30-
import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
30+
import org.apache.texera.amber.core.workflow.PortIdentity
3131
import org.apache.texera.amber.engine.common.AmberRuntime
3232
import org.apache.texera.amber.engine.e2e.TestUtils.{
3333
cleanupWorkflowExecutionData,
@@ -60,14 +60,15 @@ class ReconfigurationSpec
6060
implicit val timeout: Timeout = Timeout(5.seconds)
6161

6262
val logger = Logger("ReconfigurationSpecLogger")
63-
val ctx = new WorkflowContext()
63+
private val specId = 3
64+
val ctx = TestUtils.workflowContext(specId)
6465

6566
override protected def beforeEach(): Unit = {
66-
setUpWorkflowExecutionData()
67+
setUpWorkflowExecutionData(specId)
6768
}
6869

6970
override protected def afterEach(): Unit = {
70-
cleanupWorkflowExecutionData()
71+
cleanupWorkflowExecutionData(specId)
7172
}
7273

7374
override def beforeAll(): Unit = {

‎amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala‎

Lines changed: 47 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,12 @@ import org.apache.texera.amber.core.executor.OpExecInitInfo
2626
import org.apache.texera.amber.core.storage.DocumentFactory
2727
import org.apache.texera.amber.core.storage.model.VirtualDocument
2828
import org.apache.texera.amber.core.tuple.Tuple
29-
import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, OperatorIdentity}
30-
import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
29+
import org.apache.texera.amber.core.virtualidentity.{
30+
ExecutionIdentity,
31+
OperatorIdentity,
32+
WorkflowIdentity
33+
}
34+
import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext, WorkflowSettings}
3135
import org.apache.texera.amber.engine.architecture.controller.{
3236
ControllerConfig,
3337
ExecutionStateUpdate,
@@ -66,6 +70,22 @@ import org.apache.texera.workflow.{LogicalLink, WorkflowCompiler}
6670

6771
object TestUtils {
6872

73+
/**
74+
* A WorkflowContext whose workflow- and execution-id are both `id`. Each e2e
75+
* suite passes a distinct id so its results land in a disjoint storage
76+
* keyspace (`vfs:///wid/{id}/eid/{id}/...`) and disjoint DB rows, letting the
77+
* suites run concurrently without colliding on the shared Iceberg catalog.
78+
*/
79+
def workflowContext(
80+
id: Int,
81+
workflowSettings: WorkflowSettings = WorkflowSettings()
82+
): WorkflowContext =
83+
new WorkflowContext(
84+
workflowId = WorkflowIdentity(id.toLong),
85+
executionId = ExecutionIdentity(id.toLong),
86+
workflowSettings = workflowSettings
87+
)
88+
6989
def buildWorkflow(
7090
operators: List[LogicalOp],
7191
links: List[LogicalLink],
@@ -186,53 +206,55 @@ object TestUtils {
186206
)
187207
}
188208

189-
val testUser: User = {
209+
// All fixture rows for one suite share `id` as uid/wid/vid/eid; the email is
210+
// derived from it so concurrent suites don't collide on the unique email key.
211+
def testUser(id: Int): User = {
190212
val user = new User
191-
user.setUid(Integer.valueOf(1))
192-
user.setName("test_user")
213+
user.setUid(Integer.valueOf(id))
214+
user.setName(s"test_user_$id")
193215
user.setRole(UserRoleEnum.ADMIN)
194216
user.setPassword("123")
195-
user.setEmail("test_user@test.com")
217+
user.setEmail(s"test_user_$id@test.com")
196218
user
197219
}
198220

199-
val testWorkflowEntry: WorkflowPojo = {
221+
def testWorkflowEntry(id: Int): WorkflowPojo = {
200222
val workflow = new WorkflowPojo
201223
workflow.setName("test workflow")
202-
workflow.setWid(Integer.valueOf(1))
224+
workflow.setWid(Integer.valueOf(id))
203225
workflow.setContent("test workflow content")
204226
workflow.setDescription("test description")
205227
workflow
206228
}
207229

208-
val testWorkflowVersionEntry: WorkflowVersion = {
230+
def testWorkflowVersionEntry(id: Int): WorkflowVersion = {
209231
val workflowVersion = new WorkflowVersion
210-
workflowVersion.setWid(Integer.valueOf(1))
211-
workflowVersion.setVid(Integer.valueOf(1))
232+
workflowVersion.setWid(Integer.valueOf(id))
233+
workflowVersion.setVid(Integer.valueOf(id))
212234
workflowVersion.setContent("test version content")
213235
workflowVersion
214236
}
215237

216-
val testWorkflowExecutionEntry: WorkflowExecutions = {
238+
def testWorkflowExecutionEntry(id: Int): WorkflowExecutions = {
217239
val workflowExecution = new WorkflowExecutions
218-
workflowExecution.setEid(Integer.valueOf(1))
219-
workflowExecution.setVid(Integer.valueOf(1))
220-
workflowExecution.setUid(Integer.valueOf(1))
240+
workflowExecution.setEid(Integer.valueOf(id))
241+
workflowExecution.setVid(Integer.valueOf(id))
242+
workflowExecution.setUid(Integer.valueOf(id))
221243
workflowExecution.setStatus(3.toByte)
222244
workflowExecution.setEnvironmentVersion("test engine")
223245
workflowExecution
224246
}
225247

226-
def setUpWorkflowExecutionData(): Unit = {
248+
def setUpWorkflowExecutionData(id: Int): Unit = {
227249
val dslConfig = SqlServer.getInstance().context.configuration()
228250
val userDao = new UserDao(dslConfig)
229251
val workflowDao = new WorkflowDao(dslConfig)
230252
val workflowExecutionsDao = new WorkflowExecutionsDao(dslConfig)
231253
val workflowVersionDao = new WorkflowVersionDao(dslConfig)
232-
userDao.insert(testUser)
233-
workflowDao.insert(testWorkflowEntry)
234-
workflowVersionDao.insert(testWorkflowVersionEntry)
235-
workflowExecutionsDao.insert(testWorkflowExecutionEntry)
254+
userDao.insert(testUser(id))
255+
workflowDao.insert(testWorkflowEntry(id))
256+
workflowVersionDao.insert(testWorkflowVersionEntry(id))
257+
workflowExecutionsDao.insert(testWorkflowExecutionEntry(id))
236258
}
237259

238260
/**
@@ -318,16 +340,16 @@ object TestUtils {
318340
result
319341
}
320342

321-
def cleanupWorkflowExecutionData(): Unit = {
343+
def cleanupWorkflowExecutionData(id: Int): Unit = {
322344
val dslConfig = SqlServer.getInstance().context.configuration()
323345
val userDao = new UserDao(dslConfig)
324346
val workflowDao = new WorkflowDao(dslConfig)
325347
val workflowExecutionsDao = new WorkflowExecutionsDao(dslConfig)
326348
val workflowVersionDao = new WorkflowVersionDao(dslConfig)
327-
workflowExecutionsDao.deleteById(1)
328-
workflowVersionDao.deleteById(1)
329-
workflowDao.deleteById(1)
330-
userDao.deleteById(1)
349+
workflowExecutionsDao.deleteById(id)
350+
workflowVersionDao.deleteById(id)
351+
workflowDao.deleteById(id)
352+
userDao.deleteById(id)
331353
}
332354

333355
}

0 commit comments

Comments
 (0)