Merge pull request #644 from ConnectedHomes/refresh-metadata-on-connection-error

Refresh metadata on connection error when listing topics

Author: dasch

lib/kafka/client.rb

@@ -618,7 +618,15 @@ def create_partitions_for(name, num_partitions: 1, timeout: 30)
     #
     # @return [Array<String>] the list of topic names.
     def topics
-      @cluster.list_topics
+      attempts = 0
+      begin
+        attempts += 1
+        @cluster.list_topics
+      rescue Kafka::ConnectionError
+        @cluster.mark_as_stale!
+        retry unless attempts > 1
+        raise
+      end
     end
 
     # Lists all consumer groups in the cluster

lib/kafka/cluster.rb

@@ -397,6 +397,7 @@ def fetch_cluster_info
     end
 
     def random_broker
+      refresh_metadata_if_necessary!
       node_id = cluster_info.brokers.sample.node_id
       connect_to_broker(node_id)
     end