Backport purge from karafka-rdkafka (karafka#335)

* backport purge from karafka-rdkafka

* remove not needed changes

* remove extra spec

Author: mensfeld

.ruby-gemset

@@ -0,0 +1 @@
+rdkafka-ruby

.ruby-version

@@ -0,0 +1 @@
+3.2.2

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Rdkafka Changelog
 
 ## 0.14.0 (Unreleased)
+* [Enhancement] Provide `#purge` to remove any outstanding requests from the producer (mensfeld)
 * [Enhancement] Update `librdkafka` to `2.2.0` (mensfeld)
 * [Enhancement] Introduce producer partitions count metadata cache (mensfeld)
 * [Enhancement] Increase metadata timeout request from `250 ms` to `2000 ms` default to allow for remote cluster operations via `rdkafka-ruby` (mensfeld)

lib/rdkafka/bindings.rb

@@ -249,10 +249,13 @@ class TopicPartitionList < FFI::Struct
     RD_KAFKA_VTYPE_TIMESTAMP = 8
     RD_KAFKA_VTYPE_HEADER = 9
     RD_KAFKA_VTYPE_HEADERS = 10
+    RD_KAFKA_PURGE_F_QUEUE = 1
+    RD_KAFKA_PURGE_F_INFLIGHT = 2
 
     RD_KAFKA_MSG_F_COPY = 0x2
 
     attach_function :rd_kafka_producev, [:pointer, :varargs], :int, blocking: true
+    attach_function :rd_kafka_purge, [:pointer, :int], :int, blocking: true
     callback :delivery_cb, [:pointer, :pointer, :pointer], :void
     attach_function :rd_kafka_conf_set_dr_msg_cb, [:pointer, :delivery_cb], :void
 

lib/rdkafka/producer.rb

@@ -105,6 +105,32 @@ def flush(timeout_ms=5_000)
       raise(error)
     end
 
+    # Purges the outgoing queue and releases all resources.
+    #
+    # Useful when closing the producer with outgoing messages to unstable clusters or when for
+    # any other reasons waiting cannot go on anymore. This purges both the queue and all the
+    # inflight requests + updates the delivery handles statuses so they can be materialized into
+    # `purge_queue` errors.
+    def purge
+      closed_producer_check(__method__)
+
+      code = nil
+
+      @native_kafka.with_inner do |inner|
+        code = Bindings.rd_kafka_purge(
+          inner,
+          Bindings::RD_KAFKA_PURGE_F_QUEUE | Bindings::RD_KAFKA_PURGE_F_INFLIGHT
+        )
+      end
+
+      code.zero? || raise(Rdkafka::RdkafkaError.new(code))
+
+      # Wait for the purge to affect everything
+      sleep(0.001) until flush(100)
+
+      true
+    end
+
     # Partition count for a given topic.
     #
     # @param topic [String] The topic name.

spec/rdkafka/producer_spec.rb

@@ -645,4 +645,69 @@ def call(_, handle)
       end
     end
   end
+
+  describe '#purge' do
+    context 'when no outgoing messages' do
+      it { expect(producer.purge).to eq(true) }
+    end
+
+    context 'when librdkafka purge returns an error' do
+      before { expect(Rdkafka::Bindings).to receive(:rd_kafka_purge).and_return(-153) }
+
+      it 'expect to raise an error' do
+        expect { producer.purge }.to raise_error(Rdkafka::RdkafkaError, /retry/)
+      end
+    end
+
+    context 'when there are outgoing things in the queue' do
+      let(:producer) do
+        rdkafka_producer_config(
+          "bootstrap.servers": "localhost:9093",
+          "message.timeout.ms": 2_000
+        ).producer
+      end
+
+      it "should should purge and move forward" do
+        producer.produce(
+          topic:     "produce_test_topic",
+          payload:   "payload headers"
+        )
+
+        expect(producer.purge).to eq(true)
+        expect(producer.flush(1_000)).to eq(true)
+      end
+
+      it "should materialize the delivery handles" do
+        handle = producer.produce(
+          topic:     "produce_test_topic",
+          payload:   "payload headers"
+        )
+
+        expect(producer.purge).to eq(true)
+
+        expect { handle.wait }.to raise_error(Rdkafka::RdkafkaError, /purge_queue/)
+      end
+
+      context "when using delivery_callback" do
+        let(:delivery_reports) { [] }
+
+        let(:delivery_callback) do
+          ->(delivery_report) { delivery_reports << delivery_report }
+        end
+
+        before { producer.delivery_callback = delivery_callback }
+
+        it "should run the callback" do
+          handle = producer.produce(
+            topic:     "produce_test_topic",
+            payload:   "payload headers"
+          )
+
+          expect(producer.purge).to eq(true)
+          # queue purge
+          expect(delivery_reports[0].error).to eq(-152)
+        end
+      end
+    end
+  end
 end