provide primitives to fetch consumer group metadata pointers (#393)

Author: mensfeld

lib/rdkafka/bindings.rb

@@ -191,6 +191,9 @@ class TopicPartitionList < FFI::Struct
     attach_function :rd_kafka_seek, [:pointer, :int32, :int64, :int], :int, blocking: true
     attach_function :rd_kafka_offsets_for_times, [:pointer, :pointer, :int], :int, blocking: true
     attach_function :rd_kafka_position, [:pointer, :pointer], :int, blocking: true
+    # those two are used for eos support
+    attach_function :rd_kafka_consumer_group_metadata, [:pointer], :pointer, blocking: true
+    attach_function :rd_kafka_consumer_group_metadata_destroy, [:pointer], :void, blocking: true
 
     # Headers
     attach_function :rd_kafka_header_get_all, [:pointer, :size_t, :pointer, :pointer, SizePtr], :int

lib/rdkafka/callbacks.rb

@@ -261,7 +261,6 @@ def self.process_describe_acl(event_ptr)
           end
         end
       end
-
     end
 
     # FFI Function used for Message Delivery callbacks

lib/rdkafka/consumer.rb

@@ -19,17 +19,17 @@ def initialize(native_kafka)
       @native_kafka = native_kafka
     end
 
-    def finalizer
-      ->(_) { close }
-    end
-
     # @return [String] consumer name
     def name
       @name ||= @native_kafka.with_inner do |inner|
         ::Rdkafka::Bindings.rd_kafka_name(inner)
       end
     end
 
+    def finalizer
+      ->(_) { close }
+    end
+
     # Close this consumer
     # @return [nil]
     def close
@@ -239,7 +239,7 @@ def assignment_lost?
     # @param timeout_ms [Integer] The timeout for fetching this information.
     # @return [TopicPartitionList]
     # @raise [RdkafkaError] When getting the committed positions fails.
-    def committed(list=nil, timeout_ms=1200)
+    def committed(list=nil, timeout_ms=2000)
       closed_consumer_check(__method__)
 
       if list.nil?
@@ -672,6 +672,22 @@ def each_batch(max_items: 100, bytes_threshold: Float::INFINITY, timeout_ms: 250
       end
     end
 
+    # Returns pointer to the consumer group metadata. It is used only in the context of
+    # exactly-once-semantics in transactions, this is why it is never remapped to Ruby
+    #
+    # This API is **not** usable by itself from Ruby
+    #
+    # @note This pointer **needs** to be removed with `#rd_kafka_consumer_group_metadata_destroy`
+    #
+    # @private
+    def consumer_group_metadata_pointer
+      closed_consumer_check(__method__)
+
+      @native_kafka.with_inner do |inner|
+        Bindings.rd_kafka_consumer_group_metadata(inner)
+      end
+    end
+
     private
 
     def closed_consumer_check(method)

spec/rdkafka/consumer_spec.rb

@@ -1133,6 +1133,16 @@ def send_one_message(val)
     end
   end
 
+  describe '#consumer_group_metadata_pointer' do
+    let(:pointer) { consumer.consumer_group_metadata_pointer }
+
+    after { Rdkafka::Bindings.rd_kafka_consumer_group_metadata_destroy(pointer) }
+
+    it 'expect to return a pointer' do
+      expect(pointer).to be_a(FFI::Pointer)
+    end
+  end
+
   describe "a rebalance listener" do
     let(:consumer) do
       config = rdkafka_consumer_config