class_docstr: Kafka event broker.
class_docstr_tok: ["Kafka","event","broker","."]
docstr: Publishes events.
docstr_tok: ["Publishes","events","."]
code: def publish(
      self,
      event: Dict[Text, Any],
      retries: int = 60,
      retry_delay_in_seconds: float = 5,
  ) -> None:
      """Publishes events."""
      from confluent_kafka import KafkaException
  
      if retries == 1:
          retries = 2
  
      if self.producer is None:
          self.producer = self._create_producer()
          try:
              self._check_kafka_connection()
              logger.debug("Connection to kafka successful.")
          except KafkaException:
              logger.debug("Failed to connect kafka.")
              return
      while retries:
          try:
              self._publish(event)
              return
          except BufferError as e:
              logger.error(
                  f"Could not publish message to kafka url '{self.url}'. "
                  f"Failed with error: {e}"
              )
              self.producer.poll(1)
              retries -= 1
          except Exception as e:
              logger.error(
                  f"Could not publish message to kafka url '{self.url}'. "
                  f"Failed with error: {e}"
              )
              try:
                  self._check_kafka_connection()
              except KafkaException:
                  logger.debug(
                      "Connection to kafka lost, reconnecting..."
                  )
                  self.producer = self._create_producer()
                  try:
                      self._check_kafka_connection()
                      logger.debug("Reconnection to kafka successful")
                      self._publish(event)
                      return
                  except KafkaException:
                      pass
              retries -= 1
              time.sleep(retry_delay_in_seconds)
  
      logger.error("Failed to publish Kafka event.")
  
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 2 for calling the KafkaEventBroker obj's underlying member method code with example usage: obj.publish(event, retries, retry_delay_in_seconds) without return types