class_docstr: Kafka event broker.
class_docstr_tok: ["Kafka","event","broker","."]
code: def _publish(self, event: Dict[Text, Any]) -> None:
      if self.partition_by_sender:
          partition_key = bytes(
              event.get("sender_id"), encoding=DEFAULT_ENCODING
          )
      else:
          partition_key = None
  
      headers = []
      if self.rasa_environment:
          headers = [
              (
                  "RASA_ENVIRONMENT",
                  bytes(
                      self.rasa_environment, encoding=DEFAULT_ENCODING
                  ),
              )
          ]
  
      reduced_event = rasa.shared.core.events.remove_parse_data(event)
      structlogger.debug(
          "kafka.publish.event",
          event_info="Logging a reduced version of the Kafka event",
          topic=self.topic,
          rasa_event=reduced_event,
          partition_key=partition_key,
          headers=headers,
      )
  
      serialized_event = json.dumps(event).encode(DEFAULT_ENCODING)
  
      if self.producer is not None:
          self.producer.produce(
              self.topic,
              value=serialized_event,
              key=partition_key,
              headers=headers,
              on_delivery=delivery_report,
          )
  
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 6 for calling the KafkaEventBroker obj's underlying member method code with example usage: obj._publish(event) without return types