This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6011-b4d18656589523e3132a41607557da0d8bc44e18 in repository https://gitbox.apache.org/repos/asf/texera.git
commit f5217504c3cfd8ac1d56f6617d7849895af17f59 Author: Yicong Huang <[email protected]> AuthorDate: Sun Jul 26 13:18:17 2026 -0700 fix(amber): order worker state by version, not timestamp (#6011) ### What changes were proposed in this PR? A fast **source** operator stayed **orange (RUNNING)** in the editor after the run finished (issue #6010). **Why it's a bug.** The controller reconstructs each worker's state from several unordered channels (source `RUNNING` via the `startWorker` response, non-source `RUNNING` via `workerStateUpdated`, `COMPLETED`/`PAUSED` via `queryStatistics`/`pauseWorker` responses) and reconciled them with **last-`System.nanoTime()`-wins** in `WorkerExecution`. Worker state, however, is single-writer and strictly ordered causally. For a tiny source the run finishes almost instantly, so the `startWorker` response — carrying the stale `RUNNING` it sampled at launch — can reach the controller *after* `COMPLETED` was recorded; its later receipt timestamp wins and clobbers `COMPLETED`. Results render fine (separate path), so only the border is stuck. This PR orders worker state **causally** instead of by wall clock: | Before | After | | --- | --- | | `WorkerExecution.update(nanoTime, state)` — last-write-wins by receipt time | `updateState(version, state)` — newest **logical version** wins | | stale late RUNNING clobbers COMPLETED | RUNNING (v2) < COMPLETED (v3) ⇒ COMPLETED kept | | — | terminal states (COMPLETED/TERMINATED) are absorbing | - `WorkerStateManager` bumps a monotonic `stateVersion` on every `transitTo` (its state-machine logical clock). - The version rides on every state report: `WorkerStateResponse`, `WorkerStateUpdatedRequest`, `WorkerMetrics` (3 new proto fields). - The controller applies a state only when its version is newer. Stats stay timestamp-ordered (monotonic snapshots can share a state version). - A single per-worker counter ⇒ no cross-process clock-sync assumptions (why not worker timestamps). - **pyamber parity:** the Python worker also reports state (start/pause/resume + query-statistics). Its `StateManager` now bumps the same monotonic version and the four handlers include it; otherwise version-0 reports would be dropped after the first and reconfiguration (RUNNING→PAUSED→RUNNING) would hang. ### Any related issues, documentation, discussions? Closes #6010. The timestamp-based `update` was introduced in #3557. ### How was this PR tested? JDK 17. Scala unit + Scala/Python integration + Python unit: ``` sbt "WorkflowExecutionService/testOnly \ *WorkerExecutionSpec *WorkerStateManagerSpec *OperatorExecutionSpec \ *RegionExecutionCoordinatorSpec *WorkflowExecutionCoordinatorSpec \ *RegionExecutionSpec *WorkflowExecutionSpec" # Scala-Python reconfiguration end-to-end (spawns Python UDF workers): sbt "WorkflowExecutionService/testOnly *ReconfigurationIntegrationSpec" # 3 passed # Python unit: cd amber && pytest src/test/python/core/architecture/managers/test_state_manager.py # 9 passed ``` - `WorkerExecutionSpec`: version ordering (positive + stale/equal-version negatives), terminal-state absorption, `forceTerminate`, independent stats-vs-state ordering, and a named regression for #6010 (`COMPLETED` survives a late `RUNNING`). Verified the regression goes **red** when `updateState` is reverted to last-write-wins. - `WorkerStateManagerSpec` / `test_state_manager.py`: version starts at 0, bumps per successful transition, no bump on no-op self-transition or rejected transition. - `ReconfigurationIntegrationSpec` reproduced the failure (timeout) before the pyamber fix and passes after. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Opus 4.8 (1M context), via Claude Code --------- 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 ++++++ .../coordinator/execution/ExecutionUtils.scala | 12 ++- .../coordinator/promisehandlers/PauseHandler.scala | 7 +- .../QueryWorkerStatisticsHandler.scala | 4 +- .../promisehandlers/ResumeHandler.scala | 2 +- .../WorkerStateUpdatedHandler.scala | 2 +- .../deploysemantics/layer/WorkerExecution.scala | 70 ++++++++------ .../scheduling/RegionExecutionManager.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 +++++- .../coordinator/execution/ExecutionUtilsSpec.scala | 12 +-- .../execution/OperatorExecutionSpec.scala | 13 +-- .../execution/WorkflowExecutionSpec.scala | 10 +- .../layer/WorkerExecutionSpec.scala | 103 ++++++++++++++++----- .../RegionExecutionManagerTestSupport.scala | 3 +- .../statetransition/WorkerStateManagerSpec.scala | 41 ++++++++ 29 files changed, 347 insertions(+), 98 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 21060b6a24..8a6403a97e 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 064a27b572..c367bd9dd0 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/coordinator/execution/ExecutionUtils.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala index e89da69d43..666aeece42 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/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/coordinator/promisehandlers/PauseHandler.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/PauseHandler.scala index 5a1b8a5cd8..98b33dc783 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/PauseHandler.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/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/coordinator/promisehandlers/QueryWorkerStatisticsHandler.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/QueryWorkerStatisticsHandler.scala index 58cf25b653..82cb7cc799 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/QueryWorkerStatisticsHandler.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/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/coordinator/promisehandlers/ResumeHandler.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/ResumeHandler.scala index 7843659549..0524c99b41 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/ResumeHandler.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/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/coordinator/promisehandlers/WorkerStateUpdatedHandler.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/WorkerStateUpdatedHandler.scala index 225eeb9b19..111cb24771 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/promisehandlers/WorkerStateUpdatedHandler.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/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 59fe98c8f3..e3fce6aee8 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.coordinator.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/RegionExecutionManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala index 586d55ca25..93d085b611 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.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 @@ -214,7 +213,7 @@ class RegionExecutionManager( 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 @@ -583,12 +582,14 @@ class RegionExecutionManager( 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 feec522a9d..618fa73427 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 @@ -221,8 +221,9 @@ class DataProcessor( READY, RUNNING, () => { + val (state, stateVersion) = stateManager.getStateWithVersion asyncRPCClient.coordinatorInterface.workerStateUpdated( - WorkerStateUpdatedRequest(stateManager.getCurrentState), + WorkerStateUpdatedRequest(state, stateVersion), asyncRPCClient.mkContext(COORDINATOR) ) } 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 a9e3f4ed46..353431100c 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 2bbefe8c43..78ec517635 100644 --- a/amber/src/test/python/core/runnables/test_main_loop.py +++ b/amber/src/test/python/core/runnables/test_main_loop.py @@ -743,6 +743,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, @@ -769,6 +771,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( @@ -1176,13 +1181,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 + ) ), ) ), @@ -1197,13 +1210,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/coordinator/execution/ExecutionUtilsSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtilsSpec.scala index 1a66e3fdbb..237936fa06 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtilsSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/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/coordinator/execution/OperatorExecutionSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/OperatorExecutionSpec.scala index de7017abb5..fa16873fbc 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/OperatorExecutionSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/OperatorExecutionSpec.scala @@ -42,10 +42,10 @@ class OperatorExecutionSpec extends AnyFlatSpec { PortTupleMetricsMapping(PortIdentity(portIdx), TupleMetrics(count, size)) /** - * Push `(state, stats)` onto an existing `WorkerExecution`. Production - * code applies updates only if the timestamp is newer than the - * previously-recorded one; we use a monotonically increasing nano-clock - * surrogate so each call wins. + * Push `(state, stats)` onto an existing `WorkerExecution`. Production code + * orders state by a monotonic logical version and stats by timestamp, applying + * an update only when it is newer; we use a single monotonically increasing + * surrogate for both so each call wins. */ private var clock: Long = 0L private def applyUpdate( @@ -54,14 +54,15 @@ class OperatorExecutionSpec extends AnyFlatSpec { stats: WorkerStatistics ): Unit = { clock += 1 - worker.update(clock, state, stats) + worker.updateState(clock, state) + worker.updateStats(clock, stats) } private def setState( worker: org.apache.texera.amber.engine.architecture.deploysemantics.layer.WorkerExecution, state: WorkerState ): Unit = { clock += 1 - worker.update(clock, state) + worker.updateState(clock, state) } // --------------------------------------------------------------------------- diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/WorkflowExecutionSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/WorkflowExecutionSpec.scala index d61ac8799b..6a6cd675e0 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/WorkflowExecutionSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/WorkflowExecutionSpec.scala @@ -171,7 +171,7 @@ class WorkflowExecutionSpec extends AnyFlatSpec { val operatorExecution = regionExecution.initOperatorExecution(physicalOpId("a")) val workerExecution = operatorExecution.initWorkerExecution(ActorVirtualIdentity("w0")) - workerExecution.update(1L, WorkerState.PAUSED) + workerExecution.updateState(1L, WorkerState.PAUSED) assert(we.getState == WorkflowAggregatedState.PAUSED) assert(!we.isCompleted) @@ -182,8 +182,12 @@ class WorkflowExecutionSpec extends AnyFlatSpec { val regionExecution = we.initRegionExecution(regionWithPort(0, "a")) val operatorExecution = regionExecution.initOperatorExecution(physicalOpId("a")) - operatorExecution.initWorkerExecution(ActorVirtualIdentity("w0")).update(1L, WorkerState.PAUSED) - operatorExecution.initWorkerExecution(ActorVirtualIdentity("w1")).update(1L, WorkerState.READY) + operatorExecution + .initWorkerExecution(ActorVirtualIdentity("w0")) + .updateState(1L, WorkerState.PAUSED) + operatorExecution + .initWorkerExecution(ActorVirtualIdentity("w1")) + .updateState(1L, WorkerState.READY) assert(we.getState == WorkflowAggregatedState.UNKNOWN) assert(!we.isCompleted) 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/RegionExecutionManagerTestSupport.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala index 8a012c0c20..5fc6375ddd 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala @@ -140,7 +140,8 @@ object RegionExecutionManagerTestSupport { 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))) + } }
