BatchPublisher

@override
async def request(
    self,
    message: "SendableMessage",
    topic: str = "",
    *,
    key: bytes | Any | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    correlation_id: str | None = None,
    timeout: float = 0.5,
) -> "KafkaMessage":
    """Send a request message to Kafka topic.

    Args:
        message: Message body to send.
        topic: Topic where the message will be published.
        key: A key to associate with the message. Can be used to
            determine which partition to send the message to. If partition
            is `None` (and producer's partitioner config is left as default),
            then messages with the same key will be delivered to the same
            partition (but if key is `None`, partition is chosen randomly).
            Must be type `bytes`, or be serializable to bytes via configured
            `key_serializer`.
        partition: Specify a partition. If not set, the partition will be
            selected using the configured `partitioner`.
        timestamp_ms: Epoch milliseconds (from Jan 1 1970 UTC) to use as
            the message timestamp. Defaults to current time.
        headers: Message headers to store metainformation.
        correlation_id: Manual message **correlation_id** setter.
            **correlation_id** is a useful option to trace messages.
        timeout: Timeout to send RPC request.

    Returns:
        KafkaMessage: The response message.
    """
    cmd = KafkaPublishCommand(
        message,
        topic=topic or self.topic,
        key=key,
        partition=partition if partition is not None else self.partition,
        headers=self.headers | (headers or {}),
        correlation_id=correlation_id or gen_cor_id(),
        timestamp_ms=timestamp_ms,
        timeout=timeout,
        _publish_type=PublishType.REQUEST,
    )

    msg: KafkaMessage = await self._basic_request(
        cmd,
        producer=self._outer_config.producer,
    )
    return msg

async def start(self) -> None:
    pass

def set_test(
    self,
    *,
    mock: MagicMock,
    with_fake: bool,
) -> None:
    """Turn publisher to testing mode."""
    self.mock = mock
    self._fake_handler = with_fake

def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self._fake_handler = False
    self.mock.reset_mock()

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

async def flush(self) -> None:
    producer = cast("AioKafkaFastProducer", self._outer_config.producer)
    await producer.flush()

@override
async def publish(
    self,
    *messages: "SendableMessage",
    topic: str = "",
    key: bytes | Any | None = None,
    partition: int | None = None,
    timestamp_ms: int | None = None,
    headers: dict[str, str] | None = None,
    reply_to: str = "",
    correlation_id: str | None = None,
    no_confirm: bool = False,
) -> Union["asyncio.Future[RecordMetadata]", "RecordMetadata"]:
    """Publish a message batch as a single request to broker.

    Args:
        *messages:
            Messages bodies to send.
        topic:
            Topic where the message will be published.
        key:
            A single key to associate with every message in this batch. If a
            partition is not specified and the producer uses the default
            partitioner, messages with the same key will be routed to the
            same partition. Must be bytes or serializable to bytes via the
            configured key serializer. If omitted, falls back to the
            publisher's default key (if configured).
        partition:
            Specify a partition. If not set, the partition will be
            selected using the configured `partitioner`
        timestamp_ms:
            Epoch milliseconds (from Jan 1 1970 UTC) to use as
            the message timestamp. Defaults to current time.
        headers:
            Message headers to store metainformation.
        reply_to:
            Reply message topic name to send response.
        correlation_id:
            Manual message **correlation_id** setter.
            **correlation_id** is a useful option to trace messages.
        no_confirm:
            Do not wait for Kafka publish confirmation.

    Returns:
        `asyncio.Future[RecordMetadata]` if no_confirm = True.
        `RecordMetadata` if no_confirm = False.
    """
    cmd = KafkaPublishCommand(
        *messages,
        key=key or self.key,
        topic=topic or self.topic,
        partition=partition if partition is not None else self.partition,
        reply_to=reply_to or self.reply_to,
        headers=self.headers | (headers or {}),
        correlation_id=correlation_id or gen_cor_id(),
        timestamp_ms=timestamp_ms,
        no_confirm=no_confirm,
        _publish_type=PublishType.PUBLISH,
    )

    return await self._basic_publish_batch(
        cmd,
        producer=self._outer_config.producer,
        _extra_middlewares=(),
    )