class_docstr: Kafka event broker.
class_docstr_tok: ["Kafka","event","broker","."]
docstr: Kafka event broker.
  
  Args:
      url: 'url[:port]' string (or list of 'url[:port]'
          strings) that the producer should contact to bootstrap initial
          cluster metadata. This does not have to be the full node list.
          It just needs to have at least one broker that will respond to a
          Metadata API Request.
      topic: Topics to subscribe to.
      client_id: A name for this client. This string is passed in each request
          to servers and can be used to identify specific server-side log entries
          that correspond to this client. Also submitted to `GroupCoordinator` for
          logging with respect to producer group administration.
      partition_by_sender: Flag to configure whether messages are partitioned by
          sender_id or not
      sasl_username: Username for plain authentication.
      sasl_password: Password for plain authentication.
      sasl_mechanism: Authentication mechanism when security_protocol is
          configured for SASL_PLAINTEXT or SASL_SSL.
          Valid values are: PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256,
          SCRAM-SHA-512. Default: `PLAIN`
      ssl_cafile: Optional filename of ca file to use in certificate
          verification.
  
      ssl_certfile : Optional filename of file in pem format containing
          the client certificate, as well as any ca certificates needed to
          establish the certificate's authenticity.
  
      ssl_keyfile : Optional filename containing the client private key.
  
      ssl_check_hostname : Flag to configure whether ssl handshake
          should verify that the certificate matches the broker's hostname.
  
      security_protocol : Protocol used to communicate with brokers.
          Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.
docstr_tok: ["Kafka","event","broker",".","Args",":","url",":","'url","[",":","port","]","'","string","(","or","list","of","'url","[",":","port","]","'","strings",")","that","the","producer","should","contact","to","bootstrap","initial","cluster","metadata",".","This","does","not","have","to","be","the","full","node","list",".","It","just","needs","to","have","at","least","one","broker","that","will","respond","to","a","Metadata","API","Request",".","topic",":","Topics","to","subscribe","to",".","client_id",":","A","name","for","this","client",".","This","string","is","passed","in","each","request","to","servers","and","can","be","used","to","identify","specific","server-side","log","entries","that","correspond","to","this","client",".","Also","submitted","to","`","GroupCoordinator","`","for","logging","with","respect","to","producer","group","administration",".","partition_by_sender",":","Flag","to","configure","whether","messages","are","partitioned","by","sender_id","or","not","sasl_username",":","Username","for","plain","authentication",".","sasl_password",":","Password","for","plain","authentication",".","sasl_mechanism",":","Authentication","mechanism","when","security_protocol","is","configured","for","SASL_PLAINTEXT","or","SASL_SSL",".","Valid","values","are",":","PLAIN",",","GSSAPI",",","OAUTHBEARER",",","SCRAM-SHA-256",",","SCRAM-SHA-512",".","Default",":","`","PLAIN","`","ssl_cafile",":","Optional","filename","of","ca","file","to","use","in","certificate","verification",".","ssl_certfile",":","Optional","filename","of","file","in","pem","format","containing","the","client","certificate",",","as","well","as","any","ca","certificates","needed","to","establish","the","certificate","'s","authenticity",".","ssl_keyfile",":","Optional","filename","containing","the","client","private","key",".","ssl_check_hostname",":","Flag","to","configure","whether","ssl","handshake","should","verify","that","the","certificate","matches","the","broker","'s","hostname",".","security_protocol",":","Protocol","used","to","communicate","with","brokers",".","Valid","values","are",":","PLAINTEXT",",","SSL",",","SASL_PLAINTEXT",",","SASL_SSL","."]
code: def __init__(
      self,
      url: Union[Text, List[Text], None],
      topic: Text = "rasa_core_events",
      client_id: Optional[Text] = None,
      partition_by_sender: bool = False,
      sasl_username: Optional[Text] = None,
      sasl_password: Optional[Text] = None,
      sasl_mechanism: Optional[Text] = "PLAIN",
      ssl_cafile: Optional[Text] = None,
      ssl_certfile: Optional[Text] = None,
      ssl_keyfile: Optional[Text] = None,
      ssl_check_hostname: bool = False,
      security_protocol: Text = "SASL_PLAINTEXT",
      **kwargs: Any,
  ) -> None:
      """Kafka event broker.
  
      Args:
          url: 'url[:port]' string (or list of 'url[:port]'
              strings) that the producer should contact to bootstrap initial
              cluster metadata. This does not have to be the full node list.
              It just needs to have at least one broker that will respond to a
              Metadata API Request.
          topic: Topics to subscribe to.
          client_id: A name for this client. This string is passed in each request
              to servers and can be used to identify specific server-side log entries
              that correspond to this client. Also submitted to `GroupCoordinator` for
              logging with respect to producer group administration.
          partition_by_sender: Flag to configure whether messages are partitioned by
              sender_id or not
          sasl_username: Username for plain authentication.
          sasl_password: Password for plain authentication.
          sasl_mechanism: Authentication mechanism when security_protocol is
              configured for SASL_PLAINTEXT or SASL_SSL.
              Valid values are: PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256,
              SCRAM-SHA-512. Default: `PLAIN`
          ssl_cafile: Optional filename of ca file to use in certificate
              verification.
  
          ssl_certfile : Optional filename of file in pem format containing
              the client certificate, as well as any ca certificates needed to
              establish the certificate's authenticity.
  
          ssl_keyfile : Optional filename containing the client private key.
  
          ssl_check_hostname : Flag to configure whether ssl handshake
              should verify that the certificate matches the broker's hostname.
  
          security_protocol : Protocol used to communicate with brokers.
              Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.
      """
      self.producer: Optional[Producer] = None
      self.url = url
      self.topic = topic
      self.client_id = client_id
      self.partition_by_sender = partition_by_sender
      self.security_protocol = security_protocol.upper()
      self.sasl_username = sasl_username
      self.sasl_password = sasl_password
      self.sasl_mechanism = sasl_mechanism
      self.ssl_cafile = ssl_cafile
      self.ssl_certfile = ssl_certfile
      self.ssl_keyfile = ssl_keyfile
      self.queue_size = kwargs.get("queue_size")
      self.ssl_check_hostname = "https" if ssl_check_hostname else None
  
      # Async producer implementation followed from confluent-kafka asyncio example:
      # https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/asyncio_example.py#L88  # noqa: E501
      self._loop = asyncio.get_event_loop()
      self._cancelled = False
      self._poll_thread = threading.Thread(target=self._poll_loop)
      self._poll_thread.start()
  
imports: import asyncio
  import os
  import json
  import logging
  import structlog
  import threading
  from asyncio import AbstractEventLoop
  from typing import Any, Text, List, Optional, Union, Dict, TYPE_CHECKING
  import time
  from rasa.core.brokers.broker import EventBroker
  from rasa.core.exceptions import KafkaProducerInitializationError
  from rasa.shared.utils.io import DEFAULT_ENCODING
  from rasa.utils.endpoints import EndpointConfig
  import rasa.shared.utils.common
label_desc: Use image node_id 1 to create a new KafkaEventBroker object from inherited base classes: EventBroker with example: obj = KafkaEventBroker(url, topic, client_id, partition_by_sender, sasl_username, sasl_password, sasl_mechanism, ssl_cafile, ssl_certfile, ssl_keyfile, ssl_check_hostname, security_protocol)