check for stale cluster info before retrying

lib/kafka/cluster.rb

@@ -105,6 +105,7 @@ def refresh_metadata_if_necessary!
     # @param partition [Integer]
     # @return [Broker] the broker that's currently leader.
     def get_leader(topic, partition)
+      refresh_metadata_if_necessary!
       connect_to_broker(get_leader_id(topic, partition))
     end
 

spec/cluster_spec.rb

@@ -18,6 +18,38 @@
       allow(broker).to receive(:disconnect)
     end
 
+    it 'raises NotLeaderForPartition if the current broker is not the leader' do
+      metadata = Kafka::Protocol::MetadataResponse.new(
+        brokers: [
+          Kafka::Protocol::MetadataResponse::BrokerInfo.new(
+            node_id: 42,
+            host: "test1",
+            port: 9092,
+          )
+        ],
+        controller_id: 42,
+        topics: [
+          Kafka::Protocol::MetadataResponse::TopicMetadata.new(
+            topic_name: "greetings",
+            partitions: [
+              Kafka::Protocol::MetadataResponse::PartitionMetadata.new(
+                partition_id: 42,
+                leader: 2,
+
+                partition_error_code: 6, # <-- this is the important bit. NotLeaderForPartition
+              )
+            ]
+          )
+        ]
+      )
+
+      allow(broker).to receive(:fetch_metadata) { metadata }
+
+      expect {
+        cluster.get_leader("greetings", 42)
+      }.to raise_error Kafka::NotLeaderForPartition
+    end
+
     it "raises LeaderNotAvailable if there's no leader for the partition" do
       metadata = Kafka::Protocol::MetadataResponse.new(
         brokers: [
@@ -88,5 +120,41 @@
         cluster.get_leader("greetings", 42)
       }.to raise_exception(Kafka::ConnectionError)
     end
+
+    context 'meta data is stale' do
+      before do
+        allow(cluster).to receive(:stale) { true }
+      end
+
+      it 'should re fetch metadata' do
+        metadata = Kafka::Protocol::MetadataResponse.new(
+          brokers: [
+            Kafka::Protocol::MetadataResponse::BrokerInfo.new(
+              node_id: 42,
+              host: "test1",
+              port: 9092,
+            )
+          ],
+          controller_id: 42,
+          topics: [
+            Kafka::Protocol::MetadataResponse::TopicMetadata.new(
+              topic_name: "greetings",
+              partitions: [
+                Kafka::Protocol::MetadataResponse::PartitionMetadata.new(
+                  partition_id: 42,
+                  leader: 42,
+                  partition_error_code: 0
+                )
+              ]
+            )
+          ]
+        )
+
+        allow(broker).to receive(:fetch_metadata) { metadata }
+        expect(cluster).to receive(:refresh_metadata!)
+
+        cluster.get_leader("greetings", 42)
+      end
+    end
   end
 end