ConcurrentBetweenPartitionsSubscriber

async def start(self) -> None:
    """Start the consumer subgroup."""
    await super(LogicSubscriber, self).start()

    if self.calls:
        self.consumer_subgroup = [
            self.builder(
                group_id=self.group_id,
                client_id=self.client_id,
                **self._connection_args,
            )
            for _ in range(self.max_workers)
        ]

    else:
        # We should create single consumer to support
        # `get_one()` and `__aiter__` methods
        self.consumer = self.builder(
            group_id=self.group_id,
            client_id=self.client_id,
            **self._connection_args,
        )
        self.consumer_subgroup = [self.consumer]

    # Subscribers starting should be called concurrently
    # to balance them correctly
    async with anyio.create_task_group() as tg:
        for c in self.consumer_subgroup:
            c.subscribe(
                topics=self.topics,
                listener=make_logging_listener(
                    consumer=c,
                    logger=self._outer_config.logger.logger.logger,
                    log_extra=self.get_log_context(None),
                    listener=self._listener,
                ),
            )

            tg.start_soon(c.start)

    self._post_start()

    if self.calls:
        for c in self.consumer_subgroup:
            self.add_task(self._run_consume_loop, (c,))

async def stop(self) -> None:
    if self.consumer_subgroup:
        async with anyio.create_task_group() as tg:
            for consumer in self.consumer_subgroup:
                tg.start_soon(consumer.stop)

        self.consumer_subgroup = []

    await super().stop()

async def get_msg(self, consumer: "AIOKafkaConsumer") -> "KafkaRawMessage":
    assert consumer, "You should setup subscriber at first."
    message = await consumer.getone()
    message.consumer = consumer
    return cast("KafkaRawMessage", message)

def add_call(
    self,
    *,
    parser_: Optional["CustomCallable"],
    decoder_: Optional["CustomCallable"],
    dependencies_: Iterable["Dependant"],
) -> Self:
    self._call_options = _CallOptions(
        parser=parser_,
        decoder=decoder_,
        dependencies=dependencies_,
    )
    return self

async def consume(self, msg: MsgType) -> Any:
    """Consume a message asynchronously."""
    if not self.running:
        return None

    try:
        return await self.process_message(msg)

    except StopConsume:
        # Stop handler at StopConsume exception
        await self.stop()

    except SystemExit:
        # Stop handler at `exit()` call
        await self.stop()

        if app := self._outer_config.fd_config.context.get("app"):
            app.exit()

    except Exception:  # nosec B110
        # All other exceptions were logged by CriticalLogMiddleware
        pass

async def process_message(self, msg: MsgType) -> "Response":
    """Execute all message processing stages."""
    context = self._outer_config.fd_config.context
    logger_state = self._outer_config.logger

    async with AsyncExitStack() as stack:
        stack.enter_context(self.lock)

        # Enter context before middlewares
        stack.enter_context(context.scope("handler_", self))
        stack.enter_context(context.scope("logger", logger_state.logger.logger))
        for k, v in self._outer_config.extra_context.items():
            stack.enter_context(context.scope(k, v))

        # enter all middlewares
        middlewares: list[BaseMiddleware] = []
        for base_m in self.__build__middlewares_stack():
            middleware = base_m(msg, context=context)
            middlewares.append(middleware)
            await middleware.__aenter__()

        cache: dict[Any, Any] = {}
        parsing_error: Exception | None = None
        for h in self.calls:
            try:
                message = await h.is_suitable(msg, cache)
            except Exception as e:
                parsing_error = e
                break

            if message is not None:
                stack.enter_context(
                    context.scope("log_context", self.get_log_context(message)),
                )
                stack.enter_context(context.scope("message", message))

                # Middlewares should be exited before scope release
                for m in middlewares:
                    stack.push_async_exit(m.__aexit__)

                result_msg = ensure_response(
                    await h.call(
                        message=message,
                        # consumer middlewares
                        _extra_middlewares=(
                            m.consume_scope for m in middlewares[::-1]
                        ),
                    ),
                )

                if not result_msg.correlation_id:
                    result_msg.correlation_id = message.correlation_id

                for p in chain(
                    self.__get_response_publisher(message),
                    h.handler._publishers,
                ):
                    await p._publish(
                        result_msg.as_publish_command(),
                        _extra_middlewares=(
                            m.publish_scope for m in middlewares[::-1]
                        ),
                    )

                # Return data for tests
                return result_msg

        # Suitable handler was not found or
        # parsing/decoding exception occurred
        for m in middlewares:
            stack.push_async_exit(m.__aexit__)

        # Reraise it to catch in tests
        if parsing_error:
            raise parsing_error

        error_msg = f"There is no suitable handler for {msg=}"
        raise SubscriberNotFound(error_msg)

    # An error was raised and processed by some middleware
    return ensure_response(None)

@override
async def get_one(
    self,
    *,
    timeout: float = 5.0,
) -> "KafkaMessage | None":
    assert not self.calls, (
        "You can't use `get_one` method if subscriber has registered handlers."
    )

    assert self.consumer, "You should start subscriber at first."

    raw_messages = await self.consumer.getmany(
        timeout_ms=timeout * 1000,
        max_records=1,
    )

    if not raw_messages:
        return None

    ((raw_message,),) = raw_messages.values()

    context = self._outer_config.fd_config.context

    async_parser, async_decoder = self._get_parser_and_decoder()

    msg: KafkaMessage | None = await process_msg(  # type: ignore[assignment]
        msg=raw_message,
        middlewares=(
            m(raw_message, context=context) for m in self._broker_middlewares
        ),
        parser=async_parser,
        decoder=async_decoder,
    )
    return msg

def get_log_context(
    self,
    message: Optional["StreamMessage[ConsumerRecord]"],
) -> dict[str, str]:
    if message is None:
        topic = ",".join(self.topic_names)
    else:
        topic = message.raw_message.topic

    return self.build_log_context(
        message=message,
        topic=topic,
        group_id=self.group_id,
    )

def schema(self) -> dict[str, "SubscriberSpec"]:
    self._build_fastdepends_model()
    return self.specification.get_schema()

def add_task(
    self,
    func: Callable[..., Coroutine[Any, Any, Any]],
    func_args: tuple[Any, ...] | None = None,
    func_kwargs: dict[str, Any] | None = None,
) -> asyncio.Task[Any]:
    args = func_args or ()
    kwargs = func_kwargs or {}
    task = asyncio.create_task(func(*args, **kwargs))
    callback = TaskCallbackSupervisor(func, func_args, func_kwargs, self)
    task.add_done_callback(callback)
    self.tasks.append(task)
    return task

async def consume_one(self, msg: MsgType) -> None:
    await self.consume(msg)

@staticmethod
def build_log_context(
    message: Optional["StreamMessage[Any]"],
    topic: str,
    group_id: str | None = None,
) -> dict[str, str]:
    return {
        "topic": topic,
        "group_id": group_id or "",
        "message_id": getattr(message, "message_id", ""),
    }