Fix two crashes in the aiokafka threaded producer

[wbarnha]

Both bugs were surfaced by the type checker in #758 and left marked XXX there, because fixing them changes runtime behaviour and that PR was annotation-only. This PR fixes them, each with a regression test.

#758 has since been squash-merged, so this branch has been rebased onto master and now contains only the two fixes and their tests.

1. _shutdown_thread raised TypeError on every producer-thread shutdown

ThreadedProducer._shutdown_thread was a plain def overriding mode.threads.ServiceThread._shutdown_thread, which is async def and is awaited by _serve():

# mode/threads.py
async def _serve(self) -> None:
    try:
        ...
    finally:
        await self._shutdown_thread()   # <- await None

So every shutdown of the thread evaluated await None and raised TypeError. The thread only appeared to recover because _start_thread catches the exception and calls set_shutdown() before re-raising — meaning mode's teardown (on_thread_stop, stopping children, futures and exit stacks) never ran, and the thread died with a traceback rather than stopping cleanly.

There was a second layer to it. The override scheduled on_thread_stop() with:

asyncio.run_coroutine_threadsafe(self.on_thread_stop(), self.thread_loop)

thread_loop is both the loop already running _serve() and the loop about to stop. Because the TypeError tore down run_until_complete immediately, that coroutine never got a chance to run — so on this path the producer was never flushed and never stopped.

The fix makes it async def and awaits super()._shutdown_thread(), which runs on_thread_stop() and the rest of mode's teardown in the right order. The once-only guard is preserved, and when shutdown has already been initiated the shutdown event is still set — matching what the old TypeError path ended up doing via _start_thread, so stop() cannot hang.

2. publish_message(wait=True) could never succeed

The waiting branch called:

fut.message.channel._on_published(message=fut, state=state, producer=producer)

Topic._on_published takes the send future as a required positional parameter and reads the result off it:

def _on_published(self, fut, message, producer, state) -> None:
    res: RecordMetadata = fut.result()

Nothing was passed for fut, so this raised TypeError: Topic._on_published() missing 1 required positional argument: 'fut' for any real channel.

_on_published is the done-callback for the non-waiting branch, where add_done_callback supplies the future positionally. The waiting branch has no such future — send_and_wait has already resolved to ret — so it now completes the message directly: report the sensor, set the result, invoke the callback. That is exactly what Topic.publish_message(wait=True) does via _finalize_message. The non-waiting branch is unchanged.

Why the existing test missed it

test_publish_message_with_wait passes today because its channel is a bare Mock(), which accepts any call and swallows the missing argument — it pinned the bug rather than catching it. The new test uses a real topic; against the previous code it fails with the TypeError above.

Tests

Four tests added, all verified to fail against the pre-fix code:

• test_publish_message_with_wait__completes_the_message — real channel; asserts the future resolves to the send_and_wait result, the message callback fires, and on_send_completed is reported.
• test_shutdown_thread_is_a_coroutine — the structural guarantee mode's await depends on.
• test_shutdown_thread__runs_mode_teardown — delegates to the base implementation.
• test_shutdown_thread__already_initiated_still_sets_shutdown — no double on_thread_stop, but the shutdown event is still set.

scripts/check passes with the pinned toolchain; suites go from 2207 to 2211 passed, 4 skipped.

🤖 Generated with Claude Code