Copilot commented on code in PR #8035:
URL: https://github.com/apache/texera/pull/8035#discussion_r3869472282
##########
amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala:
##########
@@ -54,19 +83,155 @@ import scala.collection.mutable
* purpose: construction must stay side-effect-free (all throwing work is in
* `executeWorkflow`), so a future change that dereferences them during
* construction would fail here.
+ *
+ * The suite also owns two paths that need more scaffolding than that:
+ *
+ * - `executeWorkflow`'s compilation-failure arm, driven for real (not
simulated
+ * through `errorHandler`) with a CSV scan that has no file selected --
the
+ * recipe `SyncExecutionResourceSpec` establishes and
`WorkflowServiceSpec`
+ * reuses. Compilation rejects it inside
`LogicalPlan.resolveScanSourceOpFileName`
+ * before `FileResolver` is reached, so no storage is involved, and the
early
+ * `return` stops the method before
`ComputingUnitMaster.createAmberRuntime`,
+ * which would build an `AmberClient` on `AmberRuntime.actorSystem` --
null
+ * outside a started coordinator, and whatever another suite left behind
inside
+ * the shared serial JVM.
+ *
+ * - `unsubscribeAll`, the session-teardown contract. This is what makes the
+ * suite carry a TestKit `ActorSystem` and `MockTexeraDB`: the method
walks the
+ * four runtime services, and `ExecutionStatsService`'s constructor
creates its
+ * iceberg runtime-statistics table and stamps the URI onto a
`workflow_executions`
+ * row, so a non-null value for that field cannot be had more cheaply.
There is
+ * no second home for the test -- one spec file per source class -- and
stubbing
+ * the services out would leave the four teardown calls unobserved, which
is the
+ * entire content of the method.
+ *
+ * The runtime half of `executeWorkflow` (everything from
`createAmberRuntime` on) is
+ * out of reach here for the reason above and is exercised by the integration
suites.
+ *
+ * Known, pre-existing: `ExecutionStatsService` never shuts down its private
+ * `metricsPersistThread` (it does not override `unsubscribeAll`), so the
teardown
+ * test below leaks one single-thread executor into the shared JVM, as every
+ * construction of that service already does.
+ *
+ * Three further gaps are deliberately described and NOT pinned, because a
test that
+ * froze the current behaviour would cement it:
+ *
+ * - Partial construction. `executeWorkflow` assigns `client` first and the
four
+ * services after it, and `WorkflowService` catches whatever they throw
while
+ * leaving the half-built service published. `unsubscribeAll` then passes
its
+ * `client != null` guard and NPEs on the first null service, aborting
teardown.
+ *
+ * - A recovery that ends in failure. `createStateEvent` carves out
COMPLETED only,
+ * while the flag itself is lowered only by a
`WorkflowRecoveryStatus(false)` from
+ * the engine, which an execution that dies mid-recovery need not ever
send. Such an
+ * execution reaches FAILED with the flag still raised and is announced as
+ * "Recovering" from then on.
+ *
+ * - `ExecutionReconfigurationService.registerWorkerCompletionCallback`
discards the
+ * `Disposable` its `registerCallback` returns instead of handing it to
+ * `addSubscription`, so that one callback outlives the teardown below.
Not this
+ * class's defect, and not this suite's to pin.
*/
-class WorkflowExecutionServiceSpec extends AnyFlatSpec with Matchers {
+class WorkflowExecutionServiceSpec
+ extends TestKit(ActorSystem("WorkflowExecutionServiceSpec"))
+ with AnyFlatSpecLike
+ with Matchers
+ with BeforeAndAfterAll
+ with MockTexeraDB {
+
+ // Distinct from every other suite's ids: the statistics URI is derived from
wid/eid and
+ // `createDocument` truncates whatever table already sits at it
(ExecutionStatsServiceSpec
+ // owns 9107/9108).
+ private val testUid: Integer = 9207
+ private val testWid: Integer = 9207
+ private val testEid: Integer = 9208
+
+ /**
+ * The computing unit the fixture execution ran on. A different literal
from `testWid` on
+ * purpose: `getLatestExecutionID(wid, cuid)` binds two bare `Integer`s
into one predicate, so a
+ * fixture that reused one number for both would make a transposition of
the two arguments
+ * produce byte-identical SQL.
+ */
+ private val testCuid: Integer = 9209
+
+ /** A computing unit with no executions, so the cuid leg of the predicate is
not vacuous. */
+ private val otherCuid: Integer = 9210
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+
+ val user = new User
+ user.setUid(testUid)
Review Comment:
`beforeAll` overrides ScalaTest lifecycle but doesn't call
`super.beforeAll()`. If another mixed-in trait later adds its own `beforeAll`,
this override will prevent it from running (trait-stacking contract). Call
`super.beforeAll()` as the first statement to keep lifecycle hooks composable.
##########
amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala:
##########
@@ -54,19 +83,155 @@ import scala.collection.mutable
* purpose: construction must stay side-effect-free (all throwing work is in
* `executeWorkflow`), so a future change that dereferences them during
* construction would fail here.
+ *
+ * The suite also owns two paths that need more scaffolding than that:
+ *
+ * - `executeWorkflow`'s compilation-failure arm, driven for real (not
simulated
+ * through `errorHandler`) with a CSV scan that has no file selected --
the
+ * recipe `SyncExecutionResourceSpec` establishes and
`WorkflowServiceSpec`
+ * reuses. Compilation rejects it inside
`LogicalPlan.resolveScanSourceOpFileName`
+ * before `FileResolver` is reached, so no storage is involved, and the
early
+ * `return` stops the method before
`ComputingUnitMaster.createAmberRuntime`,
+ * which would build an `AmberClient` on `AmberRuntime.actorSystem` --
null
+ * outside a started coordinator, and whatever another suite left behind
inside
+ * the shared serial JVM.
+ *
+ * - `unsubscribeAll`, the session-teardown contract. This is what makes the
+ * suite carry a TestKit `ActorSystem` and `MockTexeraDB`: the method
walks the
+ * four runtime services, and `ExecutionStatsService`'s constructor
creates its
+ * iceberg runtime-statistics table and stamps the URI onto a
`workflow_executions`
+ * row, so a non-null value for that field cannot be had more cheaply.
There is
+ * no second home for the test -- one spec file per source class -- and
stubbing
+ * the services out would leave the four teardown calls unobserved, which
is the
+ * entire content of the method.
+ *
+ * The runtime half of `executeWorkflow` (everything from
`createAmberRuntime` on) is
+ * out of reach here for the reason above and is exercised by the integration
suites.
+ *
+ * Known, pre-existing: `ExecutionStatsService` never shuts down its private
+ * `metricsPersistThread` (it does not override `unsubscribeAll`), so the
teardown
+ * test below leaks one single-thread executor into the shared JVM, as every
+ * construction of that service already does.
+ *
+ * Three further gaps are deliberately described and NOT pinned, because a
test that
+ * froze the current behaviour would cement it:
+ *
+ * - Partial construction. `executeWorkflow` assigns `client` first and the
four
+ * services after it, and `WorkflowService` catches whatever they throw
while
+ * leaving the half-built service published. `unsubscribeAll` then passes
its
+ * `client != null` guard and NPEs on the first null service, aborting
teardown.
+ *
+ * - A recovery that ends in failure. `createStateEvent` carves out
COMPLETED only,
+ * while the flag itself is lowered only by a
`WorkflowRecoveryStatus(false)` from
+ * the engine, which an execution that dies mid-recovery need not ever
send. Such an
+ * execution reaches FAILED with the flag still raised and is announced as
+ * "Recovering" from then on.
+ *
+ * - `ExecutionReconfigurationService.registerWorkerCompletionCallback`
discards the
+ * `Disposable` its `registerCallback` returns instead of handing it to
+ * `addSubscription`, so that one callback outlives the teardown below.
Not this
+ * class's defect, and not this suite's to pin.
*/
-class WorkflowExecutionServiceSpec extends AnyFlatSpec with Matchers {
+class WorkflowExecutionServiceSpec
+ extends TestKit(ActorSystem("WorkflowExecutionServiceSpec"))
+ with AnyFlatSpecLike
+ with Matchers
+ with BeforeAndAfterAll
+ with MockTexeraDB {
+
+ // Distinct from every other suite's ids: the statistics URI is derived from
wid/eid and
+ // `createDocument` truncates whatever table already sits at it
(ExecutionStatsServiceSpec
+ // owns 9107/9108).
+ private val testUid: Integer = 9207
+ private val testWid: Integer = 9207
+ private val testEid: Integer = 9208
+
+ /**
+ * The computing unit the fixture execution ran on. A different literal
from `testWid` on
+ * purpose: `getLatestExecutionID(wid, cuid)` binds two bare `Integer`s
into one predicate, so a
+ * fixture that reused one number for both would make a transposition of
the two arguments
+ * produce byte-identical SQL.
+ */
+ private val testCuid: Integer = 9209
+
+ /** A computing unit with no executions, so the cuid leg of the predicate is
not vacuous. */
+ private val otherCuid: Integer = 9210
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+
+ val user = new User
+ user.setUid(testUid)
+ user.setName("workflow-execution-test-user")
+ user.setEmail(s"[email protected]")
+ new UserDao(getDSLContext.configuration()).insert(user)
+
+ val workflow = new Workflow
+ workflow.setWid(testWid)
+ workflow.setName(s"workflow-execution-test-$testWid")
+ workflow.setContent("{}")
+ workflow.setDescription("")
+ workflow.setCreationTime(new SqlTimestamp(System.currentTimeMillis()))
+ workflow.setLastModifiedTime(new SqlTimestamp(System.currentTimeMillis()))
+ new WorkflowDao(getDSLContext.configuration()).insert(workflow)
+
+ val version = new WorkflowVersion
+ version.setWid(testWid)
+ version.setContent("{}")
+ version.setCreationTime(new SqlTimestamp(System.currentTimeMillis()))
+ new WorkflowVersionDao(getDSLContext.configuration()).insert(version)
+
+ val computingUnitDao = new
WorkflowComputingUnitDao(getDSLContext.configuration())
+ List(testCuid -> "workflow-execution-test-unit", otherCuid ->
"workflow-execution-idle-unit")
+ .foreach {
+ case (cuid, name) =>
+ val unit = new WorkflowComputingUnit
+ unit.setCuid(cuid)
+ unit.setUid(testUid)
+ unit.setName(name)
+ unit.setCreationTime(new SqlTimestamp(System.currentTimeMillis()))
+ unit.setType(WorkflowComputingUnitTypeEnum.local)
+ unit.setUri("local://test")
+ unit.setResource("{}")
+ computingUnitDao.insert(unit)
+ }
+
+ // `ExecutionStatsService`'s constructor stamps the runtime-statistics URI
onto this row,
+ // reaching it from the workflow through workflow_version. `cuid` is set
because
+ // `getLatestExecutionID` matches on it and a NULL cuid matches no value
at all.
+ val execution = new WorkflowExecutions
+ execution.setEid(testEid)
+ execution.setVid(version.getVid)
+ execution.setUid(testUid)
+ execution.setCuid(testCuid)
+ execution.setStatus(0.toByte)
+ execution.setStartingTime(new SqlTimestamp(System.currentTimeMillis()))
+ execution.setBookmarked(false)
+ execution.setName("workflow-execution-test-execution")
+ execution.setEnvironmentVersion("test-env")
+ new WorkflowExecutionsDao(getDSLContext.configuration()).insert(execution)
+ }
+
+ override protected def afterAll(): Unit = {
+ try {
+ TestKit.shutdownActorSystem(system)
+ closeConnectionPool()
+ } finally super.afterAll()
+ }
Review Comment:
`afterAll` calls `TestKit.shutdownActorSystem(system)` and then
`closeConnectionPool()`. If shutting down the actor system throws (e.g.,
timeout), the DB connection pool will never be closed, leaving resources pinned
for the remainder of the sbt JVM. Wrap the DB cleanup in a `finally` so it
always runs.
--
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]
