https://github.com/PostHog/posthog/pull/3280
ee/kafka_client/client.py
value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    auto_offset_reset="latest",
    test=TEST,
    group_id=None,
    consumer_timeout_ms=float("inf"),
):
    if test:
        consumer = TestKafkaConsumer(
            topic=topic, auto_offset_reset=auto_offset_reset, max=10, consumer_timeout_ms=consumer_timeout_ms
        )
    elif KAFKA_BASE64_KEYS:
        consumer = helper.get_kafka_consumer(
            topic=topic,
            auto_offset_reset=auto_offset_reset,
            value_deserializer=value_deserializer,
            group_id=group_id,
            consumer_timeout_ms=consumer_timeout_ms,
        )
    else:
        consumer = KC(
            bootstrap_servers=KAFKA_HOSTS,
            auto_offset_reset=auto_offset_reset,
            value_deserializer=value_deserializer,
            group_id=group_id,
            consumer_timeout_ms=consumer_timeout_ms,
        )
        if topic:
            consumer.subscribe([topic])
    return consumer
class ClickhouseProducer: