adds debugging for CorruptMessage errors

Author: ronan-mch

lib/kafka/broker.rb

@@ -71,7 +71,29 @@ def list_offsets(**options)
     def produce(**options)
       request = Protocol::ProduceRequest.new(**options)
 
-      send_request(request)
+      resp = send_request(request)
+
+      return if resp.nil?
+
+      errored_partitions = []
+      resp.topics.each do |topic_info|
+        errored = topic_info.partitions.find {|part_info| part_info.error_code == 2 }
+        errored_partitions << errored unless errored.nil?
+      end
+
+      errored_partitions.each do |partition_info|
+        @logger.error "Corrupt message code received for #{request} to partition #{partition_info.partition}"
+
+        # {"topic" => {"partition" => [message, message] }}
+        request.messages_for_topics.each do |topic, messages_for_partition|
+          messages_for_partition.fetch(partition_info.partition, []).each do |message|
+            @logger.error "Message key: #{message.key}"
+            @logger.error "Message value: #{message.value}"
+          end
+        end
+      end
+
+      resp
     end
 
     def fetch_offsets(**options)

spec/broker_spec.rb

@@ -69,6 +69,31 @@ def close
   describe "#produce" do
     let(:message) { Kafka::Protocol::Message.new(key: "yo", value: "lo") }
 
+    # This spec's only function is to exercise the error handling code path
+    it "gives me a console if a partition is in error" do
+      response = Kafka::Protocol::ProduceResponse.new
+      errored_partition_info = Kafka::Protocol::ProduceResponse::PartitionInfo.new(
+        partition: 3, error_code: 2, offset: 0, timestamp: Time.now.to_s
+      )
+      topic_info = Kafka::Protocol::ProduceResponse::TopicInfo.new(
+        topic: "articles", partitions: [errored_partition_info]
+      )
+      allow(response).to receive(:topics) { [topic_info] }
+      connection.mock_response(response)
+
+      actual_response = broker.produce(
+        required_acks: -1, # -1 means all replicas must ack
+        timeout: 1,
+        messages_for_topics: {
+          "yolos" => {
+            3 => [message],
+          }
+        }
+      )
+
+      expect(actual_response).to eq response
+
+    end
     it "waits for a response if acknowledgements are required" do
       response = Kafka::Protocol::ProduceResponse.new
       connection.mock_response(response)