From 102fdea0a519eabb142beaadbb2f0f544d272ccd Mon Sep 17 00:00:00 2001
From: Michal Wrobel <sparrovv@gmail.com>
Date: Thu, 1 Mar 2018 15:37:29 +0100
Subject: [PATCH 1/3] Disconnect only when there's a @connection

---
 lib/kafka/broker.rb |  7 ++++++-
 spec/broker_spec.rb | 20 ++++++++++++++++++++
 2 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/lib/kafka/broker.rb b/lib/kafka/broker.rb
index 5dc31af72..a706d42e8 100644
--- a/lib/kafka/broker.rb
+++ b/lib/kafka/broker.rb
@@ -24,7 +24,12 @@ def to_s
 
     # @return [nil]
     def disconnect
-      connection.close
+      connection.close if connected?
+    end
+
+    # @return [Boolean]
+    def connected?
+      !@connection.nil?
     end
 
     # Fetches cluster metadata from the broker.
diff --git a/spec/broker_spec.rb b/spec/broker_spec.rb
index 95d119e8f..29403d15c 100644
--- a/spec/broker_spec.rb
+++ b/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,21 @@ 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
 end

From a50454ac244fc86b033bc6db923df17a06aa5860 Mon Sep 17 00:00:00 2001
From: Michal Wrobel <sparrovv@gmail.com>
Date: Thu, 1 Mar 2018 15:37:55 +0100
Subject: [PATCH 2/3] Don't fail on broker.to_s

This would fail if there's a problem with connecting to a broker
---
 lib/kafka/broker.rb |  6 +++++-
 spec/broker_spec.rb | 10 ++++++++++
 2 files changed, 15 insertions(+), 1 deletion(-)

diff --git a/lib/kafka/broker.rb b/lib/kafka/broker.rb
index a706d42e8..3fcd0325a 100644
--- a/lib/kafka/broker.rb
+++ b/lib/kafka/broker.rb
@@ -19,7 +19,7 @@ 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]
@@ -163,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
diff --git a/spec/broker_spec.rb b/spec/broker_spec.rb
index 29403d15c..80573c3e6 100644
--- a/spec/broker_spec.rb
+++ b/spec/broker_spec.rb
@@ -132,4 +132,14 @@ def close
       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

From d451a9a2ee2651ff51d373fa08d71322658561c5 Mon Sep 17 00:00:00 2001
From: Michal Wrobel <sparrovv@gmail.com>
Date: Thu, 1 Mar 2018 23:19:35 +0100
Subject: [PATCH 3/3] @host and @port are available in the Broker

No need for connection.to_s
---
 lib/kafka/broker.rb |  6 +-----
 spec/broker_spec.rb | 10 ----------
 2 files changed, 1 insertion(+), 15 deletions(-)

diff --git a/lib/kafka/broker.rb b/lib/kafka/broker.rb
index 3fcd0325a..f3029b65b 100644
--- a/lib/kafka/broker.rb
+++ b/lib/kafka/broker.rb
@@ -19,7 +19,7 @@ def address_match?(host, port)
 
     # @return [String]
     def to_s
-      "#{connection_to_s} (node_id=#{@node_id.inspect})"
+      "#{@host}:#{@port} (node_id=#{@node_id.inspect})"
     end
 
     # @return [nil]
@@ -163,9 +163,5 @@ 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
diff --git a/spec/broker_spec.rb b/spec/broker_spec.rb
index 80573c3e6..29403d15c 100644
--- a/spec/broker_spec.rb
+++ b/spec/broker_spec.rb
@@ -132,14 +132,4 @@ def close
       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