apache / apache/texera

Reconfiguration service leaks its engine callback past unsubscribeAll

Open
#7,822 2 comments 0 reactions 1 assignee Claimed by @aglinxinyuan View on GitHub
Dominant language
Scala
Stars
314
Forks
187
Avg merge
1d 21h
Merged PRs (30d)
214

Description

### Describe the bug

`ExecutionReconfigurationService.registerWorkerCompletionCallback` (`amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala`, around line 117) calls `client.registerCallback[UpdateExecutorCompleted](...)` in statement position and **discards the returned `Disposable`**, so the engine callback outlives the service.

It is the only such call site in `org.apache.texera.web.service` that does not wrap the result in `addSubscription`:

| File | `registerCallback` sites | Wrapped |
|---|---|---|
| `ExecutionStatsService.scala` | 5 | yes |
| `ExecutionResultService.scala` | 3 | yes |
| `ExecutionConsoleService.scala` | 2 | yes |
| `ExecutionRuntimeService.scala` | 1 | yes |
| `ExecutionReconfigurationService.scala` | 1 | **no** |

`AmberClient.registerCallback` does return the `Disposable` from its `observable.subscribe(...)` (`AmberClient.scala:116-152`), so there is a handle to keep.

The teardown path is live in production: `WorkflowExecutionService.unsubscribeAll` (`WorkflowExecutionService.scala:185`) calls `executionReconfigurationService.unsubscribeAll()`, which releases every *other* subscription but not this one.

And `client.shutdown()` is not an alternative release — it flips `isActive` and poison-pills the `ClientActor`, but never touches `registeredObservables` (written only in `registerCallback`, never cleared). So the `PublishSubject` chain keeps the subscriber closure, and through it the service, its `ExecutionStateStore` and its `Workflow`, alive for as long as the `AmberClient` is reachable — which `WorkflowExecutionService.client` ensures.

This was found during coverage work (#7692), which deliberately recorded "wrap the discarded `Disposable` in `addSubscription`" as a *surviving* mutation so a fix would not be blocked by a test asserting the leak.

### How to reproduce?

Make the client double's `registerCallback` return a `Disposable` that records disposal, tear the service down with `unsubscribeAll()`, and observe the callback was never disposed — and that a late engine event still reaches the service.

### Version/Branch

main

### Task Type

- [ ] Refactor / Cleanup
- [ ] DevOps / Deployment / CI
- [ ] Testing / QA
- [ ] Documentation
- [ ] Performance
- [x] Other

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.