error.rb in karafka-rdkafka-0.17.2.rc1

- old
+ new

@@ -4,10 +4,13 @@
   # Base error class.
   class BaseError < RuntimeError; end
 
   # Error returned by the underlying rdkafka library.
   class RdkafkaError < BaseError
+    # Empty hash for details default allocation
+    EMPTY_HASH = {}.freeze
+
     # The underlying raw error response
     # @return [Integer]
     attr_reader :rdkafka_response
 
     # Prefix to be used for human readable representation
@@ -16,10 +19,14 @@
 
     # Error message sent by the broker
     # @return [String]
     attr_reader :broker_message
 
+    # Optional details hash specific to a given error or empty hash if none or not supported
+    # @return [Hash]
+    attr_reader :details
+
     class << self
       def build_from_c(response_ptr, message_prefix = nil, broker_message: nil)
         code = Rdkafka::Bindings.rd_kafka_error_code(response_ptr)
 
         return false if code.zero?
@@ -52,11 +59,25 @@
 
           unless response_ptr_or_code[:payload].null?
             message_prefix ||= response_ptr_or_code[:payload].read_string(response_ptr_or_code[:len])
           end
 
-          new(response_ptr_or_code[:err], message_prefix, broker_message: broker_message)
+          details = if response_ptr_or_code[:rkt].null?
+                      EMPTY_HASH
+                    else
+                      {
+                        partition: response_ptr_or_code[:partition],
+                        offset: response_ptr_or_code[:offset],
+                        topic: Bindings.rd_kafka_topic_name(response_ptr_or_code[:rkt])
+                      }.freeze
+                    end
+          new(
+            response_ptr_or_code[:err],
+            message_prefix,
+            broker_message: broker_message,
+            details: details
+          )
         else
           build_from_c(response_ptr_or_code, message_prefix)
         end
       end
 
@@ -71,18 +92,20 @@
       response,
       message_prefix=nil,
       broker_message: nil,
       fatal: false,
       retryable: false,
-      abortable: false
+      abortable: false,
+      details: EMPTY_HASH
     )
       raise TypeError.new("Response has to be an integer") unless response.is_a? Integer
       @rdkafka_response = response
       @message_prefix = message_prefix
       @broker_message = broker_message
       @fatal = fatal
       @retryable = retryable
       @abortable = abortable
+      @details = details
     end
 
     # This error's code, for example `:partition_eof`, `:msg_size_too_large`.
     # @return [Symbol]
     def code