Yicong-Huang commented on code in PR #5434:
URL: https://github.com/apache/texera/pull/5434#discussion_r3524397300
##########
amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala:
##########
@@ -501,4 +509,19 @@ class ExecutionResultService(
}
}
+ /**
+ * Callback body for `OperatorPortResultUriAvailable`: persist the URI to
+ * `operator_port_executions`. Lifted to an instance method so a unit test
+ * can drive it directly without spinning up an `AmberClient`.
+ */
+ private[service] def persistOperatorPortResultUri(
+ evt: OperatorPortResultUriAvailable
+ ): Unit = {
+ WorkflowExecutionsResource.insertOperatorPortResultUri(
Review Comment:
Done — the DAO insert now lives in
`ExecutionResultService.persistOperatorPortResultUri`, the static helper on the
resource is deleted, and `WorkflowExecutionsResourceSpec` inserts its fixture
rows with `OperatorPortExecutionsDao` directly. Description updated to match.
##########
amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala:
##########
@@ -353,6 +357,10 @@ class ExecutionResultService(
)
)
+ addSubscription(
+
client.registerCallback[OperatorPortResultUriAvailable](persistOperatorPortResultUri)
Review Comment:
Added — the registration site now documents the invariant (must be
registered before `startWorkflow`) and the silent failure mode (results compute
fine, the dashboard can never look them up).
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala:
##########
@@ -580,10 +580,8 @@ class RegionExecutionCoordinator(
DocumentFactory.createDocument(stateURI, State.schema)
if (!isRestart) {
val (_, eid, _, _) = decodeURI(resultURI)
Review Comment:
Took the suggestion — `OperatorPortResultUriAvailable` now carries only
`(globalPortId, uri)`, the `decodeURI` call in `RegionExecutionCoordinator` is
gone, and the subscriber supplies its own `executionId`.
##########
amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala:
##########
@@ -159,6 +162,28 @@ object TestUtils {
p
}
+ /**
+ * Subscribe to engine-emitted result-URI events and return a lookup over
+ * the collected URIs keyed by `(logicalOpId, portId)`. Test fixtures use
+ * this to observe URIs directly from the event stream, bypassing the
+ * `operator_port_executions` table (which only gets populated in
+ * production by `ExecutionResultService`'s parallel subscription).
+ */
+ def collectResultUriLookup(
Review Comment:
Restored — the shared e2e harness registers the production callback body
(`ExecutionResultService.persistOperatorPortResultUri`) via
`TestUtils.registerResultUriPersistence`, so `DataProcessingSpec` and the
reconfiguration specs once again cover engine event → DB row →
`getResultUriByLogicalPortId` read-back end to end.
##########
amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala:
##########
@@ -184,32 +209,24 @@ object TestUtils {
ControllerConfig.default,
error => {}
)
+ val findResultUri = collectResultUriLookup(client)
Review Comment:
Done — extracted as the shared run-and-read harness in `TestUtils`; that
piece landed on main separately as #5712 and this branch is now rebased on top
of it.
##########
amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala:
##########
@@ -36,8 +56,110 @@ import
org.apache.texera.web.service.ExecutionResultService.{
}
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import java.net.URI
+import java.sql.Timestamp
+import scala.jdk.CollectionConverters._
+
+class ExecutionResultServiceSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with MockTexeraDB {
+
+ private val testWid: Integer = 9000 + scala.util.Random.nextInt(1000)
Review Comment:
Done — fixed `testWid`/`testUid` (9001) with a comment noting the spec owns
its embedded DB, so fixed values just make failures replayable.
##########
amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala:
##########
@@ -36,8 +56,110 @@ import
org.apache.texera.web.service.ExecutionResultService.{
}
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import java.net.URI
+import java.sql.Timestamp
+import scala.jdk.CollectionConverters._
+
+class ExecutionResultServiceSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with MockTexeraDB {
+
+ private val testWid: Integer = 9000 + scala.util.Random.nextInt(1000)
+ private val testUid: Integer = 9000 + scala.util.Random.nextInt(1000)
+ private var executionsDao: WorkflowExecutionsDao = _
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+ }
+
+ override protected def afterAll(): Unit = {
+ shutdownDB()
+ }
+
+ override protected def beforeEach(): Unit = {
+ val user = new User
+ user.setUid(testUid)
+ user.setName("execution-result-test-user")
+ user.setEmail(s"[email protected]")
+ user.setPassword("password")
+ new UserDao(getDSLContext.configuration()).insert(user)
+
+ val workflow = new Workflow
+ workflow.setWid(testWid)
+ workflow.setName(s"execution-result-test-$testWid")
+ workflow.setContent("{}")
+ workflow.setDescription("")
+ workflow.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ workflow.setLastModifiedTime(new Timestamp(System.currentTimeMillis()))
+ new WorkflowDao(getDSLContext.configuration()).insert(workflow)
+
+ val version = new WorkflowVersion
+ version.setWid(testWid)
+ version.setContent("{}")
+ version.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ new WorkflowVersionDao(getDSLContext.configuration()).insert(version)
+
+ executionsDao = new WorkflowExecutionsDao(getDSLContext.configuration())
+ }
+
+ override protected def afterEach(): Unit = {
+ val ctx = getDSLContext
+ ctx
+ .deleteFrom(OPERATOR_PORT_EXECUTIONS)
+ .where(
+ OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.in(
+ ctx.select(WORKFLOW_EXECUTIONS.EID).from(WORKFLOW_EXECUTIONS)
+ )
+ )
+ .execute()
+ ctx.deleteFrom(WORKFLOW_EXECUTIONS).execute()
+ import org.apache.texera.dao.jooq.generated.Tables.{USER, WORKFLOW,
WORKFLOW_VERSION}
+
ctx.deleteFrom(WORKFLOW_VERSION).where(WORKFLOW_VERSION.WID.eq(testWid)).execute()
+ ctx.deleteFrom(WORKFLOW).where(WORKFLOW.WID.eq(testWid)).execute()
+ ctx.deleteFrom(USER).where(USER.UID.eq(testUid)).execute()
+ }
-class ExecutionResultServiceSpec extends AnyFlatSpec with Matchers {
+ "persistOperatorPortResultUri" should
+ "insert the URI carried by an OperatorPortResultUriAvailable event" in {
+ val execution = new WorkflowExecutions
+ execution.setVid(1)
Review Comment:
Fixed as suggested — the spec captures `version.getVid` right after the
insert (the DAO writes the generated key back into the pojo) and uses it for
the execution row's FK, so test ordering can no longer break it.
##########
amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala:
##########
@@ -36,8 +56,110 @@ import
org.apache.texera.web.service.ExecutionResultService.{
}
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import java.net.URI
+import java.sql.Timestamp
+import scala.jdk.CollectionConverters._
+
+class ExecutionResultServiceSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with MockTexeraDB {
+
+ private val testWid: Integer = 9000 + scala.util.Random.nextInt(1000)
+ private val testUid: Integer = 9000 + scala.util.Random.nextInt(1000)
+ private var executionsDao: WorkflowExecutionsDao = _
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+ }
+
+ override protected def afterAll(): Unit = {
+ shutdownDB()
+ }
+
+ override protected def beforeEach(): Unit = {
+ val user = new User
+ user.setUid(testUid)
+ user.setName("execution-result-test-user")
+ user.setEmail(s"[email protected]")
+ user.setPassword("password")
+ new UserDao(getDSLContext.configuration()).insert(user)
+
+ val workflow = new Workflow
+ workflow.setWid(testWid)
+ workflow.setName(s"execution-result-test-$testWid")
+ workflow.setContent("{}")
+ workflow.setDescription("")
+ workflow.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ workflow.setLastModifiedTime(new Timestamp(System.currentTimeMillis()))
+ new WorkflowDao(getDSLContext.configuration()).insert(workflow)
+
+ val version = new WorkflowVersion
+ version.setWid(testWid)
+ version.setContent("{}")
+ version.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ new WorkflowVersionDao(getDSLContext.configuration()).insert(version)
+
+ executionsDao = new WorkflowExecutionsDao(getDSLContext.configuration())
+ }
+
+ override protected def afterEach(): Unit = {
Review Comment:
Both done — every delete in `afterEach` is now scoped to the test's own ids
(the port-executions delete goes through a subquery on the test uid), and the
table imports moved to the top of the file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
