class Kafka::Cluster

def topics 

def topics
  refresh_metadata_if_necessary!
  cluster_info.topics.select do |topic|
    topic.topic_error_code == 0
  end.map(&:topic_name)
end