You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "aglinxinyuan (via GitHub)" <gi...@apache.org> on 2026/08/01 06:21:05 UTC
[PR] test(amber): cover execution runtime lifecycle [texera]
aglinxinyuan opened a new pull request, #7221:
URL: https://github.com/apache/texera/pull/7221
### What changes were proposed in this PR?
Adds focused execution-runtime lifecycle coverage for pause, resume, kill, completed callbacks, and configured or missing checkpoint storage.
### Any related issues, documentation, discussions?
Closes #7218
### How was this PR tested?
- `WorkflowExecutionService/testOnly org.apache.texera.web.service.ExecutionRuntimeServiceSpec` — 6 passed.
- `WorkflowExecutionService/Test/scalafmtCheck`
- `WorkflowExecutionService/Test/scalafix --check`
Mutation proof (each production mutation was reverted):
| Production mutation | Focused test result |
| --- | --- |
| Changed the pause transition from `PAUSING` to `RUNNING` | 5 passed, 1 failed |
| Removed `performReconfigurationOnResume()` from the resume path | 5 passed, 1 failed |
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): cover execution runtime lifecycle [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #7221:
URL: https://github.com/apache/texera/pull/7221
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): cover execution runtime lifecycle [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #7221:
URL: https://github.com/apache/texera/pull/7221#issuecomment-5150213056
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/7221?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 81.06%. Comparing base ([`050c226`](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`bc72648`](https://app.codecov.io/gh/apache/texera/commit/bc7264876af191af0f167c581c02177958e27fb0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #7221 +/- ##
============================================
+ Coverage 81.02% 81.06% +0.04%
- Complexity 3965 3969 +4
============================================
Files 1162 1162
Lines 46259 46259
Branches 5153 5153
============================================
+ Hits 37480 37499 +19
+ Misses 7105 7087 -18
+ Partials 1674 1673 -1
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | *Carryforward flag |
|---|---|---|---|
| [access-control-service](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.00% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [agent-service](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.49% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [amber](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.71% <ø> (+0.11%)` | :arrow_up: | |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `20.49% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [config-service](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `65.97% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [file-service](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `66.80% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [frontend](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.22% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [notebook-migration-service](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.89% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pyamber](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `97.36% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [workflow-compiling-service](https://app.codecov.io/gh/apache/texera/pull/7221/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.31% <ø> (ø)` | | Carriedforward from [050c226](https://app.codecov.io/gh/apache/texera/commit/050c226a6e9c605c1a9c6ae8adae9920af053def?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
*This pull request uses carry forward flags. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/texera/pull/7221?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): cover execution runtime lifecycle [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #7221:
URL: https://github.com/apache/texera/pull/7221#issuecomment-5150164992
<!-- texera-reviewer-suggestion -->
### Automated Reviewer Suggestions
Based on the `git blame` history of the changed files, we recommend the following reviewers:
- No candidates found from `git blame` history.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): cover execution runtime lifecycle [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #7221:
URL: https://github.com/apache/texera/pull/7221#issuecomment-5150184873
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 2 better · 🔴 3 worse · ⚪ 10 noise (<±5%) · 0 without baseline
> Compared against main `050c226` benchmarked on **this same runner**, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.
[Dashboard](https://apache.github.io/texera/dev/bench/) · [Run](https://github.com/apache/texera/actions/runs/30687580545)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 388 | 0.237 | 25,686/35,212/35,212 us | 🔴 +14.8% / 🔴 +117.4% |
| ⚪ | bs=100 sw=10 sl=64 | 788 | 0.481 | 124,170/152,970/152,970 us | ⚪ within ±5% / 🔴 +38.1% |
| 🟢 | bs=1000 sw=10 sl=64 | 917 | 0.56 | 1,093,866/1,144,885/1,144,885 us | 🟢 -5.7% / 🔴 +7.8% |
<details><summary>Baseline details</summary>
Latest main `050c226` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 388 tuples/sec | 412 tuples/sec | 765.25 tuples/sec | -5.8% | -49.3% |
| bs=10 sw=10 sl=64 | MB/s | 0.237 MB/s | 0.252 MB/s | 0.467 MB/s | -6.0% | -49.3% |
| bs=10 sw=10 sl=64 | p50 | 25,686 us | 22,381 us | 12,703 us | +14.8% | +102.2% |
| bs=10 sw=10 sl=64 | p95 | 35,212 us | 34,060 us | 16,199 us | +3.4% | +117.4% |
| bs=10 sw=10 sl=64 | p99 | 35,212 us | 34,060 us | 19,557 us | +3.4% | +80.0% |
| bs=100 sw=10 sl=64 | throughput | 788 tuples/sec | 821 tuples/sec | 965.22 tuples/sec | -4.0% | -18.4% |
| bs=100 sw=10 sl=64 | MB/s | 0.481 MB/s | 0.501 MB/s | 0.589 MB/s | -4.0% | -18.4% |
| bs=100 sw=10 sl=64 | p50 | 124,170 us | 119,578 us | 103,830 us | +3.8% | +19.6% |
| bs=100 sw=10 sl=64 | p95 | 152,970 us | 149,372 us | 110,778 us | +2.4% | +38.1% |
| bs=100 sw=10 sl=64 | p99 | 152,970 us | 149,372 us | 118,713 us | +2.4% | +28.9% |
| bs=1000 sw=10 sl=64 | throughput | 917 tuples/sec | 917 tuples/sec | 992.33 tuples/sec | 0.0% | -7.6% |
| bs=1000 sw=10 sl=64 | MB/s | 0.56 MB/s | 0.559 MB/s | 0.606 MB/s | +0.2% | -7.5% |
| bs=1000 sw=10 sl=64 | p50 | 1,093,866 us | 1,081,527 us | 1,016,959 us | +1.1% | +7.6% |
| bs=1000 sw=10 sl=64 | p95 | 1,144,885 us | 1,213,947 us | 1,062,501 us | -5.7% | +7.8% |
| bs=1000 sw=10 sl=64 | p99 | 1,144,885 us | 1,213,947 us | 1,082,611 us | -5.7% | +5.8% |
</details>
<details><summary>Raw CSV</summary>
```csv
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,515.40,200,128000,388,0.237,25685.51,35212.42,35212.42
1,100,10,64,20,2537.02,2000,1280000,788,0.481,124169.98,152969.56,152969.56
2,1000,10,64,20,21816.39,20000,12800000,917,0.560,1093866.35,1144884.57,1144884.57
```
</details>
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): cover execution runtime lifecycle [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7221:
URL: https://github.com/apache/texera/pull/7221#discussion_r3694804961
##########
amber/src/test/scala/org/apache/texera/web/service/ExecutionRuntimeServiceSpec.scala:
##########
@@ -0,0 +1,257 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.service
+
+import com.twitter.util.{Future, Promise}
+import io.reactivex.rxjava3.disposables.Disposable
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.testkit.TestKit
+import org.apache.texera.amber.core.workflow.{PhysicalPlan, WorkflowContext}
+import org.apache.texera.amber.engine.architecture.coordinator.ExecutionStateUpdate
+import org.apache.texera.amber.engine.architecture.coordinator.CoordinatorConfig
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
+ EmptyRequest,
+ TakeGlobalCheckpointRequest
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.{
+ EmptyReturn,
+ TakeGlobalCheckpointResponse
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState._
+import org.apache.texera.amber.engine.architecture.rpc.coordinatorservice.CoordinatorServiceFs2Grpc
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.FaultToleranceConfig
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.web.WebsocketInput
+import org.apache.texera.web.model.websocket.request.{
+ WorkflowCheckpointRequest,
+ WorkflowKillRequest,
+ WorkflowPauseRequest,
+ WorkflowResumeRequest
+}
+import org.apache.texera.web.storage.ExecutionStateStore
+import org.scalamock.scalatest.MockFactory
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpecLike
+import org.scalatest.matchers.should.Matchers
+
+import java.net.URI
+import scala.collection.mutable.ListBuffer
+import scala.reflect.ClassTag
+
+class ExecutionRuntimeServiceSpec
+ extends TestKit(ActorSystem("ExecutionRuntimeServiceSpec"))
+ with AnyFlatSpecLike
+ with Matchers
+ with MockFactory
+ with BeforeAndAfterAll {
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
Review Comment:
`afterAll` should call `super.afterAll()` (preferably in a `finally`) so ScalaTest's `BeforeAndAfterAll` lifecycle runs correctly and any parent cleanup isn't skipped. Several other TestKit specs in this repo follow the `try ... finally super.afterAll()` pattern (e.g., `ClientActorSpec`).
##########
amber/src/test/scala/org/apache/texera/web/service/ExecutionRuntimeServiceSpec.scala:
##########
@@ -0,0 +1,257 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.service
+
+import com.twitter.util.{Future, Promise}
+import io.reactivex.rxjava3.disposables.Disposable
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.testkit.TestKit
+import org.apache.texera.amber.core.workflow.{PhysicalPlan, WorkflowContext}
+import org.apache.texera.amber.engine.architecture.coordinator.ExecutionStateUpdate
+import org.apache.texera.amber.engine.architecture.coordinator.CoordinatorConfig
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
+ EmptyRequest,
+ TakeGlobalCheckpointRequest
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.{
+ EmptyReturn,
+ TakeGlobalCheckpointResponse
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState._
+import org.apache.texera.amber.engine.architecture.rpc.coordinatorservice.CoordinatorServiceFs2Grpc
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.FaultToleranceConfig
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.web.WebsocketInput
+import org.apache.texera.web.model.websocket.request.{
+ WorkflowCheckpointRequest,
+ WorkflowKillRequest,
+ WorkflowPauseRequest,
+ WorkflowResumeRequest
+}
+import org.apache.texera.web.storage.ExecutionStateStore
+import org.scalamock.scalatest.MockFactory
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpecLike
+import org.scalatest.matchers.should.Matchers
+
+import java.net.URI
+import scala.collection.mutable.ListBuffer
+import scala.reflect.ClassTag
+
+class ExecutionRuntimeServiceSpec
+ extends TestKit(ActorSystem("ExecutionRuntimeServiceSpec"))
+ with AnyFlatSpecLike
+ with Matchers
+ with MockFactory
+ with BeforeAndAfterAll {
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ /** Real empty-plan client for constructor compatibility, with lifecycle collaborators overridden below. */
+ private final class TestAmberClient(
+ override val coordinatorInterface: CoordinatorServiceFs2Grpc[Future, Unit]
+ ) extends AmberClient(
+ system,
+ new WorkflowContext(),
+ PhysicalPlan(Set.empty, Set.empty),
+ CoordinatorConfig(None, None, None, None),
+ _ => ()
+ ) {
+ var shutdownCount = 0
+ var stateCallback: ExecutionStateUpdate => Unit = _
+
+ override def shutdown(): Unit = {
+ shutdownCount += 1
+ super.shutdown()
+ }
+
+ override def registerCallback[T](callback: T => Unit)(implicit ct: ClassTag[T]): Disposable = {
+ if (ct.runtimeClass == classOf[ExecutionStateUpdate]) {
+ stateCallback = callback.asInstanceOf[ExecutionStateUpdate => Unit]
+ }
+ Disposable.empty()
+ }
+
+ def dispose(): Unit = super.shutdown()
+ }
+
+ /** Records resume wiring while skipping constructor-time client and workflow subscriptions. */
+ private final class RecordingReconfigurationService
+ extends ExecutionReconfigurationService(client = null, stateStore = null, workflow = null) {
+ var resumeCallCount = 0
+
+ override def performReconfigurationOnResume(): Unit = {
+ resumeCallCount += 1
+ }
+
+ override protected def registerWorkerCompletionCallback(): Unit = {}
+
+ override protected def registerCompletionDiffHandler(): Unit = {}
+ }
+
+ private final class Fixture(
+ val client: TestAmberClient,
+ val coordinator: CoordinatorServiceFs2Grpc[Future, Unit],
+ val stateStore: ExecutionStateStore,
+ val wsInput: WebsocketInput,
+ val reconfigurationService: RecordingReconfigurationService,
+ val runtime: ExecutionRuntimeService,
+ val errors: ListBuffer[Throwable]
+ ) {
+ def close(): Unit = {
+ runtime.unsubscribeAll()
+ client.dispose()
+ }
+ }
+
+ private def fixture(
+ logConf: Option[FaultToleranceConfig] = Some(
+ FaultToleranceConfig(new URI("file:///runtime-checkpoints/"))
+ )
+ ): Fixture = {
+ val coordinator = mock[CoordinatorServiceFs2Grpc[Future, Unit]]
+ val client = new TestAmberClient(coordinator)
+ val stateStore = new ExecutionStateStore
+ val errors = ListBuffer.empty[Throwable]
+ val wsInput = new WebsocketInput(errors += _)
+ val reconfigurationService = new RecordingReconfigurationService
+
+ val runtime = new ExecutionRuntimeService(
+ client,
+ stateStore,
+ wsInput,
+ reconfigurationService,
+ logConf,
+ workflowId = 7L,
+ emailNotificationEnabled = false,
+ userEmailOpt = None,
+ sessionUri = new URI("https://texera.example/session")
+ )
+ new Fixture(
+ client,
+ coordinator,
+ stateStore,
+ wsInput,
+ reconfigurationService,
+ runtime,
+ errors
+ )
+ }
+
+ "ExecutionRuntimeService" should "mark the runtime as pausing before sending a pause command" in {
+ val f = fixture()
+ try {
+ (f.coordinator.pauseWorkflow _)
+ .expects(EmptyRequest(), ())
+ .onCall { (_: EmptyRequest, _: Unit) =>
+ f.stateStore.metadataStore.getState.state shouldBe PAUSING
+ Future.value(EmptyReturn())
+ }
+
+ f.wsInput.onNext(WorkflowPauseRequest(), None)
+
+ f.stateStore.metadataStore.getState.state shouldBe PAUSING
+ f.errors shouldBe empty
+ } finally f.close()
+ }
+
+ it should "resume pending reconfiguration and transition to running after the RPC succeeds" in {
+ val f = fixture()
+ try {
+ val resumeResult = Promise[EmptyReturn]()
+ (f.coordinator.resumeWorkflow _)
+ .expects(EmptyRequest(), ())
+ .onCall { (_: EmptyRequest, _: Unit) =>
+ f.reconfigurationService.resumeCallCount shouldBe 1
+ resumeResult
+ }
+
+ f.wsInput.onNext(WorkflowResumeRequest(), None)
+ f.stateStore.metadataStore.getState.state shouldBe RESUMING
+
+ resumeResult.setValue(EmptyReturn())
+ f.stateStore.metadataStore.getState.state shouldBe RUNNING
+ f.errors shouldBe empty
+ } finally f.close()
+ }
+
+ it should "kill the client and record a terminal timestamp when killed from the websocket" in {
+ val f = fixture()
+ try {
+ f.wsInput.onNext(WorkflowKillRequest(), None)
+
+ f.client.shutdownCount shouldBe 1
+ f.stateStore.metadataStore.getState.state shouldBe KILLED
+ f.stateStore.statsStore.getState.endTimeStamp should be > 0L
+ f.errors shouldBe empty
+ } finally f.close()
+ }
+
+ it should "dispatch a non-estimation checkpoint request below the configured log URI" in {
+ val f = fixture()
+ try {
+ var captured: TakeGlobalCheckpointRequest = null
+ (f.coordinator.takeGlobalCheckpoint _)
+ .expects(*, ())
+ .onCall { (request: TakeGlobalCheckpointRequest, _: Unit) =>
+ captured = request
+ Future.value(TakeGlobalCheckpointResponse(0L))
+ }
+
+ f.wsInput.onNext(WorkflowCheckpointRequest(), None)
+
+ captured should not be null
+ captured.estimationOnly shouldBe false
+ captured.checkpointId.id should startWith("Checkpoint_")
+ val destination = new URI(captured.destination)
+ destination.getScheme shouldBe "file"
+ destination shouldBe new URI("file:///runtime-checkpoints/").resolve(
+ captured.checkpointId.toString
+ )
+ f.errors shouldBe empty
+ } finally f.close()
+ }
+
+ it should "report a checkpoint request without fault-tolerance storage instead of dispatching it" in {
+ val f = fixture(logConf = None)
+ try {
+ f.wsInput.onNext(WorkflowCheckpointRequest(), None)
+
+ f.errors should have size 1
+ f.errors.head shouldBe a[AssertionError]
+ f.errors.head.getMessage should include("Fault tolerance log folder is not established")
Review Comment:
This test asserts the *exception type* (`AssertionError`) rather than the observable behavior (an error is reported to the websocket error handler with the expected message). Since the production code could reasonably switch from `assert(...)` to `require(...)`/a domain exception without changing the user-visible behavior, asserting on the message (and that exactly one error was reported) will make the test less brittle while still validating the lifecycle path.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org