AsyncConfluentConsumer

async def start(self) -> None:
    """Starts the Kafka consumer and subscribes to the specified topics."""
    if self.config.get("allow.auto.create.topics", True):
        topics_creation_result = await run_in_executor(
            self._thread_pool,
            self.admin_client.create_topics,
            self.topics_to_create,
        )

        for create_result in topics_creation_result:
            if create_result.error:
                self.logger_state.log(
                    log_level=logging.WARNING,
                    message=f"Failed to create topic {create_result.topic}: {create_result.error}",
                )

    else:
        self.logger_state.log(
            log_level=logging.WARNING,
            message="Auto create topics is disabled. Make sure the topics exist.",
        )

    if self.topics:
        subscribe_kwargs: dict[str, Any] = {"topics": [t.name for t in self.topics]}
        if self._on_assign is not None:
            subscribe_kwargs["on_assign"] = self._on_assign
        if self._on_revoke is not None:
            subscribe_kwargs["on_revoke"] = self._on_revoke
        if self._on_lost is not None:
            subscribe_kwargs["on_lost"] = self._on_lost
        await run_in_executor(
            self._thread_pool,
            self.consumer.subscribe,
            **subscribe_kwargs,
        )

    elif self.partitions:
        await run_in_executor(
            self._thread_pool,
            self.consumer.assign,
            [p.to_confluent() for p in self.partitions],
        )

    else:
        msg = "You must provide either `topics` or `partitions` option."
        raise SetupError(msg)

async def commit(self, asynchronous: bool = True) -> None:
    """Commits the offsets of all messages returned by the last poll operation."""
    await run_in_executor(
        self._thread_pool,
        lambda: self.consumer.commit(asynchronous=asynchronous),  # type: ignore[call-overload]
    )

async def stop(self) -> None:
    """Stops the Kafka consumer and releases all resources."""
    # NOTE: If we don't explicitly call commit and then close the consumer, the confluent consumer gets stuck.
    # We are doing this to avoid the issue.
    enable_auto_commit = self.config.get("enable.auto.commit", True)

    try:
        if enable_auto_commit:
            await self.commit(asynchronous=False)

    except Exception as e:
        # No offset stored issue is not a problem - https://github.com/confluentinc/confluent-kafka-python/issues/295#issuecomment-355907183
        if "No offset stored" in str(e):
            pass
        else:
            self.logger_state.log(
                log_level=logging.ERROR,
                message="Consumer closing error occurred.",
                exc_info=e,
            )

    # Wrap calls to async to make method cancelable by timeout
    # We shouldn't read messages and close consumer concurrently
    # https://github.com/ag2ai/faststream/issues/1904#issuecomment-2506990895
    # Now it works without lock due `ThreadPoolExecutor(max_workers=1)`
    # that makes all calls to consumer sequential
    await run_in_executor(self._thread_pool, self.consumer.close)

    self._thread_pool.shutdown(wait=False)

async def getone(self, timeout: float = 0.1) -> Message | None:
    """Consumes a single message from Kafka."""
    msg = await run_in_executor(self._thread_pool, self.consumer.poll, timeout)
    return check_msg_error(msg)

async def getmany(
    self,
    timeout: float = 0.1,
    max_records: int | None = 10,
) -> tuple[Message, ...]:
    """Consumes a batch of messages from Kafka and groups them by topic and partition."""
    raw_messages: list[Message | None] = await run_in_executor(
        self._thread_pool,
        cast(
            "Callable[..., list[Message | None]]",
            lambda: self.consumer.consume(
                num_messages=max_records or 10,
                timeout=timeout,
            ),
        ),
    )
    return tuple(x for x in map(check_msg_error, raw_messages) if x is not None)

async def seek(self, topic: str, partition: int, offset: int) -> None:
    """Seeks to the specified offset in the specified topic and partition."""
    topic_partition = TopicPartition(
        topic=topic,
        partition=partition,
        offset=offset,
    )
    await run_in_executor(
        self._thread_pool,
        self.consumer.seek,
        topic_partition.to_confluent(),
    )