support assignment lost (#358)

Author: mensfeld

CHANGELOG.md

@@ -5,6 +5,7 @@
 - **[Feature]** Add `Admin#create_partitions` (mensfeld)
 - **[Feature]** Add `Admin#delete_group` utility (piotaixr)
 - **[Feature]** Add Create and Delete ACL Feature To Admin Functions (vgnanasekaran)
+- **[Feature]** Support `#assignment_lost?` on a consumer to check for involuntary assignment revocation (mensfeld)
 - [Enhancement] Expose alternative way of managing consumer events via a separate queue (mensfeld) 
 - [Enhancement] Bump librdkafka to 2.3.0 (mensfeld)
 - [Enhancement] Increase the `#lag` and `#query_watermark_offsets` default timeouts from 100ms to 1000ms. This will compensate for network glitches and remote clusters operations (mensfeld)

lib/rdkafka/bindings.rb

@@ -179,6 +179,7 @@ class TopicPartitionList < FFI::Struct
     attach_function :rd_kafka_incremental_assign, [:pointer, :pointer], :int, blocking: true
     attach_function :rd_kafka_incremental_unassign, [:pointer, :pointer], :int, blocking: true
     attach_function :rd_kafka_assignment, [:pointer, :pointer], :int, blocking: true
+    attach_function :rd_kafka_assignment_lost, [:pointer], :int, blocking: true
     attach_function :rd_kafka_committed, [:pointer, :pointer, :int], :int, blocking: true
     attach_function :rd_kafka_commit, [:pointer, :pointer, :bool], :int, blocking: true
     attach_function :rd_kafka_poll_set_consumer, [:pointer], :void, blocking: true

lib/rdkafka/consumer.rb

@@ -221,6 +221,15 @@ def assignment
       ptr.free unless ptr.nil?
     end
 
+    # @return [Boolean] true if our current assignment has been lost involuntarily.
+    def assignment_lost?
+      closed_consumer_check(__method__)
+
+      @native_kafka.with_inner do |inner|
+        !Rdkafka::Bindings.rd_kafka_assignment_lost(inner).zero?
+      end
+    end
+
     # Return the current committed offset per partition for this consumer group.
     # The offset field of each requested partition will either be set to stored offset or to -1001
     # in case there was no stored offset for that partition.

spec/rdkafka/consumer_spec.rb

@@ -297,6 +297,28 @@ def send_one_message(val)
     end
   end
 
+  describe '#assignment_lost?' do
+    it "should not return true as we do have an assignment" do
+      consumer.subscribe("consume_test_topic")
+      expected_subscription = Rdkafka::Consumer::TopicPartitionList.new.tap do |list|
+        list.add_topic("consume_test_topic")
+      end
+
+      expect(consumer.assignment_lost?).to eq false
+      consumer.unsubscribe
+    end
+
+    it "should not return true after voluntary unsubscribing" do
+      consumer.subscribe("consume_test_topic")
+      expected_subscription = Rdkafka::Consumer::TopicPartitionList.new.tap do |list|
+        list.add_topic("consume_test_topic")
+      end
+
+      consumer.unsubscribe
+      expect(consumer.assignment_lost?).to eq false
+    end
+  end
+
   describe "#close" do
     it "should close a consumer" do
       consumer.subscribe("consume_test_topic")