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/08/23 06:08:52 UTC
(texera) branch main updated (21be7acce7 -> 649a27d8bc)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
from 21be7acce7 test(amber): cover the dataset search query builder's entry construction (#7855)
new fb1bb98760 test(workflow-operator): cover the SQL source's batch boundaries and dispatch (#7871)
new 649a27d8bc test(amber): cover the workflow lifecycle manager's deferred clean-up (#7872)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../texera/web/WorkflowLifecycleManagerSpec.scala | 413 ++++++++++++++++++++-
.../operator/source/sql/SQLSourceOpExecSpec.scala | 192 +++++++++-
2 files changed, 592 insertions(+), 13 deletions(-)
(texera) 02/02: test(amber): cover the workflow lifecycle manager's deferred clean-up (#7872)
Posted by gi...@apache.org.
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
commit 649a27d8bce33d8857ce6999dc0919f5b36c89ff
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sun Aug 23 05:55:49 2026 +0000
test(amber): cover the workflow lifecycle manager's deferred clean-up (#7872)
### What changes were proposed in this PR?
`WorkflowLifecycleManagerSpec` goes from 3 tests to 11, covering the
deferred clean-up of a finished workflow: refreshing the deadline when a
later terminal state arrives, postponing while a user is connected, and
cancelling the pending deadline when a user reconnects before it fires.
Measured with one fresh sbt JVM per run, `rm -rf` on the jacoco dir
first, and an identical suite-name filter. The "before" figure was
re-measured here from HEAD's 3-test spec rather than taken from the
build report.
| Metric | Before | After |
|---|---|---|
| Codecov (fully-covered lines) | 14/28 = 50.0% | **25/28 = 89.3%** |
| Missed lines | 3 | **0** |
| JaCoCo line-hit | 25/28 | **28/28** |
| Branch arms | 19 covered / 21 missed | **33 covered / 7 missed** |
| Instructions | 194/340 | **332/340** |
### Verification
28 mutations, **27 killed, 1 survivor.**
The first draft reported no survivors. That was false: **fourteen
semantic survivors existed** against its 10-test version, found across
two adversarial passes. The tests that failed to discriminate did so for
an instructive reason — several asserted exact log messages, which only
proves the interpolants are not *interchangeable*, not that the values
are right.
**The survivor, with its price tag:** hardcoding line 66's user-count
interpolant to the literal `2` survives 11/11. The declined-clean-up
message is read at exactly one user count, and the only path into
`cleanUp` with a user present requires holding the manager's monitor.
Killing it would cost more than it is worth; stated rather than papered
over.
### Where review was wrong, and it is worth recording
Two reviewer findings did not survive checking:
- One reported a mutation as passing 10/10. **Not reproducible on this
tree** — it fails two tests, and one of them was already in the
pre-review bundle.
- Another described a kill mechanism that does not hold: the mutation
neuters only one line, while a *different* line in `refreshDeadline`
still produces the asserted effect. The consequence claim attached to it
was also too strong — the leak it described needs the reconnecting user
to stay connected past a further condition.
A third suggestion was implemented but with tolerance added: an
exact-boundary comparison between a fired-at instant and a logged
deadline crosses pekko's `nanoTime` wheel and the wall clock, so it
carries 200ms of slack.
And one of my own test names asserted an accident as a requirement —
"does not schedule a second clean-up once the first has completed" is a
consequence of `cleanUp` never resetting its flag plus pekko's post-fire
`Cancellable` semantics, not a designed behaviour. Renamed.
### Cost, stated plainly
Suite runtime goes from 13s to **37s**. Three fixtures had to move to
longer timeouts (3s and 5s) to break a collision the reviewers found —
with a short timeout and a user count of 1, a timeout mutation and a
count mutation produce the same observable outcome — plus one new 3.5s
test. There is no injectable clock: `cleanUpTimeout` is an `Int` in
seconds and the scheduler comes through the `AmberRuntime` singleton, so
the wall-clock cost is the price of testing this file without a
production seam.
### Deliberately not included
Lines 33, 40 and 86 stay partial, so the file cannot exceed 25/28
without a production change — line 33 is the Scala 2.13 lazy-val bitmap
re-check inside `logger$lzycompute`, and the others are similar guards.
Line 97 (`setCleanUpDeadline(newState.state)`, inside a subscribe
lambda) is absent from JaCoCo's tracked set entirely: the report lists
exactly 28 lines and 97 is not among them, because `SyntheticFilter`
drops the enclosing lambda.
Nothing outside `WorkflowLifecycleManager.scala` was pinned — in
particular not `ExecutionStateStore`'s replay semantics or
`AmberRuntime`'s scheduler.
No production file is touched. Both `git diff -- '*/src/main/*'` and
`git diff --cached -- '*/src/main/*'` are empty, and the production
sha256 is byte-identical to the pre-mutation snapshot.
### Any related issues, documentation, discussions?
Closes #7869
### How was this PR tested?
```
sbt "WorkflowExecutionService/testOnly org.apache.texera.web.WorkflowLifecycleManagerSpec"
```
```
[info] Total number of tests run: 11
[info] Tests: succeeded 11, failed 0, canceled 0, ignored 0, pending 0
```
`Test/scalafmtCheck` passes.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---------
Signed-off-by: Xinyuan Lin <xi...@uci.edu>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../texera/web/WorkflowLifecycleManagerSpec.scala | 413 ++++++++++++++++++++-
1 file changed, 403 insertions(+), 10 deletions(-)
diff --git a/amber/src/test/scala/org/apache/texera/web/WorkflowLifecycleManagerSpec.scala b/amber/src/test/scala/org/apache/texera/web/WorkflowLifecycleManagerSpec.scala
index 6260de4ad6..77f8aa675a 100644
--- a/amber/src/test/scala/org/apache/texera/web/WorkflowLifecycleManagerSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/WorkflowLifecycleManagerSpec.scala
@@ -19,7 +19,10 @@
package org.apache.texera.web
-import org.apache.pekko.actor.ActorSystem
+import ch.qos.logback.classic.spi.ILoggingEvent
+import ch.qos.logback.classic.{Level, Logger => LogbackLogger}
+import ch.qos.logback.core.AppenderBase
+import org.apache.pekko.actor.{ActorSystem, Cancellable}
import org.apache.pekko.testkit.TestKit
import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.{
COMPLETED,
@@ -29,8 +32,12 @@ import org.apache.texera.amber.engine.common.AmberRuntime
import org.apache.texera.web.storage.ExecutionStateStore
import org.scalatest.BeforeAndAfterAll
import org.scalatest.flatspec.AnyFlatSpec
+import org.slf4j.LoggerFactory
-import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.time.LocalDateTime
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
+import scala.jdk.CollectionConverters._
class WorkflowLifecycleManagerSpec extends AnyFlatSpec with BeforeAndAfterAll {
@@ -69,17 +76,32 @@ class WorkflowLifecycleManagerSpec extends AnyFlatSpec with BeforeAndAfterAll {
}
private def managerWithCallback(
- cleanUpTimeout: Int = 1
+ cleanUpTimeout: Int = 1,
+ id: String = "workflow-lifecycle-manager-spec"
): (WorkflowLifecycleManager, CountDownLatch) = {
val cleaned = new CountDownLatch(1)
val manager = new WorkflowLifecycleManager(
- id = "workflow-lifecycle-manager-spec",
+ id = id,
cleanUpTimeout = cleanUpTimeout,
cleanUpCallback = () => cleaned.countDown()
)
(manager, cleaned)
}
+ /** Counts the clean-ups instead of latching on the first one, so a repeat is observable. */
+ private def managerWithCounter(
+ cleanUpTimeout: Int = 1,
+ id: String = "workflow-lifecycle-manager-spec"
+ ): (WorkflowLifecycleManager, AtomicInteger) = {
+ val cleanUps = new AtomicInteger(0)
+ val manager = new WorkflowLifecycleManager(
+ id = id,
+ cleanUpTimeout = cleanUpTimeout,
+ cleanUpCallback = () => { cleanUps.incrementAndGet(); () }
+ )
+ (manager, cleanUps)
+ }
+
private def assertCleanUpWithin(cleaned: CountDownLatch, seconds: Long): Unit = {
assert(
cleaned.await(seconds, TimeUnit.SECONDS),
@@ -87,6 +109,95 @@ class WorkflowLifecycleManagerSpec extends AnyFlatSpec with BeforeAndAfterAll {
)
}
+ /**
+ * Buffers the events a logger emits. logback's own ListAppender collects into a plain
+ * ArrayList, and the clean-up body runs on the scheduler's execution context rather than on
+ * the test thread, so the buffer has to tolerate appends from another thread.
+ */
+ private final class CollectingAppender extends AppenderBase[ILoggingEvent] {
+ private val events = new ConcurrentLinkedQueue[ILoggingEvent]
+
+ override def append(event: ILoggingEvent): Unit = events.add(event)
+
+ def messages: Seq[String] = events.asScala.map(_.getFormattedMessage).toSeq
+ }
+
+ private val managerLoggerName = classOf[WorkflowLifecycleManager].getName
+
+ /**
+ * Runs `body` with the manager's logger raised to INFO, handing it a live view of the
+ * messages logged for `id`.
+ *
+ * The manager reports every lifecycle decision at INFO, and amber's logback.xml pins
+ * `org.apache` to WARN, so those calls sit behind `isInfoEnabled` and stay dormant unless a
+ * test raises the level. Additivity is switched off so the captured lines do not also reach
+ * the shared console and rolling-file appenders.
+ */
+ private def withInfoLogs[T](id: String)(body: (() => Seq[String]) => T): T = {
+ val logger = LoggerFactory.getLogger(managerLoggerName).asInstanceOf[LogbackLogger]
+ logger.synchronized {
+ val appender = new CollectingAppender
+ val previousLevel = logger.getLevel
+ val previousAdditive = logger.isAdditive
+ appender.setContext(logger.getLoggerContext)
+ appender.setName(
+ s"workflow-lifecycle-manager-spec-appender-$id-${Thread.currentThread().getId}"
+ )
+ appender.start()
+ logger.addAppender(appender)
+ logger.setLevel(Level.INFO)
+ logger.setAdditive(false)
+ try {
+ body(() => appender.messages.filter(_.startsWith(s"[$id] ")))
+ } finally {
+ logger.detachAppender(appender)
+ logger.setAdditive(previousAdditive)
+ logger.setLevel(previousLevel)
+ appender.stop()
+ }
+ }
+ }
+
+ /** Waits for `fragment` to show up, then returns everything logged so far. */
+ private def awaitMessage(
+ messages: () => Seq[String],
+ fragment: String,
+ seconds: Long
+ ): Seq[String] = {
+ val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(seconds)
+ while (System.nanoTime() < deadline && !messages().exists(_.contains(fragment))) {
+ Thread.sleep(50)
+ }
+ val captured = messages()
+ assert(
+ captured.exists(_.contains(fragment)),
+ s"no logged message contained '$fragment', got: $captured"
+ )
+ captured
+ }
+
+ private def awaitCleanUpCount(cleanUps: AtomicInteger, expected: Int, seconds: Long): Unit = {
+ val deadline = System.currentTimeMillis() + seconds * 1000
+ while (System.currentTimeMillis() < deadline && cleanUps.get() < expected) {
+ Thread.sleep(50)
+ }
+ assert(cleanUps.get() == expected, s"expected $expected clean-up(s), saw ${cleanUps.get()}")
+ }
+
+ /**
+ * The scheduled deadline, read straight off the manager.
+ *
+ * Used only as a non-vacuity probe: pekko marks a task that has already fired as neither
+ * cancelled nor cancellable, while a task cancelled before it fired reports isCancelled. The
+ * two "reconnects as the deadline fires" tests would otherwise pass on a runner so slow that
+ * the deadline never fired at all.
+ */
+ private def scheduledCleanUp(manager: WorkflowLifecycleManager): Cancellable = {
+ val field = classOf[WorkflowLifecycleManager].getDeclaredField("cleanUpExecution")
+ field.setAccessible(true)
+ field.get(manager).asInstanceOf[Cancellable]
+ }
+
"WorkflowLifecycleManager" should "wait for the last user before scheduling cleanup" in {
val (manager, cleaned) = managerWithCallback()
@@ -117,17 +228,299 @@ class WorkflowLifecycleManagerSpec extends AnyFlatSpec with BeforeAndAfterAll {
}
it should "refresh the deadline when a later terminal state arrives" in {
- val (manager, cleaned) = managerWithCallback(cleanUpTimeout = 4)
+ val id = "refresh-deadline"
+ val (manager, cleaned) = managerWithCallback(cleanUpTimeout = 4, id = id)
val stateStore = new ExecutionStateStore
- manager.registerCleanUpOnStateChange(stateStore)
- Thread.sleep(2000)
- stateStore.metadataStore.updateState(_.withState(COMPLETED))
+ withInfoLogs(id) { messages =>
+ manager.registerCleanUpOnStateChange(stateStore)
+ // Registration alone must arm a deadline: the replayed state is not RUNNING and no user is
+ // connected. Asserting it is what makes the refresh below meaningful -- with no earlier
+ // deadline there is nothing to replace, and the timing windows on their own cannot tell a
+ // refreshed deadline from a first one armed at the update.
+ assert(
+ messages().count(_.contains("will start at")) == 1,
+ s"registration did not arm a deadline: ${messages()}"
+ )
+ Thread.sleep(2000)
+ val beforeUpdate = LocalDateTime.now()
+ stateStore.metadataStore.updateState(_.withState(COMPLETED))
+ val armed = messages().filter(_.contains("will start at"))
+ assert(armed.size == 2, s"the terminal state did not re-arm the deadline: ${messages()}")
+
+ // The refreshed deadline has to be cleanUpTimeout out, and this is the only test that reads
+ // a logged deadline at a timeout other than 3, so it is what stops the logged deadline from
+ // being satisfied by one hardcoded number of seconds.
+ val refreshed =
+ LocalDateTime.parse(armed.last.stripPrefix(s"[$id] workflow state clean up will start at "))
+ assert(
+ !refreshed.isBefore(beforeUpdate.plusSeconds(4)) &&
+ refreshed.isBefore(beforeUpdate.plusSeconds(6)),
+ s"the refreshed deadline $refreshed is not 4s past $beforeUpdate"
+ )
+
+ assert(
+ !cleaned.await(3000, TimeUnit.MILLISECONDS),
+ "a refreshed deadline must cancel the earlier one"
+ )
+ assertCleanUpWithin(cleaned, seconds = 5)
+ }
+ }
+
+ it should "postpone the clean-up while a user is connected, reporting the count and status" in {
+ val id = "postpone-with-user"
+ // cleanUpTimeout must differ from every user count this test reads, or the manager's two Int
+ // fields render interchangeably and the count interpolants are unpinned.
+ val (manager, cleaned) = managerWithCallback(cleanUpTimeout = 3, id = id)
+ val stateStore = new ExecutionStateStore
+
+ val messages = withInfoLogs(id) { messages =>
+ manager.increaseUserCount()
+ manager.increaseUserCount()
+ // metadataStore is a BehaviorSubject, so subscribing replays the current state
+ // (UNINITIALIZED) synchronously on this thread.
+ manager.registerCleanUpOnStateChange(stateStore)
+ // UNINITIALIZED is the protobuf default the replay hands over for free, so a status
+ // interpolant hardcoded to it would render identically. Push a state the manager cannot
+ // have defaulted to: userCount > 0 short-circuits the guard, so COMPLETED still takes the
+ // postpone arm and the status has to be reported as COMPLETED.
+ stateStore.metadataStore.updateState(_.withState(COMPLETED))
+ // Drop back to one user and push a third state, so the status line is read at two different
+ // counts and three different statuses. A mutant that hardcodes whichever count or status
+ // this test happens to expect cannot render all of them.
+ manager.decreaseUserCount(Some(RUNNING))
+ stateStore.metadataStore.updateState(_.withState(RUNNING))
+ messages()
+ }
+
+ assert(
+ messages == Seq(
+ s"[$id] workflow state clean up postponed. current user count = 1",
+ s"[$id] workflow state clean up postponed. current user count = 2",
+ s"[$id] workflow state clean up postponed. current user count = 2, " +
+ "workflow status = UNINITIALIZED",
+ s"[$id] workflow state clean up postponed. current user count = 2, " +
+ "workflow status = COMPLETED",
+ s"[$id] workflow state clean up postponed. current user count = 1",
+ s"[$id] workflow state clean up postponed. current user count = 1, " +
+ "workflow status = RUNNING"
+ ),
+ s"unexpected lifecycle log: $messages"
+ )
+ assert(
+ !cleaned.await(3500, TimeUnit.MILLISECONDS),
+ "no deadline may be armed while a user is connected"
+ )
+ }
+
+ it should "log the deadline it arms and the completion of the clean-up" in {
+ val id = "log-deadline"
+ // 3 seconds, not 1: the window below has to be narrow enough to reject an arithmetic slip in
+ // the logged deadline, and 3 does not collide with the user count this test reads.
+ val cleanUpTimeout = 3
+ val (manager, cleaned) = managerWithCallback(cleanUpTimeout = cleanUpTimeout, id = id)
+
+ val armedAt = LocalDateTime.now()
+ val (messages, firedAt) = withInfoLogs(id) { messages =>
+ manager.increaseUserCount()
+ manager.decreaseUserCount(None)
+ assertCleanUpWithin(cleaned, seconds = 6)
+ val firedAt = LocalDateTime.now()
+ // the completion is logged after the callback returns
+ (awaitMessage(messages, "clean up completed.", seconds = 5), firedAt)
+ }
+
+ // The expected wording is a literal, not the production string echoed back: comparing
+ // `armed.head` against itself would move both sides of the assertion together.
+ val armedPrefix = s"[$id] workflow state clean up will start at "
+ val armed = messages.filter(_.contains("will start at"))
+ assert(armed.size == 1, s"expected exactly one armed deadline, got: $messages")
+ assert(
+ armed.head.startsWith(armedPrefix),
+ s"unexpected armed-deadline wording: ${armed.head}"
+ )
+ val loggedDeadline = LocalDateTime.parse(armed.head.stripPrefix(armedPrefix))
+ assert(
+ !loggedDeadline.isBefore(armedAt.plusSeconds(cleanUpTimeout.toLong)),
+ s"the logged deadline $loggedDeadline is less than ${cleanUpTimeout}s past $armedAt"
+ )
+ assert(
+ loggedDeadline.isBefore(armedAt.plusSeconds(cleanUpTimeout + 2L)),
+ s"the logged deadline $loggedDeadline is more than ${cleanUpTimeout + 2}s past $armedAt"
+ )
+ // The advertised deadline and the deadline actually armed are separate expressions in
+ // refreshDeadline, so cross-check them: the clean-up must fire at the instant it advertised.
+ // The 200ms of slack on the lower bound absorbs the pekko wheel's tick rounding and the
+ // nanoTime/wall-clock mismatch; it is far tighter than any plausible arithmetic slip.
+ assert(
+ !firedAt.isBefore(loggedDeadline.minusNanos(200000000L)),
+ s"the clean-up fired at $firedAt, before the advertised $loggedDeadline"
+ )
assert(
- !cleaned.await(3000, TimeUnit.MILLISECONDS),
- "a refreshed deadline must cancel the earlier one"
+ firedAt.isBefore(loggedDeadline.plusSeconds(1)),
+ s"the clean-up fired at $firedAt, well past the advertised $loggedDeadline"
)
+ assert(
+ messages == Seq(
+ s"[$id] workflow state clean up postponed. current user count = 1",
+ armedPrefix + loggedDeadline.toString,
+ s"[$id] workflow state clean up completed."
+ ),
+ s"unexpected lifecycle log: $messages"
+ )
+ }
+
+ it should "schedule the clean-up when the last user leaves a finished workflow" in {
+ val (manager, cleaned) = managerWithCallback(id = "finished-workflow")
+
+ manager.increaseUserCount()
+ manager.decreaseUserCount(Some(COMPLETED))
+
assertCleanUpWithin(cleaned, seconds = 5)
}
+
+ it should "keep a still-running workflow alive when its last user leaves" in {
+ val id = "running-workflow"
+ // 5 collides with none of the counts below (0, 1, 2), so no count interpolant can be swapped
+ // for the timeout and still render.
+ val (manager, cleaned) = managerWithCallback(cleanUpTimeout = 5, id = id)
+
+ val messages = withInfoLogs(id) { messages =>
+ manager.increaseUserCount()
+ manager.increaseUserCount()
+ // 2 -> 1 takes the postpone arm because the count is not zero yet; 1 -> 0 takes it because
+ // the workflow is still RUNNING. Reading that line at a NONZERO count is the point: at zero
+ // it absorbs multiplicative mutations, so a count read only at zero is not pinned at all.
+ manager.decreaseUserCount(Some(RUNNING))
+ manager.decreaseUserCount(Some(RUNNING))
+ messages()
+ }
+
+ // The exhaustive comparison is also what proves nothing was armed: an armed deadline would
+ // add a "will start at" line here.
+ assert(
+ messages == Seq(
+ s"[$id] workflow state clean up postponed. current user count = 1",
+ s"[$id] workflow state clean up postponed. current user count = 2",
+ s"[$id] workflow state clean up postponed. current user count = 1",
+ s"[$id] workflow state clean up postponed. current user count = 0"
+ ),
+ s"unexpected lifecycle log: $messages"
+ )
+ assert(
+ !cleaned.await(2, TimeUnit.SECONDS),
+ "a running workflow must not be cleaned up when its last user leaves"
+ )
+ }
+
+ it should "cancel the pending deadline when a user reconnects before it fires" in {
+ val id = "reconnect-before-deadline"
+ val (manager, cleaned) = managerWithCallback(cleanUpTimeout = 3, id = id)
+
+ withInfoLogs(id) { messages =>
+ manager.increaseUserCount()
+ manager.decreaseUserCount(None)
+ // Well inside the 3s deadline, so the task below is certainly still pending.
+ Thread.sleep(500)
+ manager.increaseUserCount()
+ // increaseUserCount cancels the armed deadline outright. pekko marks a task cancelled
+ // before it fired as isCancelled; a task still pending, or one that already fired, is not.
+ // So this holds only if the reconnect really cancelled the live task.
+ assert(
+ scheduledCleanUp(manager).isCancelled,
+ "the pending deadline was not cancelled by the reconnect"
+ )
+ assert(
+ !messages().exists(_.contains("clean up failed.")),
+ s"a cancelled deadline still reached the clean-up: ${messages()}"
+ )
+
+ // And the cancellation leaves the manager able to arm a replacement.
+ manager.decreaseUserCount(None)
+ assertCleanUpWithin(cleaned, seconds = 6)
+ }
+ }
+
+ it should "decline the clean-up when a user reconnects as the deadline fires" in {
+ val id = "reconnect-at-deadline"
+ val (manager, cleaned) = managerWithCallback(cleanUpTimeout = 1, id = id)
+
+ val messages = withInfoLogs(id) { messages =>
+ manager.increaseUserCount()
+ manager.decreaseUserCount(None)
+ manager.synchronized {
+ // The deadline fires while this monitor is held: the scheduler marks the task as run and
+ // hands the clean-up to the execution context, where it blocks until the reconnect below
+ // has been recorded.
+ Thread.sleep(2000)
+ // Two reconnects, so the reported count (2) differs from cleanUpTimeout (which has to
+ // stay at 1 for the 2000ms hold to straddle the deadline). Otherwise the declined-clean-up
+ // message could interpolate either field and render the same.
+ manager.increaseUserCount()
+ manager.increaseUserCount()
+ assert(
+ !scheduledCleanUp(manager).isCancelled,
+ "the deadline was cancelled instead of fired, so nothing raced the reconnect"
+ )
+ }
+ awaitMessage(messages, "clean up failed.", seconds = 5)
+ }
+
+ assert(
+ messages.contains(s"[$id] workflow state clean up failed. current user count = 2"),
+ s"the declined clean-up was not reported: $messages"
+ )
+ assert(
+ !messages.exists(_.contains("clean up completed.")),
+ s"the clean-up completed even though a user had reconnected: $messages"
+ )
+ assert(cleaned.getCount == 1, "the clean-up callback ran even though a user had reconnected")
+ }
+
+ it should "decline the clean-up at the logger's default level as well" in {
+ val id = "reconnect-at-deadline-quiet"
+ val (manager, cleaned) = managerWithCallback(cleanUpTimeout = 1, id = id)
+
+ manager.increaseUserCount()
+ manager.decreaseUserCount(None)
+ manager.synchronized {
+ Thread.sleep(2000)
+ manager.increaseUserCount()
+ assert(
+ !scheduledCleanUp(manager).isCancelled,
+ "the deadline was cancelled instead of fired, so nothing raced the reconnect"
+ )
+ }
+
+ assert(
+ !cleaned.await(3, TimeUnit.SECONDS),
+ "the clean-up ran even though a user had reconnected"
+ )
+ }
+
+ it should "arm no replacement deadline after a clean-up has run, because the manager is single-use" in {
+ val (manager, cleanUps) = managerWithCounter(cleanUpTimeout = 1, id = "single-clean-up")
+
+ manager.increaseUserCount()
+ manager.decreaseUserCount(None)
+ awaitCleanUpCount(cleanUps, expected = 1, seconds = 5)
+
+ // Single-use is not a designed guarantee of this class; it falls out of two facts outside it.
+ // (1) cleanUp leaves cleanUpExecution pointing at the task that just fired -- it never resets
+ // the field. (2) pekko's LightArrayRevolverScheduler reports a fired task as neither cancelled
+ // nor cancellable, so refreshDeadline's `isCancelled || cancel()` guard is false from then on.
+ // Production never notices, because cleanUpCallback drops the WorkflowService from
+ // workflowServiceMapping and a reconnect builds a fresh manager.
+ //
+ // This is therefore a description of the current single-use contract, NOT a statement that
+ // reuse would be wrong. Anyone making WorkflowLifecycleManager reusable, and anyone hitting a
+ // change in pekko's post-fire Cancellable semantics, should expect to rewrite this test rather
+ // than treat it as a specification.
+ manager.increaseUserCount()
+ manager.decreaseUserCount(None)
+ Thread.sleep(3000)
+
+ assert(cleanUps.get() == 1, s"the clean-up ran ${cleanUps.get()} times, expected once")
+ }
}
(texera) 01/02: test(workflow-operator): cover the SQL source's batch boundaries and dispatch (#7871)
Posted by gi...@apache.org.
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
commit fb1bb987603169b9879ad12a9578fc2a226c3945
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sun Aug 23 05:54:55 2026 +0000
test(workflow-operator): cover the SQL source's batch boundaries and dispatch (#7871)
### What changes were proposed in this PR?
`SQLSourceOpExecSpec` goes from 30 tests to 40, covering the
batch-boundary arithmetic and type dispatch in `SQLSourceOpExec`.
Measured with the `FileScanSourceOpExecSpec` suite-name exclusion
applied identically to the before and after run — that suite aborts on
this machine for pre-existing reasons, and because sbt-jacoco runs
unforked a failing test task skips `saveRuntimeData` and emits an
all-zero report rather than a partial one.
| Metric | Before | After |
|---|---|---|
| Codecov (fully-covered lines) | 139/163 = 85.3% | **147/163 = 90.2%**
|
| Missed lines | 4 (230, 231, 364, 519) | **0** |
| JaCoCo line-hit | 159/163 = 97.6% | **163/163 = 100%** |
| Branch arms | 127 covered / 43 missed | **137 covered / 33 missed** |
Tests: spec 30 → 40; the `source.sql.*` package 176 → 186 across 11
suites; the module 2307 → 2317.
### The +8 is not eight lines of behaviour, and the first draft implied
it was
Of the eight newly fully-covered lines, **five are production-reachable
behaviour** — the INTEGER and DOUBLE type dispatch, the progressive arm
of the offset guard, the non-auto LONG maximum, and the non-auto
unsupported-type throw. The other **three are defensive arms no workflow
can reach**: `hasNextQuery`'s unsupported-type throw is unreachable
because a successful `open()` admits only INTEGER/LONG/TIMESTAMP/DOUBLE,
and two more are similar guards.
A reviewer put the split at +6 reachable / +2 defensive. That was closer
than the original claim but still not right; the measured split is 5 and
3.
### Verification
22 mutations, **20 killed, 2 survivors.** Every one of the nine mutants
the two reviewers reported as surviving was first *reproduced surviving*
on the pre-review spec, and eight of the nine now die.
**The two survivors are deliberate, and killing them would make the
suite worse.** `isLastBatch = nextLowerBound >= upperBound` can be
flipped to `>` on both the LONG arm (line 282) and its DOUBLE twin (line
285) and survive all 186 package tests. This was verified by running it,
not inferred. The only input that distinguishes the two makes production
emit a **duplicate final window** — so a test that killed these mutants
would cement that defect. Reported instead.
### Corrections to my own first draft
- "10 mutations, all 10 killed, zero survivors" was true of that table,
but the table was too narrow to support the bundle's claims — nine
further mutants on the very lines the bundle claimed to newly cover were
untested, and most of them survived.
- The +8/+10 figures reproduce exactly, but were stated without the
reachability qualification above.
- The conclusion drawn from line 502 was wrong even though its coverage
half was right.
- Module baselines were off by two: 208 missed lines (96.59%) and 1215
missed branches (51.96%), not 206/1214.
- Line counts and test counts in the first draft were stale.
- One assertion — `(statement.setLong _).expects(*, *).never()` —
carried no information beyond the strict mock itself and is gone,
replaced by a positive `expects(1, 2L)` whose strictness does the work.
- A CI-wiring claim from the first draft is not restated here, because
it was not re-verified in this pass.
### Deliberately not included, with evidence
- **Four `case A | B | C | _ =>` arms** (230, 286, 320, 357) keep a dead
`ifeq` side: scalac compiles the `_` as a constant before the final
branch.
- **Ten lines** are sealed-`Option` `None$.equals` checks falling
through to `new scala/MatchError`, plus an `$outer` null check and a
non-local-return rethrow.
- **Line 502's `batchByAttribute.isDefined` false arm is dead**:
`initBatchColumnBoundaries` is private with one caller, `open()`, which
always assigns `Option(schema.getAttribute(...))`, and
`Schema.getAttribute` throws rather than returning null. Widening the
private would be a production edit — refused.
- **Line 472's `if (limit > 0)` false arm is unreachable**:
`generateSqlQuery` returns `None` whenever the limit is `Some(n <= 0)`,
so 472 is only reached when the limit is positive. Getting there needs
the protected `generateSqlQuery` overridden, i.e. replacing the code
under test.
No production file is touched — sha256 unchanged from the pre-mutation
snapshot, and no stray `test_large_binary.txt` was left behind, since
the FileScan suite never ran.
### Any related issues, documentation, discussions?
Closes #7868
### How was this PR tested?
```
sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.source.sql.SQLSourceOpExecSpec"
```
```
[info] Total number of tests run: 40
[info] Tests: succeeded 40, failed 0, canceled 0, ignored 0, pending 0
```
The whole `source.sql.*` package is green at 186 tests across 11 suites,
0 aborted. `Test/scalafmtCheck` passes.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---------
Signed-off-by: Xinyuan Lin <xi...@uci.edu>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../operator/source/sql/SQLSourceOpExecSpec.scala | 192 ++++++++++++++++++++-
1 file changed, 189 insertions(+), 3 deletions(-)
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
index 7d549b2c18..f947ba284d 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/SQLSourceOpExecSpec.scala
@@ -47,6 +47,15 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
List(new Attribute("id", AttributeType.INTEGER), new Attribute("name", AttributeType.STRING))
)
+ /** `rowSchema` plus the LONG column the progressive row tests batch by. */
+ private val progressiveRowSchema = Schema(
+ List(
+ new Attribute("id", AttributeType.INTEGER),
+ new Attribute("name", AttributeType.STRING),
+ new Attribute("big", AttributeType.LONG)
+ )
+ )
+
/**
* Minimal concrete executor: the DB-touching hooks are stubbed out so the
* query-building surface can be exercised on its own.
@@ -86,6 +95,7 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
def terminator: String = render(b => terminateSQL(b))
def slidingWindow: String = render(b => addBatchSlidingWindow(b))
def batchValue(value: Number): String = batchAttributeToString(value)
+ def boundary(side: String): Number = fetchBatchByBoundary(side)
}
private def descJson(configure: PostgreSQLSourceOpDesc => Unit = _ => ()): String = {
@@ -128,10 +138,17 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
exec
}
- /** A connection answering the `SELECT MIN/MAX(col) FROM tbl;` boundary probes. */
- private def boundaryConn(column: String, stub: (ResultSet, String) => Unit): Connection = {
+ /**
+ * A connection answering the `SELECT MIN/MAX(col) FROM tbl;` boundary probes.
+ * `sides` narrows it to the probes the executor is expected to actually issue.
+ */
+ private def boundaryConn(
+ column: String,
+ stub: (ResultSet, String) => Unit,
+ sides: Seq[String] = Seq("MIN", "MAX")
+ ): Connection = {
val conn = mock[Connection]
- Seq("MIN", "MAX").foreach { side =>
+ sides.foreach { side =>
val statement = mock[PreparedStatement]
val resultSet = mock[ResultSet]
(conn
@@ -202,7 +219,47 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
exec.sqlQuery shouldBe Some(BASE + ";")
}
+ it should "skip the sliding window when progressive mode names no batch column" in {
+ // A progressive descriptor with no batchByColumn is degenerate but generateSqlQuery is
+ // defensive about it. open() is deliberately not called here: it would blow up on
+ // `desc.batchByColumn.get`, which is the current (unhelpful) behaviour, so this test
+ // pins only the query builder. That also means this arm of the progressive guard
+ // (progressive with no batch column) is defensive: no production caller reaches it,
+ // generateSqlQuery in this state, since open() throws first.
+ // min and max are populated on purpose: with all three of batchByColumn/min/max empty,
+ // re-pointing the guard at `desc.min` instead would be indistinguishable here.
+ val exec = new TestSQLSourceOpExec(descJson { desc =>
+ desc.progressive = Option(true)
+ desc.batchByColumn = None
+ desc.min = Option("0")
+ desc.max = Option("100")
+ desc.interval = 30L
+ })
+ exec.sqlQuery shouldBe Some(BASE + ";")
+ }
+
+ it should "ignore a leftover batch column when progressive mode is off" in {
+ // `progressive` only toggles batchByColumn/min/max/interval *hidden* in the UI
+ // (SQLSourceOpDesc's toggleHidden), it does not clear them, so a non-progressive
+ // descriptor can still carry a batch column. Neither the sliding window nor the
+ // fixed OFFSET may key off that column instead of the progressive flag.
+ val exec = new TestSQLSourceOpExec(descJson { desc =>
+ desc.batchByColumn = Option("big")
+ desc.min = Option("0")
+ desc.max = Option("100")
+ desc.interval = 30L
+ })
+ exec.curOffset = Some(5L)
+ exec.sqlQuery shouldBe Some(BASE + " OFFSET ?;")
+ }
+
it should "walk the LONG batch column window by window until the upper bound" in {
+ // Deliberately NOT pinned: the `>=` that decides the last window (SQLSourceOpExec's
+ // `isLastBatch`, and its DOUBLE twin) is only observable when the upper bound sits
+ // exactly one interval above the current lower bound -- and on that input production
+ // emits a redundant `>= upper AND <= upper` window that re-scans the boundary row.
+ // Asserting the current behaviour there would cement that duplication, so it is
+ // reported as a defect rather than encoded here.
val exec = openedProgressive("big", "0", "100", 30L)
exec.nextQueryAvailable shouldBe true
@@ -279,6 +336,36 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
exec.nextQueryAvailable shouldBe false
}
+ it should "offer a next query while an INTEGER batch column is below its upper bound" in {
+ val conn = boundaryConn(
+ "id",
+ (rs, side) => (rs.getInt(_: Int)).expects(1).returning(if (side == "MIN") 0 else 10)
+ )
+ val exec = openedProgressive("id", "auto", "auto", 4L, conn)
+
+ exec.nextQueryAvailable shouldBe true
+ exec.slidingWindow shouldBe " AND id >= 0 AND id < 4"
+
+ // A single-value range (min == max) must still yield exactly one window. This is the
+ // one input on which the `<=` in hasNextQuery is load-bearing, and unlike the
+ // exact-multiple case it does not exercise the duplicate-last-window defect.
+ val singleValueConn =
+ boundaryConn("id", (rs, _) => (rs.getInt(_: Int)).expects(1).returning(7))
+ val singleValue = openedProgressive("id", "auto", "auto", 4L, singleValueConn)
+ singleValue.nextQueryAvailable shouldBe true
+ singleValue.slidingWindow shouldBe " AND id >= 7 AND id <= 7"
+ singleValue.nextQueryAvailable shouldBe false
+ }
+
+ it should "refuse to decide the next query for an unsupported batch column type" in {
+ val exec = new TestSQLSourceOpExec(progressiveJson("name", Option("a"), Option("z"), 4L))
+ // DEFENSIVE-ONLY arm: no workflow can reach hasNextQuery with a STRING batch column, but
+ // this pins the type-dispatch error message without depending on open()'s failure path.
+ exec.batchByAttribute = Some(new Attribute("name", AttributeType.STRING))
+ intercept[IllegalArgumentException](exec.nextQueryAvailable).getMessage shouldBe
+ "Unexpected type: string"
+ }
+
"SQLSourceOpExec.open" should "validate the table against the loaded table names" in {
val exec = new TestSQLSourceOpExec(descJson(), knownTables = Seq("other"))
val thrown = intercept[RuntimeException](exec.open())
@@ -286,6 +373,20 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
exec.loadTableNamesCalls shouldBe 1
}
+ it should "validate the table before probing the batch column boundaries" in {
+ // fetchBatchByBoundary concatenates desc.table straight into the probe SQL, so the
+ // table-name check -- which this file documents as its SQL-injection defence -- has to
+ // run before any boundary probe. The strict mock with no expectations is the oracle:
+ // any probe at all is an unexpected call.
+ val conn = mock[Connection]
+ val exec = new TestSQLSourceOpExec(
+ progressiveJson("big", Option("auto"), Option("auto"), 30L),
+ conn,
+ knownTables = Seq("other")
+ )
+ intercept[RuntimeException](exec.open()).getMessage shouldBe "Can't find the given table `tbl`."
+ }
+
it should "leave batchByAttribute unset when progressive mode is disabled" in {
val exec = new TestSQLSourceOpExec(descJson())
exec.open()
@@ -299,6 +400,34 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
thrown.getMessage should startWith("Missing required progressive configuration")
}
+ it should "reject progressive mode with a lower but no upper boundary" in {
+ val exec = new TestSQLSourceOpExec(progressiveJson("big", Option("0"), None, 30L))
+ val thrown = intercept[IllegalArgumentException](exec.open())
+ thrown.getMessage should startWith("Missing required progressive configuration")
+ }
+
+ it should "reject progressive mode with an upper but no lower boundary" in {
+ // the mirror of the case above: without this one, dropping the min check from the
+ // configuration guard would go unnoticed and desc.min.get would raise a bare
+ // NoSuchElementException instead of this message
+ val exec = new TestSQLSourceOpExec(progressiveJson("big", None, Option("100"), 30L))
+ val thrown = intercept[IllegalArgumentException](exec.open())
+ thrown.getMessage should startWith("Missing required progressive configuration")
+ }
+
+ it should "reject a non-auto upper bound on a column that is neither LONG nor TIMESTAMP" in {
+ // the MIN side resolves through the auto probe, so the MAX side is the one that
+ // reaches the type match and rejects the DOUBLE column
+ val conn = boundaryConn(
+ "score",
+ (rs, _) => (rs.getDouble(_: Int)).expects(1).returning(1.5),
+ sides = Seq("MIN")
+ )
+ val exec =
+ new TestSQLSourceOpExec(progressiveJson("score", Option("auto"), Option("5"), 4L), conn)
+ intercept[IllegalArgumentException](exec.open()).getMessage shouldBe "Unsupported type double"
+ }
+
it should "fetch auto boundaries for every supported batch column type" in {
val intConn = boundaryConn(
"id",
@@ -338,6 +467,15 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
exec.slidingWindow shouldBe " AND score >= 1.5 AND score < 5.5"
exec.slidingWindow shouldBe " AND score >= 5.5 AND score <= 9.0"
exec.nextQueryAvailable shouldBe false
+
+ // the DOUBLE mirror of the INTEGER single-value case: min == max must still yield one
+ // window, which is the only input that makes this arm's `<=` load-bearing
+ val singleValueConn =
+ boundaryConn("score", (rs, _) => (rs.getDouble(_: Int)).expects(1).returning(1.5))
+ val singleValue = openedProgressive("score", "auto", "auto", 4L, singleValueConn)
+ singleValue.nextQueryAvailable shouldBe true
+ singleValue.slidingWindow shouldBe " AND score >= 1.5 AND score <= 1.5"
+ singleValue.nextQueryAvailable shouldBe false
}
it should "reject an auto boundary probe on an unsupported column type" in {
@@ -360,6 +498,15 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
intercept[IllegalStateException](exec.open()).getMessage shouldBe "Unexpected value: string"
}
+ it should "report a zero boundary when no batch column is configured" in {
+ // DEFENSIVE-ONLY arm, not a reachable execution path: fetchBatchByBoundary is called from
+ // exactly two production sites, both inside initBatchColumnBoundaries behind
+ // `batchByAttribute.isDefined`, and AsterixDBSourceOpExec overrides the method outright.
+ // What this pins is the protected contract subclasses inherit, nothing a workflow can run.
+ // no connection is needed: without a batchByAttribute the probe never touches JDBC
+ new TestSQLSourceOpExec(descJson()).boundary("MIN").intValue shouldBe 0
+ }
+
"SQLSourceOpExec.produceTuple" should "stream every row of the single non-progressive query" in {
val resultSet = mock[ResultSet]
val statement = mock[PreparedStatement]
@@ -435,6 +582,45 @@ class SQLSourceOpExecSpec extends AnyFlatSpec with Matchers with MockFactory {
exec.curLimit shouldBe Some(1L)
}
+ it should "bind no fixed offset in progressive mode" in {
+ val resultSet = mock[ResultSet]
+ val statement = mock[PreparedStatement]
+ val conn = mock[Connection]
+
+ // the limit is present so that progressive-ness and "has a limit" are distinguishable
+ // here, and so that the clause order (sliding window ahead of LIMIT) is pinned
+ (conn
+ .prepareStatement(_: String))
+ .expects(BASE + " AND big >= 0 AND big < 30 LIMIT ?;")
+ .returning(statement)
+ // only the limit is bound, at index 1: progressive mode carries its offset in the
+ // sliding window, and the strict mock rejects any further setLong, i.e. any offset
+ (statement.setLong _).expects(1, 2L)
+ (statement.executeQuery: () => ResultSet).expects().returning(resultSet)
+ inSequence {
+ (resultSet.next _).expects().returning(true) // consumed by the manual offset skip
+ (resultSet.next _).expects().returning(true)
+ }
+ (resultSet.getObject(_: String)).expects("id").returning(Int.box(4))
+ (resultSet.getObject(_: String)).expects("name").returning("p")
+ (resultSet.getObject(_: String)).expects("big").returning(Long.box(7L))
+
+ val exec = new TestSQLSourceOpExec(
+ progressiveJson("big", Option("0"), Option("100"), 30L),
+ conn,
+ execSchema = progressiveRowSchema
+ )
+ exec.curOffset = Some(1L)
+ exec.curLimit = Some(2L)
+ exec.open()
+
+ val first = exec.produceTuple().next().asInstanceOf[Tuple]
+ first.getField[Any]("id") shouldBe 4
+ first.getField[Any]("big") shouldBe 7L
+ exec.curOffset shouldBe Some(0L)
+ exec.curLimit shouldBe Some(1L)
+ }
+
it should "yield nothing when the remaining limit already forbids a query" in {
val conn = mock[Connection]
val exec = new TestSQLSourceOpExec(descJson(), conn, execSchema = rowSchema)