Expose alternative poll flow for consumer (#356)

Author: mensfeld

CHANGELOG.md

@@ -5,8 +5,10 @@
 - **[Feature]** Add `Admin#create_partitions` (mensfeld)
 - **[Feature]** Add `Admin#delete_group` utility (piotaixr)
 - **[Feature]** Add Create and Delete ACL Feature To Admin Functions (vgnanasekaran)
+- [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)
+- [Change] Use `SecureRandom.uuid` instead of `random` for test consumer groups (mensfeld)
 
 ## 0.14.0 (2023-11-21)
 - [Enhancement] Add `raise_response_error` flag to the `Rdkafka::AbstractHandle`.

lib/rdkafka/config.rb

@@ -112,6 +112,7 @@ def self.opaques
     def initialize(config_hash = {})
       @config_hash = DEFAULT_CONFIG.merge(config_hash)
       @consumer_rebalance_listener = nil
+      @consumer_poll_set = true
     end
 
     # Set a config option.
@@ -140,6 +141,22 @@ def consumer_rebalance_listener=(listener)
       @consumer_rebalance_listener = listener
     end
 
+    # Should we use a single queue for the underlying consumer and events.
+    #
+    # This is an advanced API that allows for more granular control of the polling process.
+    # When this value is set to `false` (`true` by defualt), there will be two queues that need to
+    # be polled:
+    #   - main librdkafka queue for events
+    #   - consumer queue with messages and rebalances
+    #
+    # It is recommended to use the defaults and only set it to `false` in advance multi-threaded
+    # and complex cases where granular events handling control is needed.
+    #
+    # @param poll_set [Boolean]
+    def consumer_poll_set=(poll_set)
+      @consumer_poll_set = poll_set
+    end
+
     # Creates a consumer with this configuration.
     #
     # @return [Consumer] The created consumer
@@ -158,8 +175,8 @@ def consumer
       # Create native client
       kafka = native_kafka(config, :rd_kafka_consumer)
 
-      # Redirect the main queue to the consumer
-      Rdkafka::Bindings.rd_kafka_poll_set_consumer(kafka)
+      # Redirect the main queue to the consumer queue
+      Rdkafka::Bindings.rd_kafka_poll_set_consumer(kafka) if @consumer_poll_set
 
       # Return consumer with Kafka client
       Rdkafka::Consumer.new(
@@ -187,7 +204,11 @@ def producer
       # Return producer with Kafka client
       partitioner_name = self[:partitioner] || self["partitioner"]
       Rdkafka::Producer.new(
-        Rdkafka::NativeKafka.new(native_kafka(config, :rd_kafka_producer), run_polling_thread: true, opaque: opaque),
+        Rdkafka::NativeKafka.new(
+          native_kafka(config, :rd_kafka_producer),
+          run_polling_thread: true,
+          opaque: opaque
+        ),
         partitioner_name
       ).tap do |producer|
         opaque.producer = producer

lib/rdkafka/consumer.rb

@@ -531,6 +531,32 @@ def poll(timeout_ms)
       end
     end
 
+    # Polls the main rdkafka queue (not the consumer one). Do **NOT** use it if `consumer_poll_set`
+    #   was set to `true`.
+    #
+    # Events will cause application-provided callbacks to be called.
+    #
+    # Events (in the context of the consumer):
+    #   - error callbacks
+    #   - stats callbacks
+    #   - any other callbacks supported by librdkafka that are not part of the consumer_poll, that
+    #     would have a callback configured and activated.
+    #
+    # This method needs to be called at regular intervals to serve any queued callbacks waiting to
+    # be called. When in use, does **NOT** replace `#poll` but needs to run complementary with it.
+    #
+    # @param timeout_ms [Integer] poll timeout. If set to 0 will run async, when set to -1 will
+    #   block until any events available.
+    #
+    # @note This method technically should be called `#poll` and the current `#poll` should be
+    #   called `#consumer_poll` though we keep the current naming convention to make it backward
+    #   compatible.
+    def events_poll(timeout_ms = 0)
+      @native_kafka.with_inner do |inner|
+        Rdkafka::Bindings.rd_kafka_poll(inner, timeout_ms)
+      end
+    end
+
     # Poll for new messages and yield for each received one. Iteration
     # will end when the consumer is closed.
     #

spec/rdkafka/config_spec.rb

@@ -113,6 +113,14 @@ def call(stats); end
       consumer.close
     end
 
+    it "should create a consumer with consumer_poll_set set to false" do
+      config = rdkafka_consumer_config
+      config.consumer_poll_set = false
+      consumer = config.consumer
+      expect(consumer).to be_a Rdkafka::Consumer
+      consumer.close
+    end
+
     it "should raise an error when creating a consumer with invalid config" do
       config = Rdkafka::Config.new('invalid.key' => 'value')
       expect {

spec/rdkafka/consumer_spec.rb

@@ -54,6 +54,30 @@
         consumer.subscription
       }.to raise_error(Rdkafka::RdkafkaError)
     end
+
+    context "when using consumer without the poll set" do
+      let(:consumer) do
+        config = rdkafka_consumer_config
+        config.consumer_poll_set = false
+        config.consumer
+      end
+
+      it "should subscribe, unsubscribe and return the subscription" do
+        expect(consumer.subscription).to be_empty
+
+        consumer.subscribe("consume_test_topic")
+
+        expect(consumer.subscription).not_to be_empty
+        expected_subscription = Rdkafka::Consumer::TopicPartitionList.new.tap do |list|
+          list.add_topic("consume_test_topic")
+        end
+        expect(consumer.subscription).to eq expected_subscription
+
+        consumer.unsubscribe
+
+        expect(consumer.subscription).to be_empty
+      end
+    end
   end
 
   describe "#pause and #resume" do
@@ -1054,6 +1078,29 @@ def send_one_message(val)
     end
   end
 
+  # Only relevant in case of a consumer with separate queues
+  describe '#events_poll' do
+    let(:stats) { [] }
+
+    before { Rdkafka::Config.statistics_callback = ->(published) { stats << published } }
+
+    after { Rdkafka::Config.statistics_callback = nil }
+
+    let(:consumer) do
+      config = rdkafka_consumer_config('statistics.interval.ms': 100)
+      config.consumer_poll_set = false
+      config.consumer
+    end
+
+    it "expect to run events_poll, operate and propagate stats on events_poll and not poll" do
+      consumer.subscribe("consume_test_topic")
+      consumer.poll(1_000)
+      expect(stats).to be_empty
+      consumer.events_poll(-1)
+      expect(stats).not_to be_empty
+    end
+  end
+
   describe "a rebalance listener" do
     let(:consumer) do
       config = rdkafka_consumer_config

spec/spec_helper.rb

@@ -36,7 +36,7 @@ def rdkafka_consumer_config(config_overrides={})
   # Add consumer specific fields to it
   config[:"auto.offset.reset"] = "earliest"
   config[:"enable.partition.eof"] = false
-  config[:"group.id"] = "ruby-test-#{Random.new.rand(0..1_000_000)}"
+  config[:"group.id"] = "ruby-test-#{SecureRandom.uuid}"
   # Enable debug mode if required
   if ENV["DEBUG_CONSUMER"]
     config[:debug] = "cgrp,topic,fetch"
@@ -135,6 +135,7 @@ def notify_listener(listener, &block)
         rake_test_topic:         3,
         watermarks_test_topic:   3,
         partitioner_test_topic: 25,
+        example_topic:           1
     }.each do |topic, partitions|
       create_topic_handle = admin.create_topic(topic.to_s, partitions, 1)
       begin