This is an automated email from the ASF dual-hosted git repository.
xuang7 pushed a commit to branch release/v1.2
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/release/v1.2 by this push:
new a9fa9bc091 fix(amber, v1.2): order worker state by version, not
timestamp (#7106)
a9fa9bc091 is described below
commit a9fa9bc091e2a8a3861c0d05586c0ab5c5bece8d
Author: Yicong Huang <[email protected]>
AuthorDate: Thu Jul 30 02:30:23 2026 -0400
fix(amber, v1.2): order worker state by version, not timestamp (#7106)
### What changes were proposed in this PR?
Backport of #6011 to `release/v1.2`, cherry-picked from
f5217504c3cfd8ac1d56f6617d7849895af17f59.
Follows the Direct Backport Push convention; opened as a PR (rather than
a direct push) as part of a backport-coverage audit for fixes merged to
`main` since early June that were never labeled for backport.
### Any related issues, documentation, discussions?
Backport of #6011. Originally linked #6010.
### How was this PR tested?
Release-branch CI runs once the conflicts are resolved and this PR is
marked ready for review. The cherry-pick **conflicted** and was
committed with conflict markers.
### Was this PR authored or co-authored using generative AI tooling?
Yes — backport prepared with Claude Code (mechanical cherry-pick;
conflicts left as markers for the original author to resolve; the change
itself is #6011 by its original author).
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../engine/architecture/rpc/controlcommands.proto | 2 +
.../engine/architecture/rpc/controlreturns.proto | 2 +
.../engine/architecture/worker/statistics.proto | 5 +-
.../handlers/control/pause_worker_handler.py | 4 +-
.../handlers/control/query_statistics_handler.py | 4 +-
.../handlers/control/resume_worker_handler.py | 4 +-
.../handlers/control/start_worker_handler.py | 3 +-
.../core/architecture/managers/state_manager.py | 26 ++++++
.../controller/execution/ExecutionUtils.scala | 12 ++-
.../controller/promisehandlers/PauseHandler.scala | 7 +-
.../QueryWorkerStatisticsHandler.scala | 4 +-
.../controller/promisehandlers/ResumeHandler.scala | 2 +-
.../WorkerStateUpdatedHandler.scala | 2 +-
.../deploysemantics/layer/WorkerExecution.scala | 70 ++++++++------
.../scheduling/RegionExecutionCoordinator.scala | 9 +-
.../engine/architecture/worker/DataProcessor.scala | 3 +-
.../worker/promisehandlers/PauseHandler.scala | 3 +-
.../promisehandlers/QueryStatisticsHandler.scala | 3 +-
.../worker/promisehandlers/ResumeHandler.scala | 3 +-
.../worker/promisehandlers/StartHandler.scala | 6 +-
.../common/statetransition/StateManager.scala | 20 ++++
.../architecture/managers/test_state_manager.py | 40 ++++++++
.../test/python/core/runnables/test_main_loop.py | 29 +++++-
.../controller/execution/ExecutionUtilsSpec.scala | 12 +--
.../layer/WorkerExecutionSpec.scala | 103 ++++++++++++++++-----
.../scheduling/RegionCoordinatorTestSupport.scala | 3 +-
.../statetransition/WorkerStateManagerSpec.scala | 41 ++++++++
27 files changed, 333 insertions(+), 89 deletions(-)
diff --git
a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto
b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto
index 1f55927e4a..b4913932a8 100644
---
a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto
+++
b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto
@@ -166,6 +166,8 @@ message PortCompletedRequest {
message WorkerStateUpdatedRequest {
worker.WorkerState state = 1 [(scalapb.field).no_box = true];
+ // Monotonic per-worker version of state, for causal ordering by the
controller.
+ int64 state_version = 2;
}
message LinkWorkersRequest {
diff --git
a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlreturns.proto
b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlreturns.proto
index 43613b5cfd..563c3ee1dd 100644
---
a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlreturns.proto
+++
b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlreturns.proto
@@ -134,6 +134,8 @@ message StartWorkflowResponse {
message WorkerStateResponse {
worker.WorkerState state = 1 [(scalapb.field).no_box = true];
+ // Monotonic per-worker version of state, for causal ordering by the
controller.
+ int64 state_version = 2;
}
message WorkerMetricsResponse {
diff --git
a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/worker/statistics.proto
b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/worker/statistics.proto
index 85d1fcf4aa..4f84dc085f 100644
---
a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/worker/statistics.proto
+++
b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/worker/statistics.proto
@@ -60,4 +60,7 @@ message WorkerStatistics {
message WorkerMetrics {
WorkerState worker_state = 1 [(scalapb.field).no_box = true];
WorkerStatistics worker_statistics = 2 [(scalapb.field).no_box = true];
-}
\ No newline at end of file
+ // Monotonic per-worker version of worker_state, used by the controller to
order
+ // state reports causally instead of by wall-clock timestamp. See
WorkerStateManager.
+ int64 state_version = 3;
+}
diff --git
a/amber/src/main/python/core/architecture/handlers/control/pause_worker_handler.py
b/amber/src/main/python/core/architecture/handlers/control/pause_worker_handler.py
index ef9188914e..6a4d9162a2 100644
---
a/amber/src/main/python/core/architecture/handlers/control/pause_worker_handler.py
+++
b/amber/src/main/python/core/architecture/handlers/control/pause_worker_handler.py
@@ -26,5 +26,5 @@ from proto.org.apache.texera.amber.engine.architecture.rpc
import (
class PauseWorkerHandler(ControlHandler):
async def pause_worker(self, req: EmptyRequest) -> WorkerStateResponse:
self.context.pause_manager.pause(PauseType.USER_PAUSE)
- state = self.context.state_manager.get_current_state()
- return WorkerStateResponse(state)
+ state, state_version =
self.context.state_manager.get_state_with_version()
+ return WorkerStateResponse(state, state_version=state_version)
diff --git
a/amber/src/main/python/core/architecture/handlers/control/query_statistics_handler.py
b/amber/src/main/python/core/architecture/handlers/control/query_statistics_handler.py
index b636249d71..b982626827 100644
---
a/amber/src/main/python/core/architecture/handlers/control/query_statistics_handler.py
+++
b/amber/src/main/python/core/architecture/handlers/control/query_statistics_handler.py
@@ -27,8 +27,10 @@ from
proto.org.apache.texera.amber.engine.architecture.worker import (
class QueryStatisticsHandler(ControlHandler):
async def query_statistics(self, req: EmptyRequest) ->
WorkerMetricsResponse:
+ state, state_version =
self.context.state_manager.get_state_with_version()
metrics = WorkerMetrics(
- worker_state=self.context.state_manager.get_current_state(),
+ worker_state=state,
worker_statistics=self.context.statistics_manager.get_statistics(),
+ state_version=state_version,
)
return WorkerMetricsResponse(metrics)
diff --git
a/amber/src/main/python/core/architecture/handlers/control/resume_worker_handler.py
b/amber/src/main/python/core/architecture/handlers/control/resume_worker_handler.py
index 3ebaadb661..5dca1fdc28 100644
---
a/amber/src/main/python/core/architecture/handlers/control/resume_worker_handler.py
+++
b/amber/src/main/python/core/architecture/handlers/control/resume_worker_handler.py
@@ -26,5 +26,5 @@ from proto.org.apache.texera.amber.engine.architecture.rpc
import (
class ResumeWorkerHandler(ControlHandler):
async def resume_worker(self, req: EmptyRequest) -> WorkerStateResponse:
self.context.pause_manager.resume(PauseType.USER_PAUSE)
- state = self.context.state_manager.get_current_state()
- return WorkerStateResponse(state)
+ state, state_version =
self.context.state_manager.get_state_with_version()
+ return WorkerStateResponse(state, state_version=state_version)
diff --git
a/amber/src/main/python/core/architecture/handlers/control/start_worker_handler.py
b/amber/src/main/python/core/architecture/handlers/control/start_worker_handler.py
index bfa1556722..b1e66ef02f 100644
---
a/amber/src/main/python/core/architecture/handlers/control/start_worker_handler.py
+++
b/amber/src/main/python/core/architecture/handlers/control/start_worker_handler.py
@@ -97,4 +97,5 @@ class StartWorkerHandler(ControlHandler):
elif self.context.input_manager.get_input_port_mat_reader_threads():
self.context.input_manager.start_input_port_mat_reader_threads()
- return
WorkerStateResponse(self.context.state_manager.get_current_state())
+ state, state_version =
self.context.state_manager.get_state_with_version()
+ return WorkerStateResponse(state, state_version=state_version)
diff --git a/amber/src/main/python/core/architecture/managers/state_manager.py
b/amber/src/main/python/core/architecture/managers/state_manager.py
index e80b6d5fc4..7cd5a16e2d 100644
--- a/amber/src/main/python/core/architecture/managers/state_manager.py
+++ b/amber/src/main/python/core/architecture/managers/state_manager.py
@@ -37,6 +37,13 @@ class StateManager:
def __init__(self, state_transition_graph: Dict[T, Set[T]], initial_state:
T):
self._state_transition_graph = state_transition_graph
self._current_state: T = initial_state
+ # Monotonically increasing version, bumped on every successful
transition.
+ # It is the state machine's logical clock: since a single owner drives
all
+ # transitions, the version totally orders them causally. Reporting it
+ # alongside the state lets the controller reject stale state reports
that
+ # arrive out of order, without comparing wall-clock timestamps across
+ # processes. Must mirror the Scala StateManager.
+ self._state_version: int = 0
def assert_state(self, state: T) -> None:
"""
@@ -75,6 +82,7 @@ class StateManager:
)
self._current_state = state
+ self._state_version += 1
def get_current_state(self) -> T:
"""
@@ -82,3 +90,21 @@ class StateManager:
:return:
"""
return self._current_state
+
+ def get_state_version(self) -> int:
+ """
+ Return the monotonic version of the current state.
+ :return:
+ """
+ return self._state_version
+
+ def get_state_with_version(self) -> Tuple[T, int]:
+ """
+ Return the current state together with its version as one pair, so a
+ report site cannot interleave a transition between reading the state
and
+ reading the version. Every state-report site must use this instead of
+ separate get_current_state/get_state_version calls. Mirrors the Scala
+ StateManager's getStateWithVersion.
+ :return:
+ """
+ return self._current_state, self._state_version
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/execution/ExecutionUtils.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/execution/ExecutionUtils.scala
index 7ee9bc0473..4dd161c484 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/execution/ExecutionUtils.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/execution/ExecutionUtils.scala
@@ -91,10 +91,14 @@ object ExecutionUtils {
readyState: T
): WorkflowAggregatedState = {
states match {
- case _ if states.isEmpty =>
WorkflowAggregatedState.UNINITIALIZED
- case _ if states.forall(_ == completedState) =>
WorkflowAggregatedState.COMPLETED
- case _ if states.forall(_ == terminatedState) =>
WorkflowAggregatedState.COMPLETED
- case _ if states.exists(_ == runningState) =>
WorkflowAggregatedState.RUNNING
+ case _ if states.isEmpty => WorkflowAggregatedState.UNINITIALIZED
+ // Every state is terminal: COMPLETED reported by the worker itself, or
TERMINATED
+ // stamped by controller-side teardown (which only runs after the region
completed,
+ // and preserves already-COMPLETED workers). A mixed terminal set is
therefore a
+ // routine end-of-region shape and means the execution is over.
+ case _ if states.forall(s => s == completedState || s ==
terminatedState) =>
+ WorkflowAggregatedState.COMPLETED
+ case _ if states.exists(_ == runningState) =>
WorkflowAggregatedState.RUNNING
case _ =>
val unCompletedStates = states.filter(_ != completedState)
if (unCompletedStates.forall(_ == uninitializedState)) {
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/PauseHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/PauseHandler.scala
index 35a85f56ae..eb13d92a0d 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/PauseHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/PauseHandler.scala
@@ -68,13 +68,16 @@ trait PauseHandler {
// send a pause message
workerInterface.pauseWorker(EmptyRequest(),
mkContext(worker)).flatMap {
resp =>
- workerExecution.update(System.nanoTime(), resp.state)
+ workerExecution.updateState(resp.stateVersion,
resp.state)
workerInterface
.queryStatistics(EmptyRequest(), mkContext(worker))
// get the stats and current input tuple from the
worker
.map {
case WorkerMetricsResponse(metrics) =>
- workerExecution.update(System.nanoTime(),
metrics.workerStatistics)
+ workerExecution.updateStats(
+ System.nanoTime(),
+ metrics.workerStatistics
+ )
}
}
}.toSeq
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/QueryWorkerStatisticsHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/QueryWorkerStatisticsHandler.scala
index 6551579f71..e437709b19 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/QueryWorkerStatisticsHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/QueryWorkerStatisticsHandler.scala
@@ -175,7 +175,9 @@ trait QueryWorkerStatisticsHandler {
processLayers(layers).map { _ =>
collectedResults.foreach {
case (wExec, resp, timestamp) =>
- wExec.update(timestamp, resp.metrics.workerState,
resp.metrics.workerStatistics)
+ // State is ordered by the worker's logical version; stats by
receipt time.
+ wExec.updateState(resp.metrics.stateVersion,
resp.metrics.workerState)
+ wExec.updateStats(timestamp, resp.metrics.workerStatistics)
}
forwardStats(msg.updateTarget)
// Record the completion timestamp before releasing the lock so that any
timer
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/ResumeHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/ResumeHandler.scala
index c94ba91c20..f1c173b9ca 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/ResumeHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/ResumeHandler.scala
@@ -52,7 +52,7 @@ trait ResumeHandler {
cp.workflowExecution
.getLatestOperatorExecution(VirtualIdentityUtils.getPhysicalOpId(workerId))
.getWorkerExecution(workerId)
- .update(System.nanoTime(), resp.state)
+ .updateState(resp.stateVersion, resp.state)
}
}
.toSeq
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/WorkerStateUpdatedHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/WorkerStateUpdatedHandler.scala
index 5ee98a4918..7226d93e0c 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/WorkerStateUpdatedHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/promisehandlers/WorkerStateUpdatedHandler.scala
@@ -49,7 +49,7 @@ trait WorkerStateUpdatedHandler {
.find(_.hasOperatorExecution(physicalOpId))
.map(_.getOperatorExecution(physicalOpId))
.foreach(operatorExecution =>
-
operatorExecution.getWorkerExecution(ctx.sender).update(System.nanoTime(),
msg.state)
+
operatorExecution.getWorkerExecution(ctx.sender).updateState(msg.stateVersion,
msg.state)
)
val stats = cp.workflowExecution.getAllRegionExecutionsStats
sendToClient(ExecutionStatsUpdate(stats))
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecution.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecution.scala
index 55e1e30918..4e46e8b124 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecution.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecution.scala
@@ -21,7 +21,11 @@ package
org.apache.texera.amber.engine.architecture.deploysemantics.layer
import org.apache.texera.amber.core.workflow.PortIdentity
import
org.apache.texera.amber.engine.architecture.controller.execution.WorkerPortExecution
-import
org.apache.texera.amber.engine.architecture.worker.statistics.WorkerState.UNINITIALIZED
+import
org.apache.texera.amber.engine.architecture.worker.statistics.WorkerState.{
+ COMPLETED,
+ TERMINATED,
+ UNINITIALIZED
+}
import
org.apache.texera.amber.engine.architecture.worker.statistics.{WorkerState,
WorkerStatistics}
import scala.collection.mutable
@@ -37,50 +41,56 @@ case class WorkerExecution() extends Serializable {
private var stats: WorkerStatistics = {
WorkerStatistics(Seq.empty, Seq.empty, 0, 0, 0)
}
- private var lastUpdateTimeStamp = 0L
+ // Logical version of the last applied state, sourced from the worker's
+ // WorkerStateManager. Starts below any real version so the first report
applies.
+ private var lastStateVersion = -1L
+ // Controller-side receipt time (System.nanoTime) of the last applied stats
snapshot.
+ private var lastStatsTimeStamp = 0L
+
+ private def isTerminal(s: WorkerState): Boolean = s == COMPLETED || s ==
TERMINATED
/**
- * Updates both the worker state and statistics if the provided timestamp
is newer
- * than the last recorded update timestamp. This ensures that only the most
recent
- * data is reflected in the execution state.
+ * Applies a worker state report, ordered causally by the worker's monotonic
+ * `stateVersion` rather than by receipt time. A report is applied only
when it
+ * is strictly newer than the last applied one, so a stale state that
arrives late
+ * (e.g. the RUNNING snapshot carried by a slow startWorker response)
cannot clobber
+ * a newer state. In addition, terminal states (COMPLETED/TERMINATED) are
absorbing:
+ * once reached, no later report can move the worker out of them.
*
- * @param timeStamp the nanosecond-timestamp of this update
- * @param state the new WorkerState to set
- * @param stats the new WorkerStatistics to set
+ * @param stateVersion the worker-side monotonic version of this state
+ * @param newState the reported WorkerState
*/
- def update(timeStamp: Long, state: WorkerState, stats: WorkerStatistics):
Unit = {
- if (this.lastUpdateTimeStamp < timeStamp) {
- this.stats = stats
- this.state = state
- this.lastUpdateTimeStamp = timeStamp
+ def updateState(stateVersion: Long, newState: WorkerState): Unit = {
+ if (isTerminal(this.state)) {
+ return
+ }
+ if (this.lastStateVersion < stateVersion) {
+ this.state = newState
+ this.lastStateVersion = stateVersion
}
}
/**
- * Updates only the worker state if the provided timestamp is newer than the
- * last recorded update timestamp.
- *
- * @param timeStamp the nanosecond-timestamp of this update
- * @param state the new WorkerState to set
+ * Forces the worker into TERMINATED, e.g. when the controller kills a
region. This
+ * still respects terminal-state absorption: a worker that already
COMPLETED on its
+ * own is left as COMPLETED. Uses the maximum version so it wins over any
in-flight
+ * non-terminal report for a worker that had not yet reached a terminal
state.
*/
- def update(timeStamp: Long, state: WorkerState): Unit = {
- if (this.lastUpdateTimeStamp < timeStamp) {
- this.state = state
- this.lastUpdateTimeStamp = timeStamp
- }
- }
+ def forceTerminate(): Unit = updateState(Long.MaxValue, TERMINATED)
/**
* Updates only the worker statistics if the provided timestamp is newer
than the
- * last recorded update timestamp.
+ * last recorded stats timestamp. Stats are monotonic snapshots, so
newest-wins by
+ * receipt time is sufficient (and necessary, since two snapshots taken
within the
+ * same state share a state version).
*
* @param timeStamp the nanosecond-timestamp of this update
- * @param stats the new WorkerStatistics to set
+ * @param newStats the new WorkerStatistics to set
*/
- def update(timeStamp: Long, stats: WorkerStatistics): Unit = {
- if (this.lastUpdateTimeStamp < timeStamp) {
- this.stats = stats
- this.lastUpdateTimeStamp = timeStamp
+ def updateStats(timeStamp: Long, newStats: WorkerStatistics): Unit = {
+ if (this.lastStatsTimeStamp < timeStamp) {
+ this.stats = newStats
+ this.lastStatsTimeStamp = timeStamp
}
}
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala
index 35e58eed7a..510893e82c 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala
@@ -51,7 +51,6 @@ import
org.apache.texera.amber.engine.architecture.scheduling.config.{
ResourceConfig
}
import
org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.Partitioning
-import
org.apache.texera.amber.engine.architecture.worker.statistics.WorkerState
import org.apache.texera.amber.engine.common.AmberLogging
import org.apache.texera.amber.engine.common.FutureBijection._
import org.apache.texera.amber.engine.common.rpc.AsyncRPCClient
@@ -219,7 +218,7 @@ class RegionExecutionCoordinator(
regionExecution.getAllOperatorExecutions.foreach {
case (_, opExec) =>
opExec.getWorkerIds.foreach { workerId =>
- opExec.getWorkerExecution(workerId).update(System.nanoTime(),
WorkerState.TERMINATED)
+ opExec.getWorkerExecution(workerId).forceTerminate()
}
}
Future.Unit // propagate success
@@ -587,12 +586,14 @@ class RegionExecutionCoordinator(
asyncRPCClient.workerInterface
.startWorker(EmptyRequest(),
asyncRPCClient.mkContext(workerId))
.map(resp =>
- // update worker state
+ // Update worker state, ordered by the worker's logical
state version
+ // (not arrival time) so this RUNNING snapshot cannot
clobber a later
+ // COMPLETED if the response arrives after the worker has
finished.
workflowExecution
.getRegionExecution(region.id)
.getOperatorExecution(opId)
.getWorkerExecution(workerId)
- .update(System.nanoTime(), resp.state)
+ .updateState(resp.stateVersion, resp.state)
)
}
}
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala
index 84f1e8ec65..2a51063d8e 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala
@@ -212,8 +212,9 @@ class DataProcessor(
READY,
RUNNING,
() => {
+ val (state, stateVersion) = stateManager.getStateWithVersion
asyncRPCClient.controllerInterface.workerStateUpdated(
- WorkerStateUpdatedRequest(stateManager.getCurrentState),
+ WorkerStateUpdatedRequest(state, stateVersion),
asyncRPCClient.mkContext(CONTROLLER)
)
}
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/PauseHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/PauseHandler.scala
index cec7ca87e6..548d13c036 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/PauseHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/PauseHandler.scala
@@ -46,7 +46,8 @@ trait PauseHandler {
dp.pauseManager.pause(UserPause)
dp.stateManager.transitTo(PAUSED)
}
- WorkerStateResponse(dp.stateManager.getCurrentState)
+ val (state, stateVersion) = dp.stateManager.getStateWithVersion
+ WorkerStateResponse(state, stateVersion)
}
}
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/QueryStatisticsHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/QueryStatisticsHandler.scala
index 74d8d14faa..feafd74a69 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/QueryStatisticsHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/QueryStatisticsHandler.scala
@@ -35,7 +35,8 @@ trait QueryStatisticsHandler {
request: EmptyRequest,
ctx: AsyncRPCContext
): Future[WorkerMetricsResponse] = {
- WorkerMetricsResponse(WorkerMetrics(dp.stateManager.getCurrentState,
dp.collectStatistics()))
+ val (state, stateVersion) = dp.stateManager.getStateWithVersion
+ WorkerMetricsResponse(WorkerMetrics(state, dp.collectStatistics(),
stateVersion))
}
}
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/ResumeHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/ResumeHandler.scala
index 434c50c914..0ce6e8d076 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/ResumeHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/ResumeHandler.scala
@@ -43,7 +43,8 @@ trait ResumeHandler {
dp.stateManager.transitTo(RUNNING)
dp.adaptiveBatchingMonitor.resumeAdaptiveBatching()
}
- WorkerStateResponse(dp.stateManager.getCurrentState)
+ val (state, stateVersion) = dp.stateManager.getStateWithVersion
+ WorkerStateResponse(state, stateVersion)
}
}
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala
index 5d1bf8ccbd..727eb7d025 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala
@@ -56,12 +56,14 @@ trait StartHandler {
dp.inputGateway.getChannel(channelId).setPortId(PortIdentity())
startChannel(request, ctx)
endChannel(request, ctx)
- WorkerStateResponse(dp.stateManager.getCurrentState)
+ val (state, stateVersion) = dp.stateManager.getStateWithVersion
+ WorkerStateResponse(state, stateVersion)
} else if (dp.inputManager.getInputPortReaderThreads.nonEmpty) {
// This means the worker should read from materialized storage for its
input ports.
// Start the reader threads
dp.inputManager.startInputPortReaderThreads()
- WorkerStateResponse(dp.stateManager.getCurrentState)
+ val (state, stateVersion) = dp.stateManager.getStateWithVersion
+ WorkerStateResponse(state, stateVersion)
} else {
throw new WorkflowRuntimeException(
s"non-source worker $actorId received unexpected StartWorker!"
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/common/statetransition/StateManager.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/common/statetransition/StateManager.scala
index 9bd951389e..dd30a9de8f 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/common/statetransition/StateManager.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/common/statetransition/StateManager.scala
@@ -43,6 +43,25 @@ class StateManager[T](
private var currentState: T = initialState
+ // Monotonically increasing version, bumped on every successful state
transition.
+ // It is the state machine's logical clock: because a single owner drives all
+ // transitions, the version totally orders them in causal order. Reporting
this
+ // version alongside the state lets remote observers (e.g. the controller)
reject
+ // stale state reports that arrive out of order, without relying on
wall-clock
+ // timestamps that cannot be compared across processes. Must mirror the
Python
+ // StateManager
(amber/src/main/python/core/architecture/managers/state_manager.py).
+ private var stateVersion: Long = 0L
+
+ def getStateVersion: Long = stateVersion
+
+ /**
+ * Returns the current state together with its version as one pair, so a
report
+ * site cannot interleave a transition between reading the state and
reading the
+ * version. Every state-report site must use this instead of separate
+ * getCurrentState/getStateVersion calls.
+ */
+ def getStateWithVersion: (T, Long) = (currentState, stateVersion)
+
def assertState(state: T): Unit = {
if (currentState != state) {
throw InvalidStateException(
@@ -84,6 +103,7 @@ class StateManager[T](
throw InvalidTransitionException(s"cannot transit from $currentState to
$state", actorId)
}
currentState = state
+ stateVersion += 1
}
}
diff --git
a/amber/src/test/python/core/architecture/managers/test_state_manager.py
b/amber/src/test/python/core/architecture/managers/test_state_manager.py
index 464c64194f..902cdb7475 100644
--- a/amber/src/test/python/core/architecture/managers/test_state_manager.py
+++ b/amber/src/test/python/core/architecture/managers/test_state_manager.py
@@ -72,3 +72,43 @@ class TestStateManager:
state_manager.transit_to(WorkerState.READY)
state_manager.transit_to(WorkerState.COMPLETED)
state_manager.assert_state(WorkerState.COMPLETED)
+
+ def test_state_version_starts_at_zero(self, state_manager):
+ assert state_manager.get_state_version() == 0
+
+ def test_state_version_bumps_on_every_successful_transition(self,
state_manager):
+ # The controller relies on this monotonic version to order
Python-worker
+ # state reports causally; without it, RUNNING -> PAUSED -> RUNNING
during
+ # reconfiguration would be dropped as stale. Mirrors the Scala
StateManager.
+ assert state_manager.get_state_version() == 0
+ state_manager.transit_to(WorkerState.READY)
+ assert state_manager.get_state_version() == 1
+ state_manager.transit_to(WorkerState.RUNNING)
+ assert state_manager.get_state_version() == 2
+ state_manager.transit_to(WorkerState.COMPLETED)
+ assert state_manager.get_state_version() == 3
+
+ def test_state_version_does_not_bump_on_noop_self_transition(self,
state_manager):
+ state_manager.transit_to(WorkerState.READY)
+ before = state_manager.get_state_version()
+ state_manager.transit_to(WorkerState.READY) # no-op
+ assert state_manager.get_state_version() == before
+
+ def test_state_version_does_not_bump_on_rejected_transition(self,
state_manager):
+ # UNINITIALIZED -> RUNNING is illegal (must pass through READY).
+ with pytest.raises(InvalidTransitionException):
+ state_manager.transit_to(WorkerState.RUNNING)
+ assert state_manager.get_state_version() == 0
+
+ def test_get_state_with_version_returns_matching_pair(self, state_manager):
+ # Report sites read state and version through this single accessor so
the
+ # pair can never come from two different transitions. Mirrors the Scala
+ # StateManager's getStateWithVersion.
+ assert state_manager.get_state_with_version() == (
+ WorkerState.UNINITIALIZED,
+ 0,
+ )
+ state_manager.transit_to(WorkerState.READY)
+ assert state_manager.get_state_with_version() == (WorkerState.READY, 1)
+ state_manager.transit_to(WorkerState.RUNNING)
+ assert state_manager.get_state_with_version() == (WorkerState.RUNNING,
2)
diff --git a/amber/src/test/python/core/runnables/test_main_loop.py
b/amber/src/test/python/core/runnables/test_main_loop.py
index c32f45b888..772b000df5 100644
--- a/amber/src/test/python/core/runnables/test_main_loop.py
+++ b/amber/src/test/python/core/runnables/test_main_loop.py
@@ -666,6 +666,8 @@ class TestMainLoop:
stats_invocation = elem.payload.return_invocation
worker_metrics_response =
stats_invocation.return_value.worker_metrics_response
stats = worker_metrics_response.metrics.worker_statistics
+ # a missing/dropped version would echo through the read-back below;
guard it
+ assert worker_metrics_response.metrics.state_version > 0
metrics = WorkerMetrics(
worker_state=WorkerState.RUNNING,
@@ -692,6 +694,9 @@ class TestMainLoop:
control_processing_time=stats.control_processing_time,
idle_time=stats.idle_time,
),
+ # version is the worker's logical state clock; read it from the
actual
+ # report rather than pinning a brittle count (covered by
StateManager tests).
+ state_version=worker_metrics_response.metrics.state_version,
)
assert elem == DCMElement(
@@ -1099,13 +1104,21 @@ class TestMainLoop:
output_queue,
):
input_queue.put(mock_pause)
- assert output_queue.get() == DCMElement(
+ elem = output_queue.get()
+ # version is the worker's logical state clock; read it from the actual
+ # report rather than pinning a brittle count (covered by StateManager
tests).
+ state_version =
elem.payload.return_invocation.return_value.worker_state_response.state_version
+ # a missing/dropped version would echo through the read-back; guard it
+ assert state_version > 0
+ assert elem == DCMElement(
tag=mock_control_output_channel,
payload=DirectControlMessagePayloadV2(
return_invocation=ReturnInvocation(
command_id=command_sequence,
return_value=ControlReturn(
-
worker_state_response=WorkerStateResponse(WorkerState.PAUSED)
+ worker_state_response=WorkerStateResponse(
+ WorkerState.PAUSED, state_version=state_version
+ )
),
)
),
@@ -1120,13 +1133,21 @@ class TestMainLoop:
output_queue,
):
input_queue.put(mock_resume)
- assert output_queue.get() == DCMElement(
+ elem = output_queue.get()
+ # version is the worker's logical state clock; read it from the actual
+ # report rather than pinning a brittle count (covered by StateManager
tests).
+ state_version =
elem.payload.return_invocation.return_value.worker_state_response.state_version
+ # a missing/dropped version would echo through the read-back; guard it
+ assert state_version > 0
+ assert elem == DCMElement(
tag=mock_control_output_channel,
payload=DirectControlMessagePayloadV2(
return_invocation=ReturnInvocation(
command_id=command_sequence,
return_value=ControlReturn(
-
worker_state_response=WorkerStateResponse(WorkerState.RUNNING)
+ worker_state_response=WorkerStateResponse(
+ WorkerState.RUNNING, state_version=state_version
+ )
),
)
),
diff --git
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/controller/execution/ExecutionUtilsSpec.scala
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/controller/execution/ExecutionUtilsSpec.scala
index cf07c22843..d9405dad41 100644
---
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/controller/execution/ExecutionUtilsSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/controller/execution/ExecutionUtilsSpec.scala
@@ -91,12 +91,12 @@ class ExecutionUtilsSpec extends AnyFlatSpec {
// Anti / boundary cases — make sure unexpected inputs cannot smuggle in a
wrong
// state, and that branch precedence is what the contract claims.
- it should "return UNKNOWN when completed and terminated are mixed (neither
forall branch matches)" in {
- // Both `forall(_ == completed)` and `forall(_ == terminated)` fail, no
running
- // sentinel is present, and the non-completed remainder is purely
terminated —
- // which is none of uninitialized / paused / ready, so the result must be
- // UNKNOWN rather than COMPLETED.
- assert(aggregate(Completed, Terminated) == WorkflowAggregatedState.UNKNOWN)
+ it should "return COMPLETED when completed and terminated are mixed (all
states terminal)" in {
+ // Routine after region teardown: workers whose final COMPLETED report
reached the
+ // controller stay COMPLETED (terminal absorption), while stragglers are
stamped
+ // TERMINATED by forceTerminate. All workers being terminal means the
execution is
+ // over, so the aggregate must be COMPLETED rather than UNKNOWN.
+ assert(aggregate(Completed, Terminated) ==
WorkflowAggregatedState.COMPLETED)
}
it should "give running precedence over completed and terminated" in {
diff --git
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecutionSpec.scala
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecutionSpec.scala
index 2c39954d04..e2ac39595d 100644
---
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecutionSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecutionSpec.scala
@@ -36,51 +36,110 @@ class WorkerExecutionSpec extends AnyFlatSpec {
assert(we.getStats.controlProcessingTime == 0L)
}
- "WorkerExecution.update(state)" should "apply when the timestamp is newer"
in {
+ // ---- state ordering by logical version ----
+
+ "WorkerExecution.updateState" should "apply a state with a newer version" in
{
val we = WorkerExecution()
- we.update(timeStamp = 10L, state = WorkerState.RUNNING)
+ we.updateState(stateVersion = 1L, WorkerState.READY)
+ we.updateState(stateVersion = 2L, WorkerState.RUNNING)
assert(we.getState == WorkerState.RUNNING)
}
- it should "ignore updates with a non-newer timestamp" in {
+ it should "apply the very first report (version 0)" in {
val we = WorkerExecution()
- we.update(timeStamp = 10L, state = WorkerState.RUNNING)
- we.update(timeStamp = 10L, state = WorkerState.PAUSED) // not strictly
newer
- we.update(timeStamp = 5L, state = WorkerState.COMPLETED) // older
- assert(we.getState == WorkerState.RUNNING)
+ we.updateState(stateVersion = 0L, WorkerState.READY)
+ assert(we.getState == WorkerState.READY)
}
- "WorkerExecution.update(state, stats)" should "update both atomically when
newer" in {
+ it should "ignore a stale report whose version is lower (out-of-order
arrival)" in {
val we = WorkerExecution()
- we.update(timeStamp = 10L, state = WorkerState.RUNNING, stats = stats(idle
= 7L))
- assert(we.getState == WorkerState.RUNNING)
- assert(we.getStats.idleTime == 7L)
+ we.updateState(stateVersion = 5L, WorkerState.PAUSED)
+ we.updateState(stateVersion = 4L, WorkerState.RUNNING) // arrives late,
older version
+ assert(we.getState == WorkerState.PAUSED)
}
- it should "ignore updates with a non-newer timestamp" in {
+ it should "ignore a report whose version is not strictly newer" in {
val we = WorkerExecution()
- we.update(timeStamp = 10L, state = WorkerState.RUNNING, stats = stats(idle
= 7L))
- we.update(timeStamp = 5L, state = WorkerState.COMPLETED, stats =
stats(idle = 99L))
+ we.updateState(stateVersion = 3L, WorkerState.RUNNING)
+ we.updateState(stateVersion = 3L, WorkerState.PAUSED) // same version
assert(we.getState == WorkerState.RUNNING)
- assert(we.getStats.idleTime == 7L)
}
- "WorkerExecution.update(stats)" should "update only the stats when newer" in
{
+ // ---- terminal-state absorption ----
+
+ it should "treat COMPLETED as absorbing even against a higher-version
report" in {
+ val we = WorkerExecution()
+ we.updateState(stateVersion = 2L, WorkerState.COMPLETED)
+ we.updateState(stateVersion = 99L, WorkerState.RUNNING) // higher version,
but illegal
+ assert(we.getState == WorkerState.COMPLETED)
+ }
+
+ it should "treat TERMINATED as absorbing" in {
+ val we = WorkerExecution()
+ we.updateState(stateVersion = 2L, WorkerState.TERMINATED)
+ we.updateState(stateVersion = 99L, WorkerState.RUNNING)
+ assert(we.getState == WorkerState.TERMINATED)
+ }
+
+ // Regression for issue #6010: a fast source's startWorker response carries
a stale
+ // RUNNING snapshot that can reach the controller AFTER COMPLETED was
recorded. With
+ // wall-clock ordering the late RUNNING won and the operator was stuck
orange; with
+ // version ordering (and terminal absorption) COMPLETED must survive.
+ it should "not let a late startWorker RUNNING snapshot clobber COMPLETED
(#6010)" in {
+ val we = WorkerExecution()
+ we.updateState(stateVersion = 1L, WorkerState.READY)
+ we.updateState(stateVersion = 3L, WorkerState.COMPLETED) // via completion
stats query
+ we.updateState(stateVersion = 2L, WorkerState.RUNNING) // late startWorker
response
+ assert(we.getState == WorkerState.COMPLETED)
+ }
+
+ // ---- forceTerminate ----
+
+ "WorkerExecution.forceTerminate" should "move a non-terminal worker to
TERMINATED" in {
+ val we = WorkerExecution()
+ we.updateState(stateVersion = 2L, WorkerState.RUNNING)
+ we.forceTerminate()
+ assert(we.getState == WorkerState.TERMINATED)
+ }
+
+ it should "leave a worker that already COMPLETED as COMPLETED" in {
val we = WorkerExecution()
- we.update(timeStamp = 10L, state = WorkerState.RUNNING, stats = stats(idle
= 7L))
- we.update(timeStamp = 20L, stats = stats(idle = 42L))
+ we.updateState(stateVersion = 3L, WorkerState.COMPLETED)
+ we.forceTerminate()
+ assert(we.getState == WorkerState.COMPLETED)
+ }
+
+ // ---- stats ordering by timestamp ----
+
+ "WorkerExecution.updateStats" should "apply newer stats and keep state
untouched" in {
+ val we = WorkerExecution()
+ we.updateState(stateVersion = 2L, WorkerState.RUNNING)
+ we.updateStats(timeStamp = 10L, stats(idle = 7L))
+ we.updateStats(timeStamp = 20L, stats(idle = 42L))
assert(we.getState == WorkerState.RUNNING)
assert(we.getStats.idleTime == 42L)
}
- it should "ignore stats updates with a non-newer timestamp" in {
+ it should "ignore stats with a non-newer timestamp" in {
val we = WorkerExecution()
- we.update(timeStamp = 20L, stats = stats(idle = 42L))
- we.update(timeStamp = 20L, stats = stats(idle = 99L)) // not strictly newer
- we.update(timeStamp = 5L, stats = stats(idle = 0L)) // older
+ we.updateStats(timeStamp = 20L, stats(idle = 42L))
+ we.updateStats(timeStamp = 20L, stats(idle = 99L)) // not strictly newer
+ we.updateStats(timeStamp = 5L, stats(idle = 0L)) // older
assert(we.getStats.idleTime == 42L)
}
+ it should "track state version and stats timestamp independently" in {
+ val we = WorkerExecution()
+ // A high stats timestamp must not block a later (higher-version) state
update,
+ // and vice versa: the two orderings are independent.
+ we.updateStats(timeStamp = 1000L, stats(idle = 1L))
+ we.updateState(stateVersion = 1L, WorkerState.RUNNING)
+ assert(we.getState == WorkerState.RUNNING)
+ assert(we.getStats.idleTime == 1L)
+ }
+
+ // ---- port executions ----
+
"WorkerExecution.getInputPortExecution" should "lazily create and reuse a
port execution per port id" in {
val we = WorkerExecution()
val first = we.getInputPortExecution(PortIdentity(0))
diff --git
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala
index 64d85972ff..a6ded77587 100644
---
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.scala
@@ -140,7 +140,8 @@ object RegionCoordinatorTestSupport {
case InitializeExecutor | OpenExecutor =>
Some(EmptyReturn())
case StartWorker =>
- Some(WorkerStateResponse(WorkerState.RUNNING))
+ // RUNNING is the worker's 2nd transition (UNINITIALIZED -> READY ->
RUNNING).
+ Some(WorkerStateResponse(WorkerState.RUNNING, stateVersion = 2L))
case EndWorker =>
endWorkerResponse(call)
case other =>
diff --git
a/amber/src/test/scala/org/apache/texera/amber/engine/common/statetransition/WorkerStateManagerSpec.scala
b/amber/src/test/scala/org/apache/texera/amber/engine/common/statetransition/WorkerStateManagerSpec.scala
index 751c029ee4..e4b6d375fc 100644
---
a/amber/src/test/scala/org/apache/texera/amber/engine/common/statetransition/WorkerStateManagerSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/common/statetransition/WorkerStateManagerSpec.scala
@@ -128,4 +128,45 @@ class WorkerStateManagerSpec extends AnyFlatSpec {
sm.transitTo(COMPLETED)
}
}
+
+ // -- State version (logical clock) --
+
+ it should "start the state version at 0" in {
+ assert(newManager(UNINITIALIZED).getStateVersion == 0L)
+ }
+
+ it should "bump the state version on every successful transition" in {
+ val sm = newManager(UNINITIALIZED)
+ assert(sm.getStateVersion == 0L)
+ sm.transitTo(READY)
+ assert(sm.getStateVersion == 1L)
+ sm.transitTo(RUNNING)
+ assert(sm.getStateVersion == 2L)
+ sm.transitTo(COMPLETED)
+ assert(sm.getStateVersion == 3L)
+ }
+
+ it should "not bump the state version on a no-op self-transition" in {
+ val sm = newManager(RUNNING)
+ val before = sm.getStateVersion
+ sm.transitTo(RUNNING) // no-op
+ assert(sm.getStateVersion == before)
+ }
+
+ it should "not bump the state version when a transition is rejected" in {
+ val sm = newManager(UNINITIALIZED)
+ intercept[InvalidTransitionException] {
+ sm.transitTo(RUNNING)
+ }
+ assert(sm.getStateVersion == 0L)
+ }
+
+ it should "return the state and its version as one pair via
getStateWithVersion" in {
+ val sm = newManager(UNINITIALIZED)
+ assert(sm.getStateWithVersion == ((UNINITIALIZED, 0L)))
+ sm.transitTo(READY)
+ assert(sm.getStateWithVersion == ((READY, 1L)))
+ sm.transitTo(RUNNING)
+ assert(sm.getStateWithVersion == ((RUNNING, 2L)))
+ }
}
