Flush buffered events before the async tracking client's worker exits - #1692
Open
AmirF194 wants to merge 1 commit into
Open
Flush buffered events before the async tracking client's worker exits#1692AmirF194 wants to merge 1 commit into
AmirF194 wants to merge 1 commit into
Conversation
BasicAsynchronousHamiltonClient had no stop(), and nothing ever put the None sentinel worker() waits for, so a buffered batch sat until the periodic flush timer happened to catch it, or was lost outright if the process exited first. Add stop(), track the worker task, and delegate to it from AsyncHamiltonTracker.stop(). The sentinel branch in worker() had never actually run before this change: it appended the item to the batch before checking whether it was the None sentinel, so the first real stop() call flushed a batch containing None and crashed. Fixed the check order to match the synchronous client, which already gets this right. Fixes apache#1214
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
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.
Fixes #1214.
Changes
BasicAsynchronousHamiltonClient(ui/sdk/src/hamilton_sdk/api/clients.py) had nostop()and registered no atexit hook, unlike its synchronous sibling.
worker()'s only exit pathwaits for a
Nonesentinel ondata_queue, and nothing ever put one there for the asyncclient, so a buffered batch sat until the periodic flush timer happened to catch it, or was
lost outright if the process (or the event loop that owned the worker task) exited first.
async def stop()onBasicAsynchronousHamiltonClient: sends the sentinel and awaitsthe worker task so it flushes and exits. This is not wired to
atexit, deliberately: theworker is a task on the caller's event loop, and by the time atexit callbacks run at
interpreter shutdown that loop may already be closed, so callers must await
stop()themselves before the loop stops running (mirroring the sync client isn't possible here for
that reason; happy to add an
atexit-scheduled best-effort variant too if maintainers wantone, but that has its own shutdown-ordering edge cases worth discussing first).
async def stop()onAsyncHamiltonTracker(adapters.py), delegatingto the client.
worker()'s sentinel handling that this exercises for the first time: itappended the received item to
batchbefore checking whether it was theNonesentinel, sothe first real
stop()call flushed a batch containingNoneand raised. Reordered to checkfirst, matching the synchronous client's worker.
How I tested this
Ran in a clean
python:3.13-slimandpython:3.10-slimDocker container (the SDK's CI matrixruns 3.10-3.14; I could not run the 3.11/3.12/3.14 legs locally but the change has no
version-specific code paths):
tests/test_clients.pyandtests/test_adapters.py: fail withAttributeErroron unmodifiedmain(confirmed viagit stash), pass on this branch.test_async_client_stop_flushes_buffered_item_and_exits_workerenqueues an item, callsstop()before the flush interval elapses, and asserts the mocked backend request actuallyfired and the queue drained; it caught the sentinel-ordering bug above on first run.
pytest tests/(minus theray/pyspark-gated tests, which arepytest.importorskipguarded and unrelated to this change): 140 passed, 5 skipped, no failures.
ruff checkandruff format --diffon the four changed files: clean. This repo's CI doesnot run a lint/static-checks job for
ui/sdk-only changes (checkedgh pr checkson arecently merged
ui/sdk-only PR, fix(ui/sdk): fix failing CI on master due to formatting issues #1626), so this was a courtesy check, not a required gate.Not verified: behavior under a real ASGI/FastAPI shutdown sequence (e.g. the app in
examples/async/fastapi_example.py, which doesn't call the newstop()either); the fix isscoped to the tracker itself, not to updating that example.