def follow_topic(kafka_class, name, retry_interval=1, **kafka_init):
    """Dump each message from kafka topic to stdio."""
    while True:
        try:
            client = kafka_class(**kafka_init)
            topic = client.topics[name]
            consumer = topic.get_simple_consumer(reset_offset_on_start=True)
        except Exception as e:
            if not should_try_kafka_again(e):
                raise
            with flushing(sys.stderr):
                print(
                    'Failed attempt to connect to Kafka. Will retry ...',
                    file=sys.stderr)
            sleep(retry_interval)
        else:
            with flushing(sys.stdout):
                print('Connected to Kafka.')
            break

    dump = Dump()
    for message in consumer:
        with flushing(sys.stdout, sys.stderr):
            status = load(message.value)
            if status:
                dump(status)
Dump each message from kafka topic to stdio.