module Kafka::BrokerUri

def self.parse(str)
  # Make sure there's a scheme part if it's missing.
  str = "kafka://" + str unless str.include?("://")
  uri = URI.parse(str)
  uri.port ||= DEFAULT_PORT
  # Map some schemes to others.
  case uri.scheme
  when 'plaintext'
    uri.scheme = 'kafka'
  when 'ssl'
    uri.scheme = 'kafka+ssl'
  end
  unless URI_SCHEMES.include?(uri.scheme)
    raise Kafka::Error, "invalid protocol `#{uri.scheme}` in `#{str}`"
  end
  uri
end