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-7823-51d7564309bf74ad75b4f28844ba2516e6bad9a6 in repository https://gitbox.apache.org/repos/asf/texera.git
commit f642c183dc5461f31b84f6bd37dc12e917ca3fdd Author: Xinyuan Lin <[email protected]> AuthorDate: Mon Sep 21 01:26:58 2026 +0000 fix(amber): release the reconfiguration service's engine callback on teardown (#7823) ### What changes were proposed in this PR? `ExecutionReconfigurationService.registerWorkerCompletionCallback` discarded the `Disposable` returned by `client.registerCallback`, so its engine callback outlived the service. It now hands it to `addSubscription(...)`, matching `registerCompletionDiffHandler` directly below it. It was the only `registerCallback` site in `org.apache.texera.web.service` not doing this: | File | sites | wrapped | |---|---|---| | `ExecutionStatsService.scala` | 5 | yes | | `ExecutionResultService.scala` | 3 | yes | | `ExecutionConsoleService.scala` | 2 | yes | | `ExecutionRuntimeService.scala` | 1 | yes | | `ExecutionReconfigurationService.scala` | 1 | **no** | ### Why this actually leaks, rather than being tidied up by something else The teardown path is live: `WorkflowExecutionService.unsubscribeAll` (`WorkflowExecutionService.scala:185`) calls `executionReconfigurationService.unsubscribeAll()`, which released 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`, which is written only in `registerCallback` and never cleared. The `PublishSubject` chain therefore keeps the subscriber closure, and through it the service, its `ExecutionStateStore` and its `Workflow`, for as long as the `AmberClient` is reachable — which `WorkflowExecutionService.client` guarantees. ### The fix is pinned New test: `"the worker completion callback" should "release the engine subscription once the service is unsubscribed"`. Verified with the production file reverted and restored: | | production reverted | with fix | |---|---|---| | `ExecutionReconfigurationServiceSpec` | **13 passed, 1 failed** | **14 passed** | The before-state failure is the right one — `ArrayBuffer() did not contain element class …UpdateExecutorCompleted`, i.e. the disposable was never disposed because it was never registered. Two additions to the existing `TestAmberClient` double were needed, and the reasons are worth stating: an explicit `disposedCallbacks` record, because disposal-removal from the `callbacks` map alone is indistinguishable from "never registered" or "cleared by `reset`"; and a `fireIfRegistered` helper, because the strict `fire` calls `fail(...)` when nothing is registered and so cannot express "a late event is inert". ### Two comments were rewritten, not just tests added Both would otherwise have contradicted the code: 1. `TestAmberClient.registerCallback`'s scaladoc said production "currently DISCARDS it" and that "no test below fires an engine event after `unsubscribeAll`, so the suite neither depends on the leak nor breaks when it is fixed." Both clauses are now false. 2. The routing comment in `"stop announcing completions once the service is unsubscribed"` called the engine callback an "unrelated, currently broken, seam". The routing decision — drive that test through `onWorkerReconfigured` rather than the engine event — is still right and unchanged, but re-justified: the two are separate subscriptions, so routing through the engine event would leave an empty batch explainable by either one being released. That test's assertion is untouched, so it stays agnostic exactly as #7692 intended. ### Verification - `ExecutionReconfigurationServiceSpec` 14/14. - Dependents and neighbours: `ExecutionRuntimeServiceSpec`, `WorkflowExecutionServiceSpec`, `ExecutionStatsServiceSpec`, `ExecutionConsoleServiceSpec`, `WorkflowWebsocketResourceSpec`, `TexeraWebSocketEventSpec` — **59/59 across 6 suites, 0 aborted**. - `scalafmtCheck`, `Test/scalafmtCheck`, `scalafixAll --check` all pass. (The one scalafix warning is a pre-existing `// scalafix:ok` in `OutputManagerSpec`, unrelated.) ### One thing deliberately left alone The suite still carries an assertion recorded as *observed, not endorsed*: an N-worker operator fires N `ModifyLogicCompletedEvent`s, which contradicts `registerCompletionDiffHandler`'s own comment claiming the frontend is notified once **all** workers finish. That is a separate defect with its own decision to make, so this PR does not touch it. ### Any related issues, documentation, discussions? Closes #7822 ### How was this PR tested? ``` STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly org.apache.texera.web.service.ExecutionReconfigurationServiceSpec" ``` ``` [info] Total number of tests run: 14 [info] Tests: succeeded 14, failed 0, canceled 0, ignored 0, pending 0 ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../service/ExecutionReconfigurationService.scala | 8 +-- .../ExecutionReconfigurationServiceSpec.scala | 62 +++++++++++++++++----- 2 files changed, 55 insertions(+), 15 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala index 31611413fd..2147231abb 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala @@ -114,9 +114,11 @@ class ExecutionReconfigurationService( // events into the reconfiguration store so the diff handler above can fire // ModifyLogicCompletedEvent for the frontend. protected def registerWorkerCompletionCallback(): Unit = { - client.registerCallback[UpdateExecutorCompleted]((evt: UpdateExecutorCompleted) => { - onWorkerReconfigured(evt.id) - }) + addSubscription( + client.registerCallback[UpdateExecutorCompleted]((evt: UpdateExecutorCompleted) => { + onWorkerReconfigured(evt.id) + }) + ) } // Exposed (instead of inlined in the callback) so tests can drive the diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala index 45275ef308..dfc3569385 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala @@ -259,6 +259,12 @@ class ExecutionReconfigurationServiceSpec /** (method name, request) of every call made through `coordinatorInterface`. */ val coordinatorCalls: mutable.ArrayBuffer[(String, Any)] = mutable.ArrayBuffer.empty + /** Event class of every Disposable handed out by `registerCallback` that was later disposed. + * Recorded separately from `callbacks` because disposal removes the entry, which on its own + * is indistinguishable from a callback that was never registered (or one dropped by `reset`). + */ + val disposedCallbacks: mutable.ArrayBuffer[Class[_]] = mutable.ArrayBuffer.empty + override val coordinatorInterface: CoordinatorServiceFs2Grpc[Future, Unit] = Proxy .newProxyInstance( @@ -281,19 +287,18 @@ class ExecutionReconfigurationServiceSpec .asInstanceOf[CoordinatorServiceFs2Grpc[Future, Unit]] /** - * Honours the Disposable it hands out. Production - * (`ExecutionReconfigurationService.registerWorkerCompletionCallback`) currently DISCARDS - * it, so `unsubscribeAll()` does not release the engine callback -- the only - * `registerCallback` call site in this package that is not wrapped in `addSubscription` - * (contrast ExecutionStatsService, ExecutionConsoleService, ExecutionResultService and - * ExecutionRuntimeService). A fake that ignored disposal would silently absolve that line. - * No test below fires an engine event after `unsubscribeAll`, so the suite neither depends - * on the leak nor breaks when it is fixed. + * Honours the Disposable it hands out, the way the real client's rx subscription does: + * disposing it detaches the callback, so nothing is delivered to it afterwards. A fake that + * ignored disposal would make every `registerCallback` call site in this package look + * correctly scoped, including one that dropped the Disposable on the floor. */ override def registerCallback[T](callback: T => Unit)(implicit ct: ClassTag[T]): Disposable = { val clazz = ct.runtimeClass callbacks(clazz) = callback.asInstanceOf[Any => Unit] - Disposable.fromAction(() => callbacks.remove(clazz)) + Disposable.fromAction(() => { + callbacks.remove(clazz) + disposedCallbacks += clazz + }) } /** Delivers an engine event the way the client's observable would. */ @@ -303,10 +308,17 @@ class ExecutionReconfigurationServiceSpec fail(s"no callback is registered for ${ct.runtimeClass.getSimpleName}") )(event) + /** Same delivery as `fire`, but tolerates there being no live callback instead of failing -- + * the only way to check that an event arriving after teardown really is inert. + */ + def fireIfRegistered[T <: AnyRef](event: T)(implicit ct: ClassTag[T]): Unit = + callbacks.get(ct.runtimeClass).foreach(deliver => deliver(event)) + /** Per-test isolation for the one client the suite shares. */ def reset(): Unit = { callbacks.clear() coordinatorCalls.clear() + disposedCallbacks.clear() } def dispose(): Unit = super.shutdown() @@ -684,9 +696,9 @@ class ExecutionReconfigurationServiceSpec f.service.unsubscribeAll() // Driven through the service's own entry point rather than the engine event on purpose. // What this test is about is the diff handler's disposable reaching the - // SubscriptionManager; the engine callback has an independent lifetime (see - // TestAmberClient.registerCallback), and routing through it would make this test's - // verdict depend on that unrelated, currently broken, seam. + // SubscriptionManager; the engine callback is a separate subscription with its own + // disposable (covered by the test below), and routing through it would leave the empty + // batch explainable by either one being released. f.service.onWorkerReconfigured(workerB) } @@ -700,4 +712,30 @@ class ExecutionReconfigurationServiceSpec Set(workerA, workerB) } } + + it should "release the engine subscription once the service is unsubscribed" in { + val workerA = mkWorker("opA", 0) + val workerB = mkWorker("opB", 0) + withLiveService(physicalOps = Set(mkPhysicalOp("opA"), mkPhysicalOp("opB"))) { f => + // A live registration first, so a disposed one below cannot be confused with one that was + // never made. + f.client.fire(UpdateExecutorCompleted(workerA)) + f.stateStore.reconfigurationStore.getState.completedReconfigurations shouldBe Set(workerA) + + f.service.unsubscribeAll() + + // The Disposable `registerCallback` handed back has to reach the SubscriptionManager, the + // same way the diff handler's does; this is the only thing that can release it. The + // client's own `shutdown` is not an alternative: it flips `isActive` and poison-pills the + // ClientActor but never touches `registeredObservables`, so the subject keeps the + // subscriber -- and with it this service, its state store and its Workflow -- attached for + // as long as the AmberClient is reachable, which WorkflowExecutionService keeps it. + f.client.disposedCallbacks should contain(classOf[UpdateExecutorCompleted]) + + // ...and the detached callback is inert: a late engine event does not write into the store + // of a finished execution. + f.client.fireIfRegistered(UpdateExecutorCompleted(workerB)) + f.stateStore.reconfigurationStore.getState.completedReconfigurations shouldBe Set(workerA) + } + } }
