Provide errors describing Admin API

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 0.16.0 (Unreleased)
 - **[Feature]** Oauthbearer token refresh callback (bruce-szalwinski-he)
+- [Enhancement] Provide `Rrdkafka::Admin#describe_errors` to get errors descriptions (mensfeld)
 - [Enhancement] Replace time poll based wait engine with an event based to improve response times on blocking operations and wait (nijikon + mensfeld)
 - [Enhancement] Allow for usage of the second regex engine of librdkafka by setting `RDKAFKA_DISABLE_REGEX_EXT` during build (mensfeld)
 - [Enhancement] name polling Thread as `rdkafka.native_kafka#<name>` (nijikon)

lib/rdkafka/admin.rb

@@ -4,6 +4,50 @@ module Rdkafka
   class Admin
     include Helpers::OAuth
 
+    class << self
+      # Allows us to retrieve librdkafka errors with descriptions
+      # Useful for debugging and building UIs, etc.
+      #
+      # @return [Hash<Integer, Hash>] hash with errors mapped by code
+      def describe_errors
+        # Memory pointers for the array of structures and count
+        p_error_descs = FFI::MemoryPointer.new(:pointer)
+        p_count = FFI::MemoryPointer.new(:size_t)
+
+        # Call the attached function
+        Bindings.rd_kafka_get_err_descs(p_error_descs, p_count)
+
+        # Retrieve the number of items in the array
+        count = p_count.read_uint
+
+        # Get the pointer to the array of error descriptions
+        array_of_errors = FFI::Pointer.new(Bindings::NativeErrorDesc, p_error_descs.read_pointer)
+
+        errors = {}
+
+        count.times do |i|
+          # Get the pointer to each struct
+          error_ptr = array_of_errors[i]
+
+          # Create a new instance of NativeErrorDesc for each item
+          error_desc = Bindings::NativeErrorDesc.new(error_ptr)
+
+          # Read values from the struct
+          code = error_desc[:code]
+
+          name = ''
+          desc = ''
+
+          name = error_desc[:name].read_string unless error_desc[:name].null?
+          desc = error_desc[:desc].read_string unless error_desc[:desc].null?
+
+          errors[code] = { code: code, name: name, description: desc }
+        end
+
+        errors
+      end
+    end
+
     # @private
     def initialize(native_kafka)
       @native_kafka = native_kafka

lib/rdkafka/bindings.rb

@@ -90,9 +90,15 @@ class TopicPartitionList < FFI::Struct
     attach_function :rd_kafka_topic_partition_list_copy, [:pointer], :pointer
 
     # Errors
+    class NativeErrorDesc < FFI::Struct
+      layout :code, :int,
+             :name, :pointer,
+             :desc, :pointer
+    end
 
     attach_function :rd_kafka_err2name, [:int], :string
     attach_function :rd_kafka_err2str, [:int], :string
+    attach_function :rd_kafka_get_err_descs, [:pointer, :pointer], :void
 
     # Configuration
 

spec/rdkafka/admin_spec.rb

@@ -31,6 +31,14 @@
   let(:operation)             {Rdkafka::Bindings::RD_KAFKA_ACL_OPERATION_READ}
   let(:permission_type)       {Rdkafka::Bindings::RD_KAFKA_ACL_PERMISSION_TYPE_ALLOW}
 
+  describe '#describe_errors' do
+    let(:errors) { admin.class.describe_errors }
+
+    it { expect(errors.size).to eq(162) }
+    it { expect(errors[-184]).to eq(code: -184, description: 'Local: Queue full', name: '_QUEUE_FULL') }
+    it { expect(errors[21]).to eq(code: 21, description: 'Broker: Invalid required acks value', name: 'INVALID_REQUIRED_ACKS') }
+  end
+
   describe 'admin without auto-start' do
     let(:admin) { config.admin(native_kafka_auto_start: false) }