Skip to content

fix(amber): reject duplicate worker initialization #8083

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Yicong-Huang merged 1 commit into from
Aug 31, 2026
Merged

fix(amber): reject duplicate worker initialization #8083

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...org/apache/texera/amber/engine/architecture/coordinator/execution/OperatorExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ case class OperatorExecution() {
*/
def initWorkerExecution(workerId: ActorVirtualIdentity): WorkerExecution = {
assert(
!workerExecutions.contains(workerId),
!workerExecutions.containsKey(workerId),
Comment thread
carloea2 marked this conversation as resolved.
s"WorkerExecution already exists for workerId: $workerId"
)
workerExecutions.put(workerId, WorkerExecution())
Expand Down
23 changes: 2 additions & 21 deletions ...apache/texera/amber/engine/architecture/coordinator/execution/OperatorExecutionSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,34 +78,15 @@ class OperatorExecutionSpec extends AnyFlatSpec {
assert(opExec.getWorkerExecution(w) eq workerExec)
}

// The class docstring claims `initWorkerExecution` throws
// `AssertionError` on a duplicate worker id, but the implementation's
// `workerExecutions.contains(workerId)` call resolves to Java
// `ConcurrentHashMap.contains(Object)`, which checks VALUES rather than
// KEYS — so the assertion never fires and the second call silently
// overwrites the prior WorkerExecution. We pin the CURRENT (broken)
// behavior here so a future fix is noticed in CI, and document the
// intended contract with `pendingUntilFixed` so the failure surfaces
// the day the implementation is corrected.

it should
"currently overwrite the previous WorkerExecution on a second init for the same id " +
"(characterization of the contains-by-value bug)" in {
"reject a second init for the same id without replacing the existing execution" in {
val opExec = OperatorExecution()
val w = workerId("w-1")
val firstExec = opExec.initWorkerExecution(w)
val secondExec = opExec.initWorkerExecution(w)
assert(firstExec ne secondExec, "current impl replaces the prior WorkerExecution instance")
assert(opExec.getWorkerExecution(w) eq secondExec)
}

it should "(desired) throw AssertionError when initWorkerExecution is called twice for the same id" in pendingUntilFixed {
val opExec = OperatorExecution()
val w = workerId("w-1")
opExec.initWorkerExecution(w)
assertThrows[AssertionError] {
opExec.initWorkerExecution(w)
}
assert(opExec.getWorkerExecution(w) eq firstExec)
}

"OperatorExecution.getWorkerIds" should "be empty on a freshly constructed operator" in {
Expand Down
Loading