Refactoring of amber engine - #2208
Merged
Merged
Conversation
shengquan-ni
force-pushed
the
shengquan-refactor-1
branch
from
November 6, 2023 19:46
0cf7408 to
059f143
Compare
This is a follow-up PR after #2208, which focused on refactoring of the controller and workflow execution states. ## Problem of Serializing Controller State The controller holds the workflow, scheduler, and execution state, however, many of them are not serializable because they have deep dependencies with third-party libs such as jgrapht, akka. ## Changes in This PR Separated the execution state from the workflow in another class called `ExecutionState`, which is completely serializable. It includes the execution runtime state of all the workers, edges, and regions. ### Controller Refactoring Following #2208, the controller also creates its core state (ControllerProcessor) and all serializable components are put into this core state.
shengquan-ni
changed the title
Yicong-Huang
approved these changes
Nov 15, 2023
Xiao-zhen-Liu
pushed a commit
that referenced
this pull request
Jan 16, 2024
## Purpose of Refactoring ### Supporting Checkpoints: 1. **Channel Clarity and Message Handling**: Previously, the worker's handling of messages in channels was ambiguous, using various message classes to differentiate channels. This led to code duplication and complexity for implementing checkpoints. 2. **State Management**: The components within a worker are well-organized. This disorganization would complicate future implementations of state serialization and deserialization. ### Architectural Improvements: 1. **Worker Architecture Cleanup**: The existing worker architecture has evolved in an ad-hoc manner, leading to a lack of clarity. An example is the Network Communicate Actor, which needs a redesign for better maintainability. ### Current Worker Architecture See the current architecture diagram here:  ## Key Refactorings in This PR ### Supporting Checkpoints: 1. **Message Class Unification**: Introduced a unified message type `WorkflowFIFOMessage` with a `ChannelID` to clearly identify communication channels, reducing code duplication. 2. **Actor Component Extraction**: Separated actor-related components (like Actor context, timers, flow control and congestion control) from DP, integrating them into the main actor for better alignment with the Akka actor system and avoiding serialization. 3. **Core State Introduction**: Established a core state within the actor(i.e. DataProcessor, AmberProcessor) for serialization during checkpoints, ensuring non-core states are excluded from serialization processes. ### Architectural Improvements: 1. **Network Communication Actor Removal**: Transferred all messaging responsibilities to the main actor, including sending messages and receiving acks. Flow and congestion controls have been refactored on a per-channel basis. 2. **Renaming for Clarity**: Changed `NetworkInputPort` and `NetworkOutputPort` to `NetworkInputGateway` and `NetworkOutputGateway`, reflecting their roles in managing multiple channels. 3. **Enhanced Testing**: Added additional test cases for Java/Python workers. ### Temporary Changes: - The controller has been temporarily removed and end-to-end tests are currently disabled. These will be reinstated in PR #2209. ### Post-Refactoring Worker Architecture View the updated architecture diagram here:  ### Flow control After this PR, the credit of flow control is not sent back with the acks. We poll from the downstream operator whenever the credit is not enough. This will be fixed later with a new flow control design. --------- Co-authored-by: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>
yangzhang75
pushed a commit
to yangzhang75/texera
that referenced
this pull request
Jun 22, 2026
## Purpose of Refactoring ### Supporting Checkpoints: 1. **Channel Clarity and Message Handling**: Previously, the worker's handling of messages in channels was ambiguous, using various message classes to differentiate channels. This led to code duplication and complexity for implementing checkpoints. 2. **State Management**: The components within a worker are well-organized. This disorganization would complicate future implementations of state serialization and deserialization. ### Architectural Improvements: 1. **Worker Architecture Cleanup**: The existing worker architecture has evolved in an ad-hoc manner, leading to a lack of clarity. An example is the Network Communicate Actor, which needs a redesign for better maintainability. ### Current Worker Architecture See the current architecture diagram here:  ## Key Refactorings in This PR ### Supporting Checkpoints: 1. **Message Class Unification**: Introduced a unified message type `WorkflowFIFOMessage` with a `ChannelID` to clearly identify communication channels, reducing code duplication. 2. **Actor Component Extraction**: Separated actor-related components (like Actor context, timers, flow control and congestion control) from DP, integrating them into the main actor for better alignment with the Akka actor system and avoiding serialization. 3. **Core State Introduction**: Established a core state within the actor(i.e. DataProcessor, AmberProcessor) for serialization during checkpoints, ensuring non-core states are excluded from serialization processes. ### Architectural Improvements: 1. **Network Communication Actor Removal**: Transferred all messaging responsibilities to the main actor, including sending messages and receiving acks. Flow and congestion controls have been refactored on a per-channel basis. 2. **Renaming for Clarity**: Changed `NetworkInputPort` and `NetworkOutputPort` to `NetworkInputGateway` and `NetworkOutputGateway`, reflecting their roles in managing multiple channels. 3. **Enhanced Testing**: Added additional test cases for Java/Python workers. ### Temporary Changes: - The controller has been temporarily removed and end-to-end tests are currently disabled. These will be reinstated in PR apache#2209. ### Post-Refactoring Worker Architecture View the updated architecture diagram here:  ### Flow control After this PR, the credit of flow control is not sent back with the acks. We poll from the downstream operator whenever the credit is not enough. This will be fixed later with a new flow control design. --------- Co-authored-by: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>
This was referenced Jul 29, 2026
renovate-bot
pushed a commit
to renovate-bot/apache-_-texera
that referenced
this pull request
Jul 30, 2026
### What changes were proposed in this PR? Deletes `OrderingEnforcer`, a generic helper in the amber engine with no production caller. Pure deletion, no behaviour change: **−203 lines**. | Removed | Lines | Why it is dead | | --- | ---: | --- | | `architecture/messaginglayer/OrderingEnforcer.scala` | 53 | last caller removed by apache#2208 | | `architecture/messaginglayer/OrderingEnforcerSpec.scala` | 150 | covers only the above | `OrderingEnforcer` was the extracted, generic form of the FIFO / exactly-once reorder buffer. apache#2208 ("Refactoring of amber engine", 2023-11-15) inlined that logic into `AmberFIFOChannel`, which still carries the identical header comment `/* The abstracted FIFO/exactly-once logic */` and re-declares the same members: ``` OrderingEnforcer[T] (deleted) AmberFIFOChannel (live, unchanged) ofoMap: LongMap[T] ofoMap: HashMap[Long, WorkflowFIFOMessage] current: Long current: Long isDuplicated / isAhead isDuplicated / isAhead stash / enforceFIFO stash / enforceFIFO ``` The generic copy has been stranded for two and a half years. > Reviewer note: there is a separate, **live** `OrderEnforcer` trait (no `-ing`) in `architecture/logreplay/`, used by `InputGateway`/`NetworkInputGateway`. It is unrelated and untouched — only the `messaginglayer` `OrderingEnforcer` is removed here. It acquired unit tests during the 2026 coverage work (apache#4721), which is why it currently looks live despite having been unreachable since 2023. **Scope change after review:** this PR originally also removed `Utils.retry`. Per @Yicong-Huang's review it has been restored and is untouched here — `Utils.scala` and `UtilsSpec.scala` are byte-identical to `main`. ### Any related issues, documentation, discussions? Closes apache#7027 ### How was this PR tested? Existing tests only — this PR adds none, since it removes code and the spec that covered it. Locally, from the repo root with Java 17: - `sbt "scalafixAll --check"` — clean. - `sbt scalafmtCheckAll` — clean. - `sbt "WorkflowExecutionService/testOnly *UtilsSpec"` — 74 tests, all pass (`Utils.retry` and its three tests are back and green). Verification that nothing references the removed class, re-runnable by a reviewer: ``` git grep -nw OrderingEnforcer # only the two deleted files ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose of Refactoring
Supporting Checkpoints:
Architectural Improvements:
Current Worker Architecture
See the current architecture diagram here:
Key Refactorings in This PR
Supporting Checkpoints:
WorkflowFIFOMessagewith aChannelIDto clearly identify communication channels, reducing code duplication.Architectural Improvements:
NetworkInputPortandNetworkOutputPorttoNetworkInputGatewayandNetworkOutputGateway, reflecting their roles in managing multiple channels.Temporary Changes:
Post-Refactoring Worker Architecture
View the updated architecture diagram here:
Flow control
After this PR, the credit of flow control is not sent back with the acks. We poll from the downstream operator whenever the credit is not enough. This will be fixed later with a new flow control design.