AsyncConfluentProducer

async def stop(self) -> None:
    """Stop the Kafka producer and flush remaining messages."""
    if self.__running:
        self.__running = False
        if not self._poll_task.done():
            self._poll_task.cancel()
        await call_or_await(self.producer.flush)

async def flush(self) -> None:
    await call_or_await(self.producer.flush)

async def send(
    self,
    topic: str,
    value: bytes | str | None = None,
    key: bytes | str | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: list[tuple[str, str | bytes]] | None = None,
    no_confirm: bool = False,
) -> "asyncio.Future[Message | None] | Message | None":
    """Sends a single message to a Kafka topic."""
    kwargs: _SendKwargs = {
        "value": value,
        "key": key,
        "headers": headers,
    }

    if partition is not None:
        kwargs["partition"] = partition

    if timestamp_ms is not None:
        kwargs["timestamp"] = timestamp_ms

    loop = asyncio.get_running_loop()
    result_future: asyncio.Future[Message | None] = loop.create_future()

    def ack_callback(err: Any, msg: Message | None) -> None:
        if err or (msg is not None and (err := msg.error())):
            loop.call_soon_threadsafe(
                result_future.set_exception,
                KafkaException(err),
            )
        else:
            loop.call_soon_threadsafe(result_future.set_result, msg)

    kwargs["on_delivery"] = ack_callback

    # should be sync to prevent segfault
    # confluent stub expects bytes|None for value/key; we accept str and encode
    produce_value: bytes | None = (
        kwargs["value"]
        if isinstance(kwargs["value"], (bytes, type(None)))
        else kwargs["value"].encode()
    )
    produce_key: bytes | None = (
        kwargs["key"]
        if isinstance(kwargs["key"], (bytes, type(None)))
        else kwargs["key"].encode()
    )
    produce_headers: (
        dict[str, str | bytes | None] | list[tuple[str, str | bytes | None]] | None
    ) = cast("Any", kwargs["headers"]) if kwargs.get("headers") is not None else None
    produce_kwargs: dict[str, Any] = {
        "value": produce_value,
        "key": produce_key,
        "headers": produce_headers,
        "on_delivery": kwargs["on_delivery"],
    }
    if kwargs.get("partition") is not None:
        produce_kwargs["partition"] = kwargs["partition"]
    if kwargs.get("timestamp") is not None:
        produce_kwargs["timestamp"] = kwargs["timestamp"]
    self.producer.produce(topic, **produce_kwargs)

    if no_confirm:
        return result_future
    return await result_future

def create_batch(self) -> "BatchBuilder":  # noqa: PLR6301
    """Creates a batch for sending multiple messages."""
    return BatchBuilder()

async def send_batch(
    self,
    batch: "BatchBuilder",
    topic: str,
    *,
    partition: int | None,
    no_confirm: bool = False,
) -> None:
    """Sends a batch of messages to a Kafka topic."""
    async with anyio.create_task_group() as tg:
        for msg in batch._builder:
            _ = tg.start_soon(
                self.send,
                topic,
                msg["value"],
                msg["key"],
                partition,
                msg["timestamp_ms"],
                msg["headers"],
                no_confirm,
            )

async def ping(
    self,
    timeout: float | None = 5.0,
) -> bool:
    """Implement ping using `list_topics` information request."""
    if timeout is None:
        timeout = -1

    try:
        cluster_metadata = await call_or_await(
            self.producer.list_topics,
            timeout=timeout,
        )

        return bool(cluster_metadata)

    except Exception:
        return False