fix(test): isolate e2e suites by unique workflow/execution id [release/v1.2 backport]

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

@@ -28,7 +28,7 @@ import org.apache.texera.amber.clustering.SingleNodeListener
 import org.apache.texera.amber.core.executor.{OpExecInitInfo, OpExecWithCode}
 import org.apache.texera.amber.core.tuple.Tuple
 import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
-import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
+import org.apache.texera.amber.core.workflow.PortIdentity
 import org.apache.texera.amber.engine.architecture.controller.{
   ControllerConfig,
   ExecutionStateUpdate
@@ -80,14 +80,15 @@ class ReconfigurationIntegrationSpec
   implicit val timeout: Timeout = Timeout(5.seconds)
 
   val logger = Logger("ReconfigurationIntegrationSpecLogger")
-  val ctx = new WorkflowContext()
+  private val specId = 4
+  val ctx = TestUtils.workflowContext(specId)
 
   override protected def beforeEach(): Unit = {
-    setUpWorkflowExecutionData()
+    setUpWorkflowExecutionData(specId)
   }
 
   override protected def afterEach(): Unit = {
-    cleanupWorkflowExecutionData()
+    cleanupWorkflowExecutionData(specId)
   }
 
   override def beforeAll(): Unit = {
@@ -117,12 +118,12 @@ class ReconfigurationIntegrationSpec
     */
   private def warmupOnce(): Unit = {
     val warmupCap = Duration.fromSeconds(10)
-    setUpWorkflowExecutionData()
+    setUpWorkflowExecutionData(specId)
     var client: AmberClient = null
     try {
       val src = new TextInputSourceOpDesc()
       src.textInput = "warmup"
-      val warmupCtx = new WorkflowContext()
+      val warmupCtx = TestUtils.workflowContext(specId)
       val workflow = buildWorkflow(List(src), List.empty, warmupCtx)
       client = new AmberClient(
         system,
@@ -150,7 +151,7 @@ class ReconfigurationIntegrationSpec
         try client.shutdown()
         catch { case _: Throwable => () }
       }
-      cleanupWorkflowExecutionData()
+      cleanupWorkflowExecutionData(specId)
     }
   }
 

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

@@ -72,21 +72,24 @@ class DataProcessingSpec
 
   implicit val timeout: Timeout = Timeout(5.seconds)
 
-  val workflowContext: WorkflowContext = new WorkflowContext()
+  private val specId = 1
 
-  val materializedWorkflowContext: WorkflowContext = new WorkflowContext(
-    workflowSettings = WorkflowSettings(
+  val workflowContext: WorkflowContext = TestUtils.workflowContext(specId)
+
+  val materializedWorkflowContext: WorkflowContext = TestUtils.workflowContext(
+    specId,
+    WorkflowSettings(
       dataTransferBatchSize = 400,
       executionMode = ExecutionMode.MATERIALIZED
     )
   )
 
   override protected def beforeEach(): Unit = {
-    setUpWorkflowExecutionData()
+    setUpWorkflowExecutionData(specId)
   }
 
   override protected def afterEach(): Unit = {
-    cleanupWorkflowExecutionData()
+    cleanupWorkflowExecutionData(specId)
   }
 
   override def beforeAll(): Unit = {

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

@@ -25,7 +25,7 @@ import org.apache.pekko.util.Timeout
 import com.twitter.util.{Await, Duration, Promise}
 import com.typesafe.scalalogging.Logger
 import org.apache.texera.amber.clustering.SingleNodeListener
-import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
+import org.apache.texera.amber.core.workflow.PortIdentity
 import org.apache.texera.amber.engine.architecture.controller.{
   ControllerConfig,
   ExecutionStateUpdate
@@ -70,12 +70,14 @@ class PauseSpec
 
   val logger = Logger("PauseSpecLogger")
 
+  private val specId = 2
+
   override protected def beforeEach(): Unit = {
-    setUpWorkflowExecutionData()
+    setUpWorkflowExecutionData(specId)
   }
 
   override protected def afterEach(): Unit = {
-    cleanupWorkflowExecutionData()
+    cleanupWorkflowExecutionData(specId)
   }
 
   override def beforeAll(): Unit = {
@@ -95,7 +97,7 @@ class PauseSpec
       links: List[LogicalLink]
   ): Unit = {
     val workflow =
-      TestUtils.buildWorkflow(operators, links, new WorkflowContext())
+      TestUtils.buildWorkflow(operators, links, TestUtils.workflowContext(specId))
     val client =
       new AmberClient(
         system,

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

@@ -27,7 +27,7 @@ import org.apache.texera.amber.clustering.SingleNodeListener
 import org.apache.texera.amber.core.executor.OpExecInitInfo
 import org.apache.texera.amber.core.tuple.Tuple
 import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
-import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
+import org.apache.texera.amber.core.workflow.PortIdentity
 import org.apache.texera.amber.engine.common.AmberRuntime
 import org.apache.texera.amber.engine.e2e.TestUtils.{
   cleanupWorkflowExecutionData,
@@ -60,14 +60,15 @@ class ReconfigurationSpec
   implicit val timeout: Timeout = Timeout(5.seconds)
 
   val logger = Logger("ReconfigurationSpecLogger")
-  val ctx = new WorkflowContext()
+  private val specId = 3
+  val ctx = TestUtils.workflowContext(specId)
 
   override protected def beforeEach(): Unit = {
-    setUpWorkflowExecutionData()
+    setUpWorkflowExecutionData(specId)
   }
 
   override protected def afterEach(): Unit = {
-    cleanupWorkflowExecutionData()
+    cleanupWorkflowExecutionData(specId)
   }
 
   override def beforeAll(): Unit = {

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

@@ -26,8 +26,12 @@ import org.apache.texera.amber.core.executor.OpExecInitInfo
 import org.apache.texera.amber.core.storage.DocumentFactory
 import org.apache.texera.amber.core.storage.model.VirtualDocument
 import org.apache.texera.amber.core.tuple.Tuple
-import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
-import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext}
+import org.apache.texera.amber.core.virtualidentity.{
+  ExecutionIdentity,
+  OperatorIdentity,
+  WorkflowIdentity
+}
+import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext, WorkflowSettings}
 import org.apache.texera.amber.engine.architecture.controller.{
   ControllerConfig,
   ExecutionStateUpdate,
@@ -65,6 +69,22 @@ import org.apache.texera.workflow.{LogicalLink, WorkflowCompiler}
 
 object TestUtils {
 
+  /**
+    * A WorkflowContext whose workflow- and execution-id are both `id`. Each e2e
+    * suite passes a distinct id so its results land in a disjoint storage
+    * keyspace (`vfs:///wid/{id}/eid/{id}/...`) and disjoint DB rows, letting the
+    * suites run concurrently without colliding on the shared Iceberg catalog.
+    */
+  def workflowContext(
+      id: Int,
+      workflowSettings: WorkflowSettings = WorkflowSettings()
+  ): WorkflowContext =
+    new WorkflowContext(
+      workflowId = WorkflowIdentity(id.toLong),
+      executionId = ExecutionIdentity(id.toLong),
+      workflowSettings = workflowSettings
+    )
+
   def buildWorkflow(
       operators: List[LogicalOp],
       links: List[LogicalLink],
@@ -91,53 +111,55 @@ object TestUtils {
     )
   }
 
-  val testUser: User = {
+  // All fixture rows for one suite share `id` as uid/wid/vid/eid; the email is
+  // derived from it so concurrent suites don't collide on the unique email key.
+  def testUser(id: Int): User = {
     val user = new User
-    user.setUid(Integer.valueOf(1))
-    user.setName("test_user")
+    user.setUid(Integer.valueOf(id))
+    user.setName(s"test_user_$id")
     user.setRole(UserRoleEnum.ADMIN)
     user.setPassword("123")
-    user.setEmail("test_user@test.com")
+    user.setEmail(s"test_user_$id@test.com")
     user
   }
 
-  val testWorkflowEntry: WorkflowPojo = {
+  def testWorkflowEntry(id: Int): WorkflowPojo = {
     val workflow = new WorkflowPojo
     workflow.setName("test workflow")
-    workflow.setWid(Integer.valueOf(1))
+    workflow.setWid(Integer.valueOf(id))
     workflow.setContent("test workflow content")
     workflow.setDescription("test description")
     workflow
   }
 
-  val testWorkflowVersionEntry: WorkflowVersion = {
+  def testWorkflowVersionEntry(id: Int): WorkflowVersion = {
     val workflowVersion = new WorkflowVersion
-    workflowVersion.setWid(Integer.valueOf(1))
-    workflowVersion.setVid(Integer.valueOf(1))
+    workflowVersion.setWid(Integer.valueOf(id))
+    workflowVersion.setVid(Integer.valueOf(id))
     workflowVersion.setContent("test version content")
     workflowVersion
   }
 
-  val testWorkflowExecutionEntry: WorkflowExecutions = {
+  def testWorkflowExecutionEntry(id: Int): WorkflowExecutions = {
     val workflowExecution = new WorkflowExecutions
-    workflowExecution.setEid(Integer.valueOf(1))
-    workflowExecution.setVid(Integer.valueOf(1))
-    workflowExecution.setUid(Integer.valueOf(1))
+    workflowExecution.setEid(Integer.valueOf(id))
+    workflowExecution.setVid(Integer.valueOf(id))
+    workflowExecution.setUid(Integer.valueOf(id))
     workflowExecution.setStatus(3.toByte)
     workflowExecution.setEnvironmentVersion("test engine")
     workflowExecution
   }
 
-  def setUpWorkflowExecutionData(): Unit = {
+  def setUpWorkflowExecutionData(id: Int): Unit = {
     val dslConfig = SqlServer.getInstance().context.configuration()
     val userDao = new UserDao(dslConfig)
     val workflowDao = new WorkflowDao(dslConfig)
     val workflowExecutionsDao = new WorkflowExecutionsDao(dslConfig)
     val workflowVersionDao = new WorkflowVersionDao(dslConfig)
-    userDao.insert(testUser)
-    workflowDao.insert(testWorkflowEntry)
-    workflowVersionDao.insert(testWorkflowVersionEntry)
-    workflowExecutionsDao.insert(testWorkflowExecutionEntry)
+    userDao.insert(testUser(id))
+    workflowDao.insert(testWorkflowEntry(id))
+    workflowVersionDao.insert(testWorkflowVersionEntry(id))
+    workflowExecutionsDao.insert(testWorkflowExecutionEntry(id))
   }
 
   /**
@@ -245,16 +267,16 @@ object TestUtils {
     result
   }
 
-  def cleanupWorkflowExecutionData(): Unit = {
+  def cleanupWorkflowExecutionData(id: Int): Unit = {
     val dslConfig = SqlServer.getInstance().context.configuration()
     val userDao = new UserDao(dslConfig)
     val workflowDao = new WorkflowDao(dslConfig)
     val workflowExecutionsDao = new WorkflowExecutionsDao(dslConfig)
     val workflowVersionDao = new WorkflowVersionDao(dslConfig)
-    workflowExecutionsDao.deleteById(1)
-    workflowVersionDao.deleteById(1)
-    workflowDao.deleteById(1)
-    userDao.deleteById(1)
+    workflowExecutionsDao.deleteById(id)
+    workflowVersionDao.deleteById(id)
+    workflowDao.deleteById(id)
+    userDao.deleteById(id)
   }
 
 }