Add admin#delete_group utility (#353)

* Add admin#delete_group utility

* change test order

Author: piotaixr

lib/rdkafka.rb

@@ -11,6 +11,8 @@
 require "rdkafka/admin"
 require "rdkafka/admin/create_topic_handle"
 require "rdkafka/admin/create_topic_report"
+require "rdkafka/admin/delete_groups_handle"
+require "rdkafka/admin/delete_groups_report"
 require "rdkafka/admin/delete_topic_handle"
 require "rdkafka/admin/delete_topic_report"
 require "rdkafka/admin/create_acl_handle"

lib/rdkafka/admin.rb

@@ -106,6 +106,59 @@ def create_topic(topic_name, partition_count, replication_factor, topic_config={
       create_topic_handle
     end
 
+    def delete_group(group_id)
+      closed_admin_check(__method__)
+
+      # Create a rd_kafka_DeleteGroup_t representing the new topic
+      delete_groups_ptr = Rdkafka::Bindings.rd_kafka_DeleteGroup_new(
+        FFI::MemoryPointer.from_string(group_id)
+      )
+
+      pointer_array = [delete_groups_ptr]
+      groups_array_ptr = FFI::MemoryPointer.new(:pointer)
+      groups_array_ptr.write_array_of_pointer(pointer_array)
+
+      # Get a pointer to the queue that our request will be enqueued on
+      queue_ptr = @native_kafka.with_inner do |inner|
+        Rdkafka::Bindings.rd_kafka_queue_get_background(inner)
+      end
+      if queue_ptr.null?
+        Rdkafka::Bindings.rd_kafka_DeleteTopic_destroy(delete_topic_ptr)
+        raise Rdkafka::Config::ConfigError.new("rd_kafka_queue_get_background was NULL")
+      end
+
+      # Create and register the handle we will return to the caller
+      delete_groups_handle = DeleteGroupsHandle.new
+      delete_groups_handle[:pending] = true
+      delete_groups_handle[:response] = -1
+      DeleteGroupsHandle.register(delete_groups_handle)
+      admin_options_ptr = @native_kafka.with_inner do |inner|
+        Rdkafka::Bindings.rd_kafka_AdminOptions_new(inner, Rdkafka::Bindings::RD_KAFKA_ADMIN_OP_DELETETOPICS)
+      end
+      Rdkafka::Bindings.rd_kafka_AdminOptions_set_opaque(admin_options_ptr, delete_groups_handle.to_ptr)
+
+      begin
+        @native_kafka.with_inner do |inner|
+          Rdkafka::Bindings.rd_kafka_DeleteGroups(
+            inner,
+            groups_array_ptr,
+            1,
+            admin_options_ptr,
+            queue_ptr
+          )
+        end
+      rescue Exception
+        DeleteGroupsHandle.remove(delete_groups_handle.to_ptr.address)
+        raise
+      ensure
+        Rdkafka::Bindings.rd_kafka_AdminOptions_destroy(admin_options_ptr)
+        Rdkafka::Bindings.rd_kafka_queue_destroy(queue_ptr)
+        Rdkafka::Bindings.rd_kafka_DeleteGroup_destroy(delete_groups_ptr)
+      end
+
+      delete_groups_handle
+    end
+
     # Deletes the named topic
     #
     # @return [DeleteTopicHandle] Delete topic handle that can be used to wait for the result of

lib/rdkafka/admin/delete_groups_handle.rb

@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+module Rdkafka
+  class Admin
+    class DeleteGroupsHandle < AbstractHandle
+      layout :pending, :bool, # TODO: ???
+             :response, :int,
+             :error_string, :pointer,
+             :result_name, :pointer
+
+      # @return [String] the name of the operation
+      def operation_name
+        "delete groups"
+      end
+
+      def create_result
+        DeleteGroupsReport.new(self[:error_string], self[:result_name])
+      end
+
+      def raise_error
+        raise RdkafkaError.new(
+            self[:response],
+            broker_message: create_result.error_string
+        )
+      end
+    end
+  end
+end

lib/rdkafka/admin/delete_groups_report.rb

@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+module Rdkafka
+  class Admin
+    class DeleteGroupsReport
+      # Any error message generated from the DeleteTopic
+      # @return [String]
+      attr_reader :error_string
+
+      # The name of the topic deleted
+      # @return [String]
+      attr_reader :result_name
+
+      def initialize(error_string, result_name)
+        if error_string != FFI::Pointer::NULL
+          @error_string = error_string.read_string
+        end
+        if result_name != FFI::Pointer::NULL
+          @result_name = @result_name = result_name.read_string
+        end
+      end
+    end
+  end
+end

lib/rdkafka/bindings.rb

@@ -302,6 +302,17 @@ def self.partitioner(str, partition_count, partitioner_name = "consistent_random
     attach_function :rd_kafka_event_DeleteTopics_result, [:pointer], :pointer, blocking: true
     attach_function :rd_kafka_DeleteTopics_result_topics, [:pointer, :pointer], :pointer, blocking: true
 
+    # Delete Group
+
+    RD_KAFKA_ADMIN_OP_DELETEGROUPS = 7   # rd_kafka_admin_op_t
+    RD_KAFKA_EVENT_DELETEGROUPS_RESULT = 106 # rd_kafka_event_type_t
+
+    attach_function :rd_kafka_DeleteGroups, [:pointer, :pointer, :size_t, :pointer, :pointer], :void, blocking: true
+    attach_function :rd_kafka_DeleteGroup_new, [:pointer], :pointer, blocking: true
+    attach_function :rd_kafka_DeleteGroup_destroy, [:pointer], :void, blocking: true
+    attach_function :rd_kafka_event_DeleteGroups_result, [:pointer], :pointer, blocking: true # rd_kafka_event_t* => rd_kafka_DeleteGroups_result_t*
+    attach_function :rd_kafka_DeleteGroups_result_groups, [:pointer, :pointer], :pointer, blocking: true # rd_kafka_DeleteGroups_result_t*, size_t* => rd_kafka_group_result_t**
+
     # Background Queue and Callback
 
     attach_function :rd_kafka_queue_get_background, [:pointer], :pointer
@@ -408,5 +419,18 @@ def self.partitioner(str, partition_count, partitioner_name = "consistent_random
     attach_function :rd_kafka_event_error, [:pointer], :int32
     attach_function :rd_kafka_event_error_string, [:pointer], :pointer
     attach_function :rd_kafka_AclBinding_error, [:pointer], :pointer
+
+
+    # Extracting data from group results
+    class NativeError < FFI::Struct # rd_kafka_error_t
+      layout :code, :int32,
+             :errstr, :pointer,
+             :fatal, :uint8_t,
+             :retriable, :uint8_t,
+             :txn_requires_abort, :uint8_t
+    end
+
+    attach_function :rd_kafka_group_result_error, [:pointer], NativeError.by_ref # rd_kafka_group_result_t* => rd_kafka_error_t*
+    attach_function :rd_kafka_group_result_name, [:pointer], :pointer
   end
 end

lib/rdkafka/callbacks.rb

@@ -23,6 +23,29 @@ def self.create_topic_results_from_array(count, array_pointer)
       end
     end
 
+    class GroupResult
+      attr_reader :result_error, :error_string, :result_name
+      def initialize(group_result_pointer)
+        native_error = Rdkafka::Bindings.rd_kafka_group_result_error(group_result_pointer)
+
+        if native_error.null?
+          @result_error = 0
+          @error_string = FFI::Pointer::NULL
+        else
+          @result_error = native_error[:code]
+          @error_string = native_error[:errstr]
+        end
+
+        @result_name = Rdkafka::Bindings.rd_kafka_group_result_name(group_result_pointer)
+      end
+      def self.create_group_results_from_array(count, array_pointer)
+        (1..count).map do |index|
+          result_pointer = (array_pointer + (index - 1)).read_pointer
+          new(result_pointer)
+        end
+      end
+    end
+
     # Extracts attributes of rd_kafka_acl_result_t
     #
     # @private
@@ -111,6 +134,8 @@ def self.call(_, event_ptr, _)
           process_delete_acl(event_ptr)
         elsif event_type == Rdkafka::Bindings::RD_KAFKA_EVENT_DESCRIBEACLS_RESULT
           process_describe_acl(event_ptr)
+        elsif event_type == Rdkafka::Bindings::RD_KAFKA_EVENT_DELETEGROUPS_RESULT
+          process_delete_groups(event_ptr)
         end
       end
 
@@ -133,6 +158,23 @@ def self.process_create_topic(event_ptr)
         end
       end
 
+      def self.process_delete_groups(event_ptr)
+        delete_groups_result = Rdkafka::Bindings.rd_kafka_event_DeleteGroups_result(event_ptr)
+
+        # Get the number of delete group results
+        pointer_to_size_t = FFI::MemoryPointer.new(:size_t)
+        delete_group_result_array = Rdkafka::Bindings.rd_kafka_DeleteGroups_result_groups(delete_groups_result, pointer_to_size_t)
+        delete_group_results = GroupResult.create_group_results_from_array(pointer_to_size_t.read_int, delete_group_result_array) # TODO fix this
+        delete_group_handle_ptr = Rdkafka::Bindings.rd_kafka_event_opaque(event_ptr)
+
+        if (delete_group_handle = Rdkafka::Admin::DeleteGroupsHandle.remove(delete_group_handle_ptr.address))
+          delete_group_handle[:response] = delete_group_results[0].result_error
+          delete_group_handle[:error_string] = delete_group_results[0].error_string
+          delete_group_handle[:result_name] = delete_group_results[0].result_name
+          delete_group_handle[:pending] = false
+        end
+      end
+
       def self.process_delete_topic(event_ptr)
         delete_topics_result = Rdkafka::Bindings.rd_kafka_event_DeleteTopics_result(event_ptr)
 

spec/rdkafka/admin_spec.rb

@@ -20,6 +20,7 @@
   let(:topic_replication_factor) { 1 }
   let(:topic_config)             { {"cleanup.policy" => "compact", "min.cleanable.dirty.ratio" => 0.8} }
   let(:invalid_topic_config)     { {"cleeeeenup.policee" => "campact"} }
+  let(:group_name)               { "test-group-#{Random.new.rand(0..1_000_000)}" }
 
   let(:resource_name)         {"acl-test-topic"}
   let(:resource_type)         {Rdkafka::Bindings::RD_KAFKA_RESOURCE_TOPIC}
@@ -306,7 +307,63 @@
         delete_acl_report = delete_acl_handle.wait(max_wait_timeout: 15.0)
         expect(delete_acl_handle[:response]).to eq(0)
         expect(delete_acl_report.deleted_acls.length).to eq(2)
+
+      end
+    end
+  end
+
+  describe('Group tests') do
+    describe "#delete_group" do
+      describe("with an existing group") do
+        let(:consumer_config) { rdkafka_consumer_config('group.id': group_name) }
+        let(:producer_config) { rdkafka_producer_config }
+        let(:producer) { producer_config.producer }
+        let(:consumer) { consumer_config.consumer }
+
+        before do
+          # Create a topic, post a message to it, consume it and commit offsets, this will create a group that we can then delete.
+          admin.create_topic(topic_name, topic_partition_count, topic_replication_factor).wait(max_wait_timeout: 15.0)
+
+          producer.produce(topic: topic_name, payload: "test", key: "test").wait(max_wait_timeout: 15.0)
+
+          consumer.subscribe(topic_name)
+          wait_for_assignment(consumer)
+          message = consumer.poll(100)
+
+          expect(message).to_not be_nil
+
+          consumer.commit
+          consumer.close
+        end
+
+        after do
+          producer.close
+          consumer.close
+        end
+
+        it "deletes the group" do
+          delete_group_handle = admin.delete_group(group_name)
+          report = delete_group_handle.wait(max_wait_timeout: 15.0)
+
+          expect(report.result_name).to eql(group_name)
+        end
       end
+
+      describe "called with invalid input" do
+        describe "with the name of a group that does not exist" do
+          it "raises an exception" do
+            delete_group_handle = admin.delete_group(group_name)
+
+            expect {
+              delete_group_handle.wait(max_wait_timeout: 15.0)
+            }.to raise_exception { |ex|
+              expect(ex).to be_a(Rdkafka::RdkafkaError)
+              expect(ex.message).to match(/Broker: The group id does not exist \(group_id_not_found\)/)
+            }
+          end
+        end
+      end
+
     end
   end
 end