AsyncConfluentFastProducerImpl

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

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

async def ping(self, timeout: float) -> bool:
    return await self._producer.ping(timeout=timeout)

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

@override
async def publish(
    self,
    cmd: "KafkaPublishCommand",
) -> "asyncio.Future[Message | None] | Message | None":
    """Publish a message to a topic."""
    if cmd.body is None:
        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(),
    }

    return await self._producer.producer.send(
        topic=cmd.destination,
        value=message,
        key=cmd.key,
        partition=cmd.partition,
        timestamp_ms=cmd.timestamp_ms,
        headers=[(i, (j or "").encode()) for i, j in headers_to_send.items()],
        no_confirm=cmd.no_confirm,
    )

@override
async def publish_batch(self, cmd: "KafkaPublishCommand") -> None:
    """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(msg, self.serializer) for msg 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()

        batch.append(
            key=cmd.key_for(message_position),
            value=message,
            timestamp=cmd.timestamp_ms,
            headers=[(i, j.encode()) for i, j in final_headers.items()],
        )

    await self._producer.producer.send_batch(
        batch,
        cmd.destination,
        partition=cmd.partition,
        no_confirm=cmd.no_confirm,
    )

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