villebro commented on code in PR #43678:
URL: https://github.com/apache/superset/pull/43678#discussion_r3886844975
##########
superset/coordination/types.py:
##########
@@ -38,22 +38,30 @@ class SignalListener:
"""Handle for a background listener started by
:meth:`~superset.coordination.base.CoordinationService.listen_for_signal`.
- Wraps the daemon thread and its stop flag. :meth:`stop` sets the flag and
joins;
- the listener's bounded blocking read means it notices the flag within one
short
- tick.
+ Wraps the daemon thread and its stop flag. :meth:`stop` sets the flag and,
+ when a ``wake`` is provided, nudges the backend stream so a listener
parked in
+ a blocking read returns at once rather than waiting out its block interval
—
+ keeping task teardown from paying the full read timeout.
"""
def __init__(
self,
thread: threading.Thread,
stop_event: threading.Event,
+ wake: "Callable[[], None] | None" = None,
) -> None:
self._thread = thread
self._stop_event = stop_event
+ self._wake = wake
def stop(self) -> None:
"""Signal the listener to stop and wait briefly for the thread to
finish."""
self._stop_event.set()
+ # Wake a listener blocked in a backend read so it observes the stop
flag
+ # immediately (the no-backend loop already wakes on the event).
Best-effort:
+ # a failed nudge just falls back to the bounded join below.
+ if self._wake is not None:
+ self._wake()
Review Comment:
Fixed in 551d62c13d. The wake nudge now runs on a daemon thread, so a
synchronous Redis `xadd`/`expire` on a degraded backend (no socket timeout) can
no longer block `stop()` past its bounded 2s join + daemon-reap fallback —
teardown stays bounded regardless of backend health. Added
`test_signal_listener_stop_bounded_when_wake_hangs` covering exactly this (wake
that never returns).
##########
superset/models/tasks.py:
##########
@@ -187,6 +187,22 @@ def update_properties(self, updates: TaskProperties) ->
None:
current.update(updates) # Merge updates
self.properties = serialize_properties(current)
+ def update_private_properties(self, updates: dict[str, Any]) -> None:
+ """
+ Merge keys into the ``private`` properties bucket (internal runtime
state).
+
+ ``private`` holds framework plumbing (job/cancel handles) that is never
+ surfaced to user-facing API payloads. Merges rather than replaces, so a
+ later write (e.g. the engine cancel handle) preserves an earlier one
(e.g.
+ the Celery job id).
+
+ :param updates: private keys to set/merge
+ """
+ current = cast(TaskProperties, dict(self.properties_dict))
+ private: dict[str, Any] = {**(current.get("private") or {}), **updates}
Review Comment:
Addressed in 551d62c13d. The flat `update_private_properties` this line
referenced was replaced by a recursive per-namespace merge, now a shared
`merge_private_subtree()` helper that treats a non-dict subtree (or namespace)
as empty instead of unpacking it, so a malformed value cannot raise
`TypeError`. Note `private` is framework-managed (never user-supplied), so this
is defensive hardening. Covered by
`test_private_merge_tolerates_malformed_existing_value`.
##########
superset/tasks/context.py:
##########
@@ -393,8 +393,14 @@ def set_cancellation(self, database_id: int,
cancel_query_id: str) -> None:
write. The orphan reaper reads it to cancel the query out-of-band when
this worker dies; the live abort path uses its in-memory closure
instead.
"""
- self._properties_cache["cancel_database_id"] = database_id
- self._properties_cache["cancel_query_id"] = cancel_query_id
+ self._properties_cache["private"] = cast(
+ "Any",
+ {
+ **(self._properties_cache.get("private") or {}),
+ "cancel_database_id": database_id,
+ "cancel_query_id": cancel_query_id,
+ },
+ )
Review Comment:
Fixed in 551d62c13d — `set_cancellation` now routes through the same
hardened `merge_private_subtree()` helper as the model, so a non-dict existing
`private`/`task` value is treated as empty rather than raising `TypeError`
during cancellation setup.
##########
tests/unit_tests/coordination/test_service.py:
##########
@@ -321,3 +321,39 @@ def test_signal_listener_stop_signals_and_joins(mocker:
MockerFixture) -> None:
assert stop_event.is_set()
thread.join.assert_called_once_with(timeout=2.0)
+
+
+def test_signal_listener_stop_wakes_before_join(mocker: MockerFixture) -> None:
+ """stop() nudges the wake (so a blocked read returns) before joining."""
+ thread = mocker.MagicMock(name="thread")
+ thread.is_alive.side_effect = [True, False]
+ stop_event = threading.Event()
+ wake = mocker.MagicMock(name="wake")
+
+ SignalListener(thread, stop_event, wake=wake).stop()
+
+ assert stop_event.is_set()
+ wake.assert_called_once_with()
+ thread.join.assert_called_once_with(timeout=2.0)
+
+
+def test_listen_stop_wakes_blocked_backend_read(
+ app_context: None, mocker: MockerFixture
+) -> None:
+ """With a backend, stop() writes a wake entry so a listener parked in a
+ blocking XREAD returns at once instead of waiting out the block
interval."""
+ backend = mocker.MagicMock(name="backend")
+ backend.stream_last_id.return_value = "0-0"
+ backend.xread.return_value = [] # no entries → the loop keeps reading
+ mocker.patch.object(CoordinationService, "get_backend",
return_value=backend)
Review Comment:
Good call — strengthened in 551d62c13d.
`test_listen_stop_wakes_blocked_backend_read` now parks the listener in a real
blocking `xread` (gated on a `threading.Event` released by the wake `xadd`) and
asserts the thread actually terminates after `stop()`, so a broken wake would
now fail the test. Also added
`test_signal_listener_stop_bounded_when_wake_hangs`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
