Don't fail on retry when the cluster is secured

lib/kafka/broker.rb

@@ -19,12 +19,17 @@ def address_match?(host, port)
 
     # @return [String]
     def to_s
-      "#{connection} (node_id=#{@node_id.inspect})"
+      "#{connection_to_s} (node_id=#{@node_id.inspect})"
     end
 
     # @return [nil]
     def disconnect
-      connection.close
+      connection.close if connected?
+    end
+
+    # @return [Boolean]
+    def connected?
+      [email protected]?
     end
 
     # Fetches cluster metadata from the broker.
@@ -158,5 +163,9 @@ def send_request(request)
     def connection
       @connection ||= @connection_builder.build_connection(@host, @port)
     end
+
+    def connection_to_s
+      connection.to_s rescue nil
+    end
   end
 end

spec/broker_spec.rb

@@ -31,6 +31,9 @@ def mock_response(response)
     def send_request(request)
       @mocked_response
     end
+
+    def close
+    end
   end
 
   describe "#address_match?" do
@@ -112,4 +115,31 @@ def send_request(request)
       expect(actual_response.topics).to eq []
     end
   end
+
+  describe "#disconnect" do
+    it "doesn't close a connection if it's not connected yet " do
+      expect(connection).not_to receive(:close)
+      broker.disconnect
+    end
+
+    it "closes a connection if the connection is present" do
+      expect(connection).to receive(:close)
+
+      broker.fetch_messages(
+        max_wait_time: 0, min_bytes: 0, max_bytes: 10 * 1024, topics: {}
+      )
+
+      broker.disconnect
+    end
+  end
+
+  describe "#to_s" do
+    it "doesn't fail when building connection fails" do
+      expect(connection_builder).to receive(:build_connection).and_raise(Kafka::ConnectionError)
+
+      expect do
+        broker.to_s
+      end.not_to raise_error
+    end
+  end
 end