[DMINQ]

Description

Adds configurable exception handlers following the proposal in:
#3095 (comment)

The subscriber handler runs first. If it is absent or returns False,
the broker handler runs. Both synchronous and asynchronous handlers
are supported.

This draft currently exposes the API for Kafka and adds shared
dispatch logic to SubscriberUsecase.consume(). Unhandled exceptions
retain the existing consume() behavior. Middleware logging and
acknowledgement behavior remain unchanged.

Related to #2945. Background-task error handling is not implemented
yet, so this draft does not resolve the original Redis scenario.

Validation

• Four automated test cases pass, covering handler precedence,
  fallback results and dispatch through consume().
• The consume() test was verified to fail when the dispatch call
  was removed, and to pass with it restored.
• Manually verified with a real Kafka broker: local handling,
  broker fallback, unhandled exceptions and continued processing.
• Ruff lint and formatting checks passed for the changed files.
• Mypy, Pyright and Pyrefly passed for the initial implementation.
• Targeted Mypy checks passed after adding the consume() test.

Remaining work

• Cover missing handlers and remaining sync/async combinations.
• Integrate background-task error handling.
• Expose the API across the remaining brokers.
• Add documentation and runnable examples.
• Complete the remaining project checks.

Type of change

• New feature
• This change requires a documentation update

Checklist

• Self-reviewed the current changes.
• Added automated tests for the implemented behavior.
• Added documentation and tested examples.
• Passed the full project lint checks (just lint).
• Passed the relevant coverage checks.
• Passed the full static analysis suite (just static-analysis).

[CLAassistant]

All committers have signed the CLA.

[IvanKirpichnikov]

Hi. Sorry it took so long.

The idea of using to_async in __post_init__ didn’t really work for me. It ended up feeling a bit cumbersome. Let’s go back to your approach.

Second, why are you wrapping it in _build_fastdepends_model? Let’s do it in __init__ instead. In theory, nothing should break.

Third, the tests.

Let’s write e2e tests. The current tests are fragile because you monkeypatch a lot of things and call private methods.

I think it would be better to write e2e tests like this:

async def test_subscriber_exception_handler_true_suppresses_error(
    self,
    queue: str,
    event: Event,
    mock: MagicMock,
) -> None:
    error = RuntimeError()

    async def exception_handler(exc: BaseException) -> bool:
        mock(exc)
        return True

    broker = self.get_broker()

    args, kwargs = self.get_subscriber_params(queue, exception_handler=exception_handler)

    @broker.subscriber(*args, **kwargs)
    async def handler(msg: Any) -> NoReturn:
        event.set()
        raise error

    async with self.patch_broker(broker) as br:
        await asyncio.wait(
            (
                asyncio.create_task(br.publish(queue, None)),
                asyncio.create_task(event.wait()),
            ),
            timeout=self.timeout,
        )

    mock.assert_called_one_with(error)

[IvanKirpichnikov]

We also need to add another global exception handler at the FastStream level, which will also be taken into account when handling errors, as the very last one. Tests will need to be added for it.

[IvanKirpichnikov]

And add docs

[IvanKirpichnikov]

Could you tell me why this particular condition is in place?

[IvanKirpichnikov]

may remove inject context__

[IvanKirpichnikov]

This won’t work, will it?

[IvanKirpichnikov]

I think we should split this test into two, because it’s hard to read the test with such parametrization.

[IvanKirpichnikov]

I think we should split this test into two, because it’s hard to read the test with such parametrization.

[IvanKirpichnikov]

and pull main