def METHOD_NAME(self, *args, **kwargs):
    result = super().METHOD_NAME(*args, **kwargs)
    self._start_kafka_cluster()
    @backoff.on_predicate(backoff.expo, lambda x: x > 0, max_time=30)
    def _await_kafka_cluster():
        return socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect_ex(
            ("localhost", KAFKA_BROKER_PORT)
        )
    _await_kafka_cluster()
    return result