def brokers
      @brokers_mutex.synchronize do
        @brokers ||= begin
          brokers = zk.get_children(path: "/brokers/ids")

          if brokers.fetch(:rc) != Zookeeper::Constants::ZOK
            raise NoClusterRegistered, "No Kafka cluster registered on this Zookeeper location."
          end

          result, mutex = {}, Mutex.new
          threads = brokers.fetch(:children).map do |id|
            Thread.new do
              Thread.abort_on_exception = true
              broker_info = zk.get(path: "/brokers/ids/#{id}")
              raise Kazoo::Error, "Failed to retrieve broker info. Error code: #{broker_info.fetch(:rc)}" unless broker_info.fetch(:rc) == Zookeeper::Constants::ZOK

              broker = Kazoo::Broker.from_json(self, id, JSON.parse(broker_info.fetch(:data)))
              mutex.synchronize { result[id.to_i] = broker }
            end
          end
          threads.each(&:join)
          result
        end
      end
    end
def brokers
      @brokers_mutex.synchronize do
        @brokers ||= begin
          brokers = zk.get_children(path: "/brokers/ids")

          if brokers.fetch(:rc) != Zookeeper::Constants::ZOK
            raise NoClusterRegistered, "No Kafka cluster registered on this Zookeeper location."
          end

          result, mutex = {}, Mutex.new
          threads = brokers.fetch(:children).map do |id|
            Thread.new do
              Thread.abort_on_exception = true
              broker_info = zk.get(path: "/brokers/ids/#{id}")
              raise Kazoo::Error, "Failed to retrieve broker info. Error code: #{broker_info.fetch(:rc)}" unless broker_info.fetch(:rc) == Zookeeper::Constants::ZOK

              broker = Kazoo::Broker.from_json(self, id, JSON.parse(broker_info.fetch(:data)))
              mutex.synchronize { result[id.to_i] = broker }
            end
          end
          threads.each(&:join)
          result
        end
      end
    end
Returns a hash of all the brokers in the