AioKafkaFastProducerImpl

async def connect(
    self,
    producer: "AIOKafkaProducer",
    serializer: Optional["SerializerProto"],
    codec: Optional["CodecProto"] = None,
) -> None:
    self.serializer = serializer
    self.codec = codec or DefaultCodec()
    await producer.start()
    self._producer = RealProducer(producer)

async def disconnect(self) -> None:
    await self._producer.stop()
    self._producer = EmptyProducerState()

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

@override
async def publish(
    self,
    cmd: "KafkaPublishCommand",
) -> Union["asyncio.Future[RecordMetadata]", "RecordMetadata"]:
    """Publish a message to a topic."""
    if cmd.body is None and cmd.key is not None:
        # keyed None is a tombstone: aiokafka requires at least key or value,
        # so a keyless None still goes through the codec as b""
        message, content_type = None, None
    else:
        message, content_type = await self.codec.encode(cmd.body, self.serializer)

    headers_to_send = {
        "content-type": content_type or "",
        **cmd.headers_to_publish(),
    }

    send_future = await self._producer.producer.send(
        topic=cmd.destination,
        value=message,
        key=cmd.key,
        partition=cmd.partition,
        timestamp_ms=cmd.timestamp_ms,
        headers=encode_headers(headers_to_send),
    )

    if not cmd.no_confirm:
        return await send_future
    return send_future

@override
async def publish_batch(
    self,
    cmd: "KafkaPublishCommand",
) -> Union["asyncio.Future[RecordMetadata]", "RecordMetadata"]:
    """Publish a batch of messages to a topic."""
    batch = self._producer.producer.create_batch()

    headers_to_send = cmd.headers_to_publish()

    if isinstance(self.codec, BatchCodecProto):
        encoded_batch = await self.codec.encode_batch(
            cmd.batch_bodies, self.serializer
        )
    else:
        encoded_batch = [
            await self.codec.encode(body, self.serializer)
            for body in cmd.batch_bodies
        ]

    for message_position, (message, content_type) in enumerate(encoded_batch):
        if content_type:
            final_headers = {
                "content-type": content_type,
                **headers_to_send,
            }
        else:
            final_headers = headers_to_send.copy()

        metadata = batch.append(
            key=cmd.key_for(message_position),
            value=message,
            timestamp=cmd.timestamp_ms,
            headers=encode_headers(final_headers),
        )
        if metadata is None:
            raise BatchBufferOverflowException(message_position=message_position)

    send_future = await self._producer.producer.send_batch(
        batch,
        cmd.destination,
        partition=cmd.partition,
    )
    if not cmd.no_confirm:
        return await send_future
    return send_future

@override
async def request(self, cmd: "KafkaPublishCommand") -> Any:
    msg = "Kafka doesn't support `request` method without test client."
    raise FeatureNotSupportedException(msg)