class_docstr: Kafka event broker.
class_docstr_tok: ["Kafka","event","broker","."]
code: def _get_kafka_config(self) -> Dict[Text, Any]:
      config = {
          "client.id": self.client_id,
          "bootstrap.servers": self.url,
          "error_cb": kafka_error_callback,
      }
      if self.queue_size:
          config["queue.buffering.max.messages"] = self.queue_size
  
      if self.security_protocol == "PLAINTEXT":
          authentication_params: Dict[Text, Any] = {
              "security.protocol": self.security_protocol,
          }
      elif self.security_protocol == "SASL_PLAINTEXT":
          authentication_params = {
              "sasl.username": self.sasl_username,
              "sasl.password": self.sasl_password,
              "sasl.mechanism": self.sasl_mechanism,
              "security.protocol": self.security_protocol,
          }
      elif self.security_protocol == "SSL":
          authentication_params = {
              "ssl.ca.location": self.ssl_cafile,
              "ssl.certificate.location": self.ssl_certfile,
              "ssl.key.location": self.ssl_keyfile,
              "security.protocol": self.security_protocol,
          }
      elif self.security_protocol == "SASL_SSL":
          authentication_params = {
              "sasl.username": self.sasl_username,
              "sasl.password": self.sasl_password,
              "ssl.ca.location": self.ssl_cafile,
              "ssl.certificate.location": self.ssl_certfile,
              "ssl.key.location": self.ssl_keyfile,
              "ssl.endpoint.identification.algorithm": self.ssl_check_hostname,
              "security.protocol": self.security_protocol,
              "sasl.mechanism": self.sasl_mechanism,
          }
      else:
          raise ValueError(
              f"Cannot initialise `KafkaEventBroker`: "
              f"Invalid `security_protocol` ('{self.security_protocol}')."
          )
  
      return {**config, **authentication_params}
  
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 4 for calling the KafkaEventBroker obj's underlying member method code with example usage: obj._get_kafka_config() and returns: dict