https://github.com/PostHog/posthog/pull/3280
ee/kafka_client/client.py
if not value_serializer:
            value_serializer = self.json_serializer
        b = value_serializer(data)
        if key is not None:
            key = key.encode("utf-8")
        self.producer.send(topic, value=b, key=key)
    def close(self):
        self.producer.flush()
def can_connect():
    """
    This is intended to validate if we are able to connect to kafka, without
    actually sending any messages. I'm not amazingly pleased with this as a
    solution. Would have liked to have validated that the singleton producer was
    connected. It does expose `bootstrap_connected`, but this becomes false if
    the cluster restarts despite still being able to successfully send messages.
    I'm hoping that the load this generates on the cluster will be
    insignificant, even if it is occuring from, say, 30 separate pods, say,
    every 10 seconds.
    """
    try:
        _KafkaProducer(test=TEST)
    except kafka.errors.KafkaError:
        logger.debug("kafka_connection_failure", exc_info=True)
        return False
    return True
KafkaProducer = SingletonDecorator(_KafkaProducer)
def build_kafka_consumer(
    topic: Optional[str],