def METHOD_NAME(self, event: Union[RunEvent, DatasetEvent, JobEvent]) -> None:  # noqa: UP007
    if self._is_airflow_sqlalchemy:
        self._setup_producer(self.kafka_config.config)
    self.producer.produce(  # type: ignore[attr-defined]
        topic=self.topic,
        value=Serde.to_json(event).encode("utf-8"),
        on_delivery=on_delivery,
    )
    if self.flush:
        rest = self.producer.flush(timeout=10)  # type: ignore[attr-defined]
        log.debug("Amount of messages left in Kafka buffers after flush %d", rest)
    if self._is_airflow_sqlalchemy:
        self.producer = None