AsyncConfluentProducer

def __init__(
    self,
    *,
    loop: Optional[asyncio.AbstractEventLoop] = None,
    bootstrap_servers: Union[str, List[str]] = "localhost",
    client_id: Optional[str] = None,
    metadata_max_age_ms: int = 300000,
    request_timeout_ms: int = 40000,
    api_version: str = "auto",
    acks: Any = _missing,
    key_serializer: Optional[Callable[[bytes], bytes]] = None,
    value_serializer: Optional[Callable[[bytes], bytes]] = None,
    compression_type: Optional[str] = None,
    max_batch_size: int = 16384,
    partitioner: str = "consistent_random",
    max_request_size: int = 1048576,
    linger_ms: int = 0,
    send_backoff_ms: int = 100,
    retry_backoff_ms: int = 100,
    security_protocol: str = "PLAINTEXT",
    ssl_context: Optional[SSLContext] = None,
    connections_max_idle_ms: int = 540000,
    enable_idempotence: bool = False,
    transactional_id: Optional[Union[str, int]] = None,
    transaction_timeout_ms: int = 60000,
    sasl_mechanism: Optional[str] = None,
    sasl_plain_password: Optional[str] = None,
    sasl_plain_username: Optional[str] = None,
    sasl_kerberos_service_name: str = "kafka",
    sasl_kerberos_domain_name: Optional[str] = None,
    sasl_oauth_token_provider: Optional[str] = None,
) -> None:
    """Initializes the AsyncConfluentProducer with the given configuration.

    Args:
        loop (Optional[asyncio.AbstractEventLoop]): The event loop to use for asynchronous operations.
        bootstrap_servers (Union[str, List[str]]): A list of bootstrap servers for Kafka.
        client_id (Optional[str]): A unique identifier for the client.
        metadata_max_age_ms (int): The maximum age of metadata before a refresh is forced.
        request_timeout_ms (int): The maximum time to wait for a request to complete.
        api_version (str): The Kafka API version to use.
        acks (Any): The number of acknowledgments the producer requires before considering a request complete.
        key_serializer (Optional[Callable[[bytes], bytes]]): A callable to serialize the key.
        value_serializer (Optional[Callable[[bytes], bytes]]): A callable to serialize the value.
        compression_type (Optional[str]): The compression type for message batches.
        max_batch_size (int): The maximum size of a message batch.
        partitioner (str): The partitioning strategy to use when sending messages.
        max_request_size (int): The maximum size of a request in bytes.
        linger_ms (int): The time to wait before sending a batch in milliseconds.
        send_backoff_ms (int): The time to back off when sending fails.
        retry_backoff_ms (int): The time to back off when a retry is needed.
        security_protocol (str): The security protocol to use.
        ssl_context (Optional[SSLContext]): The SSL context for secure connections.
        connections_max_idle_ms (int): The maximum time a connection can be idle.
        enable_idempotence (bool): Whether to enable idempotent producer capabilities.
        transactional_id (Optional[Union[str, int]]): The transactional ID for transactional delivery.
        transaction_timeout_ms (int): The maximum time allowed for transactions.
        sasl_mechanism (str): The SASL mechanism to use for authentication.
        sasl_plain_password (Optional[str]): The password for SASL/PLAIN authentication.
        sasl_plain_username (Optional[str]): The username for SASL/PLAIN authentication.
        sasl_kerberos_service_name (str): The Kerberos service name for SASL/GSSAPI.
        sasl_kerberos_domain_name (Optional[str]): The Kerberos domain name for SASL/GSSAPI.
        sasl_oauth_token_provider (Optional[str]): The OAuth token provider for SASL/OAUTHBEARER.

    Raises:
        ValueError: If the provided bootstrap_servers is not a string or list of strings.
    """
    if isinstance(bootstrap_servers, Iterable) and not isinstance(
        bootstrap_servers, str
    ):
        bootstrap_servers = ",".join(bootstrap_servers)

    if compression_type is None:
        compression_type = "none"

    if acks is _missing or acks == "all":
        acks = -1

    self.config = {
        # "topic.metadata.refresh.interval.ms": 1000,
        "bootstrap.servers": bootstrap_servers,
        "client.id": client_id,
        "metadata.max.age.ms": metadata_max_age_ms,
        "request.timeout.ms": request_timeout_ms,
        "acks": acks,
        "compression.type": compression_type,
        "partitioner": partitioner,
        "message.max.bytes": max_request_size,
        "linger.ms": linger_ms,
        "enable.idempotence": enable_idempotence,
        "transactional.id": transactional_id,
        "transaction.timeout.ms": transaction_timeout_ms,
        "retry.backoff.ms": retry_backoff_ms,
        "security.protocol": security_protocol.lower(),
        "connections.max.idle.ms": connections_max_idle_ms,
        "sasl.kerberos.service.name": sasl_kerberos_service_name,
    }
    if sasl_mechanism:
        self.config.update(
            {
                "sasl.mechanism": sasl_mechanism,
                "sasl.username": sasl_plain_username,
                "sasl.password": sasl_plain_password,
            }
        )

    self.producer = Producer(self.config)
    # self.producer.init_transactions()
    self.producer.list_topics()
    self.loop = loop or asyncio.get_event_loop()