Class: Rdkafka::Consumer

Inherits:
Object
  • Object
show all
Includes:
Enumerable, Helpers::OAuth, Helpers::Time
Defined in:
lib/rdkafka/consumer.rb,
lib/rdkafka/consumer/headers.rb,
lib/rdkafka/consumer/message.rb,
lib/rdkafka/consumer/partition.rb,
lib/rdkafka/consumer/topic_partition_list.rb

Overview

A consumer of Kafka messages. It uses the high-level consumer approach where the Kafka brokers automatically assign partitions and load balance partitions over consumers that have the same ‘:“group.id”` set in their configuration.

To create a consumer set up a Config and call consumer on that. It is mandatory to set ‘:“group.id”` in the configuration.

Consumer implements ‘Enumerable`, so you can use `each` to consume messages, or for example `each_slice` to consume batches of messages.

Defined Under Namespace

Modules: Headers Classes: Message, Partition, TopicPartitionList

Instance Method Summary collapse

Methods included from Helpers::OAuth

#oauthbearer_set_token, #oauthbearer_set_token_failure

Methods included from Helpers::Time

#monotonic_now, #monotonic_now_ms

Constructor Details

#initialize(native_kafka) ⇒ Consumer

Returns a new instance of Consumer.

Parameters:

  • native_kafka (NativeKafka) —

    wrapper around the native Kafka consumer handle



20
21
22
 
# File 'lib/rdkafka/consumer.rb', line 20

def initialize(native_kafka)
  @native_kafka = native_kafka
end

Instance Method Details

#assign(list) ⇒ Object

Atomic assignment of partitions to consume

Parameters:

Raises:



221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'lib/rdkafka/consumer.rb', line 221

def assign(list)
  closed_consumer_check(__method__)

  unless list.is_a?(TopicPartitionList)
    raise TypeError.new("list has to be a TopicPartitionList")
  end

  tpl = list.to_native_tpl

  begin
    @native_kafka.with_inner do |inner|
      response = Rdkafka::Bindings.rd_kafka_assign(inner, tpl)
      Rdkafka::RdkafkaError.validate!(response, "Error assigning '#{list.to_h}'", client_ptr: inner)
    end
  ensure
    Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(tpl)
  end
end

#assignment ⇒ TopicPartitionList

Returns the current partition assignment.

Returns:

Raises:



244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
 
# File 'lib/rdkafka/consumer.rb', line 244

def assignment
  closed_consumer_check(__method__)

  ptr = FFI::MemoryPointer.new(:pointer)
  @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_assignment(inner, ptr)
    Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
  end

  tpl = ptr.read_pointer

  if !tpl.null?
    begin
      Rdkafka::Consumer::TopicPartitionList.from_native_tpl(tpl)
    ensure
      Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy tpl
    end
  end
ensure
  ptr&.free
end

#assignment_lost? ⇒ Boolean

Returns true if our current assignment has been lost involuntarily.

Returns:

  • (Boolean) —

    true if our current assignment has been lost involuntarily.



267
268
269
270
271
272
273
 
# File 'lib/rdkafka/consumer.rb', line 267

def assignment_lost?
  closed_consumer_check(__method__)

  @native_kafka.with_inner do |inner|
    !Rdkafka::Bindings.rd_kafka_assignment_lost(inner).zero?
  end
end

#close ⇒ nil

Close this consumer

Returns:

  • (nil) 


86
87
88
89
90
91
92
93
94
95
 
# File 'lib/rdkafka/consumer.rb', line 86

def close
  return if closed?
  ObjectSpace.undefine_finalizer(self)

  @native_kafka.synchronize do |inner|
    Rdkafka::Bindings.rd_kafka_consumer_close(inner)
  end

  @native_kafka.close
end

#closed? ⇒ Boolean

Whether this consumer has closed

Returns:

  • (Boolean) 


98
99
100
 
# File 'lib/rdkafka/consumer.rb', line 98

def closed?
  @native_kafka.closed?
end

#cluster_id ⇒ String?

Returns the ClusterId as reported in broker metadata.

Returns:

  • (String, nil) 


397
398
399
400
401
402
 
# File 'lib/rdkafka/consumer.rb', line 397

def cluster_id
  closed_consumer_check(__method__)
  @native_kafka.with_inner do |inner|
    Rdkafka::Bindings.rd_kafka_clusterid(inner)
  end
end

#commit(list = nil, async = false) ⇒ nil

Manually commit the current offsets of this consumer.

To use this set ‘enable.auto.commit`to `false` to disable automatic triggering of commits.

If ‘enable.auto.offset.store` is set to `true` the offset of the last consumed message for every partition is used. If set to `false` you can use #store_offset to indicate when a message has been fully processed.

Parameters:

  • list (TopicPartitionList, nil) (defaults to: nil) —

    The topic with partitions to commit

  • async (Boolean) (defaults to: false) —

    Whether to commit async or wait for the commit to finish

Returns:

  • (nil)

Raises:



557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
 
# File 'lib/rdkafka/consumer.rb', line 557

def commit(list = nil, async = false)
  closed_consumer_check(__method__)

  if !list.nil? && !list.is_a?(TopicPartitionList)
    raise TypeError.new("list has to be nil or a TopicPartitionList")
  end

  tpl = list&.to_native_tpl

  begin
    @native_kafka.with_inner do |inner|
      response = Rdkafka::Bindings.rd_kafka_commit(inner, tpl, async)
      Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
    end

    nil
  ensure
    Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(tpl) if tpl
  end
end

#committed(list = nil, timeout_ms = Defaults::CONSUMER_COMMITTED_TIMEOUT_MS) ⇒ TopicPartitionList

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.

Parameters:

  • list (TopicPartitionList, nil) (defaults to: nil) —

    The topic with partitions to get the offsets for or nil to use the current subscription.

  • timeout_ms (Integer) (defaults to: Defaults::CONSUMER_COMMITTED_TIMEOUT_MS) —

    The timeout for fetching this information.

Returns:

Raises:

  • (RdkafkaError) —

    When getting the committed positions fails.



284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
 
# File 'lib/rdkafka/consumer.rb', line 284

def committed(list = nil, timeout_ms = Defaults::CONSUMER_COMMITTED_TIMEOUT_MS)
  closed_consumer_check(__method__)

  if list.nil?
    list = assignment
  elsif !list.is_a?(TopicPartitionList)
    raise TypeError.new("list has to be nil or a TopicPartitionList")
  end

  tpl = list.to_native_tpl

  begin
    @native_kafka.with_inner do |inner|
      response = Rdkafka::Bindings.rd_kafka_committed(inner, tpl, timeout_ms)
      Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
    end

    TopicPartitionList.from_native_tpl(tpl)
  ensure
    Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(tpl)
  end
end

#consumer_group_metadata_pointer ⇒ Object

Note:

This pointer needs to be removed with ‘#rd_kafka_consumer_group_metadata_destroy`

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



748
749
750
751
752
753
754
 
# File 'lib/rdkafka/consumer.rb', line 748

def consumer_group_metadata_pointer
  closed_consumer_check(__method__)

  @native_kafka.with_inner do |inner|
    Bindings.rd_kafka_consumer_group_metadata(inner)
  end
end

#each(timeout_ms: Defaults::CONSUMER_POLL_TIMEOUT_MS) {|message| ... } ⇒ nil

Poll for new messages and yield for each received one. Iteration will end when the consumer is closed.

If ‘enable.partition.eof` is turned on in the config this will raise an error when an eof is reached, so you probably want to disable that when using this method of iteration.

Parameters:

  • timeout_ms (Integer) (defaults to: Defaults::CONSUMER_POLL_TIMEOUT_MS) —

    The timeout for each poll

Yield Parameters:

  • message (Message) —

    Received message

Returns:

  • (nil)

Raises:



702
703
704
705
706
707
708
709
710
711
712
713
 
# File 'lib/rdkafka/consumer.rb', line 702

def each(timeout_ms: Defaults::CONSUMER_POLL_TIMEOUT_MS)
  loop do
    message = poll(timeout_ms)
    if message
      yield(message)
    elsif closed?
      break
    else
      next
    end
  end
end

#each_batch(max_items: 100, bytes_threshold: Float::INFINITY, timeout_ms: 250, yield_on_error: false, &block) ⇒ Object

Deprecated.

This method has been removed due to data consistency concerns

Parameters:

  • max_items (Integer) (defaults to: 100) —

    unused

  • bytes_threshold (Numeric) (defaults to: Float::INFINITY) —

    unused

  • timeout_ms (Integer) (defaults to: 250) —

    unused

  • yield_on_error (Boolean) (defaults to: false) —

    unused

  • block (Proc) —

    unused block

Raises:

  • (NotImplementedError) —

    Always raises as this method is no longer supported



722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
 
# File 'lib/rdkafka/consumer.rb', line 722

def each_batch(max_items: 100, bytes_threshold: Float::INFINITY, timeout_ms: 250, yield_on_error: false, &block)
  raise NotImplementedError, <<~ERROR
    `each_batch` has been removed due to data consistency concerns.

    This method was removed because it did not properly handle partition reassignments,
    which could lead to processing messages from partitions that were no longer owned
    by this consumer, resulting in duplicate message processing and data inconsistencies.

    Recommended alternatives:

    1. Implement your own batching logic using rebalance callbacks to properly handle
       partition revocations and ensure message processing correctness.

    2. Use a high-level batching library that supports proper partition reassignment
       handling out of the box (such as the Karafka framework).
  ERROR
end

#enable_background_queue_io_events(fd, payload = "\x01") ⇒ nil

Enable IO event notifications for background events

Parameters:

  • fd (Integer) —

    file descriptor to signal (from IO.pipe or eventfd)

  • payload (String) (defaults to: "\x01") —

    data to write to fd (default: “x01”)

Returns:

  • (nil)

Raises:



74
75
76
 
# File 'lib/rdkafka/consumer.rb', line 74

def enable_background_queue_io_events(fd, payload = "\x01")
  @native_kafka.enable_background_queue_io_events(fd, payload)
end

#enable_queue_io_events(fd, payload = "\x01") ⇒ nil

Enable IO event notifications for fiber scheduler integration When the consumer queue has messages, librdkafka will write to your FD

Examples:

Using with fiber scheduler

consumer = config.consumer
consumer.subscribe("topic")

# Create notification FD
signal_r, signal_w = IO.pipe

# Enable librdkafka to signal when messages arrive
consumer.enable_queue_io_events(signal_w.fileno)

# Monitor with select/poll
loop do
  readable, = IO.select([signal_r], nil, nil, timeout)
  if readable
    signal_r.read_nonblock(1024) rescue nil  # Drain signal
    while msg = consumer.poll(0)
      process(msg)
    end
  end
end

Parameters:

  • fd (Integer) —

    file descriptor to signal (from IO.pipe or eventfd)

  • payload (String) (defaults to: "\x01") —

    data to write to fd (default: “x01”)

Returns:

  • (nil)

Raises:



65
66
67
 
# File 'lib/rdkafka/consumer.rb', line 65

def enable_queue_io_events(fd, payload = "\x01")
  @native_kafka.enable_main_queue_io_events(fd, payload)
end

#events_poll(timeout_ms = Defaults::CONSUMER_EVENTS_POLL_TIMEOUT_MS) ⇒ Object

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.

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.

Parameters:

  • timeout_ms (Integer) (defaults to: Defaults::CONSUMER_EVENTS_POLL_TIMEOUT_MS) —

    poll timeout. If set to 0 will run async, when set to -1 will block until any events available.



670
671
672
673
674
 
# File 'lib/rdkafka/consumer.rb', line 670

def events_poll(timeout_ms = Defaults::CONSUMER_EVENTS_POLL_TIMEOUT_MS)
  @native_kafka.with_inner do |inner|
    Rdkafka::Bindings.rd_kafka_poll(inner, timeout_ms)
  end
end

#events_poll_nb(timeout_ms = 0) ⇒ Integer

Polls the main rdkafka queue without releasing the GVL (Global VM Lock).

This is more efficient than regular events_poll for non-blocking poll(0) calls, particularly useful in fiber scheduler contexts where GVL release/reacquire overhead is wasteful since we don’t expect to wait.

Parameters:

  • timeout_ms (Integer) (defaults to: 0) —

    poll timeout (default: 0 for non-blocking)

Returns:

  • (Integer) —

    the number of events served

See Also:



686
687
688
689
690
 
# File 'lib/rdkafka/consumer.rb', line 686

def events_poll_nb(timeout_ms = 0)
  @native_kafka.with_inner do |inner|
    Rdkafka::Bindings.rd_kafka_poll_nb(inner, timeout_ms)
  end
end

#finalizer ⇒ Proc

Returns finalizer proc for closing the consumer.

Returns:

  • (Proc) —

    finalizer proc for closing the consumer



80
81
82
 
# File 'lib/rdkafka/consumer.rb', line 80

def finalizer
  ->(_) { close }
end

#lag(topic_partition_list, watermark_timeout_ms = Defaults::CONSUMER_LAG_TIMEOUT_MS) ⇒ Hash{String => Hash{Integer => Integer}}

Calculate the consumer lag per partition for the provided topic partition list. You can get a suitable list by calling #committed or #position (TODO). It is also possible to create one yourself, in this case you have to provide a list that already contains all the partitions you need the lag for.

Parameters:

  • topic_partition_list (TopicPartitionList) —

    The list to calculate lag for.

  • watermark_timeout_ms (Integer) (defaults to: Defaults::CONSUMER_LAG_TIMEOUT_MS) —

    The timeout for each query watermark call.

Returns:

  • (Hash{String => Hash{Integer => Integer}}) —

    A hash containing all topics with the lag per partition

Raises:



373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
 
# File 'lib/rdkafka/consumer.rb', line 373

def lag(topic_partition_list, watermark_timeout_ms = Defaults::CONSUMER_LAG_TIMEOUT_MS)
  out = {}

  topic_partition_list.to_h.each do |topic, partitions|
    # Query high watermarks for this topic's partitions
    # and compare to the offset in the list.
    topic_out = {}
    partitions.each do |p|
      next if p.offset.nil?
      _low, high = query_watermark_offsets(
        topic,
        p.partition,
        watermark_timeout_ms
      )
      topic_out[p.partition] = high - p.offset
    end
    out[topic] = topic_out
  end
  out
end

#member_id ⇒ String?

Returns this client’s broker-assigned group member id

This currently requires the high-level KafkaConsumer

Returns:

  • (String, nil) 


409
410
411
412
413
414
 
# File 'lib/rdkafka/consumer.rb', line 409

def member_id
  closed_consumer_check(__method__)
  @native_kafka.with_inner do |inner|
    Rdkafka::Bindings.rd_kafka_memberid(inner)
  end
end

#name ⇒ String

Returns consumer name.

Returns:

  • (String) —

    consumer name



31
32
33
34
35
 
# File 'lib/rdkafka/consumer.rb', line 31

def name
  @name ||= @native_kafka.with_inner do |inner|
    ::Rdkafka::Bindings.rd_kafka_name(inner)
  end
end

#offsets_for_times(list, timeout_ms = Defaults::CONSUMER_OFFSETS_FOR_TIMES_TIMEOUT_MS) ⇒ TopicPartitionList

Lookup offset for the given partitions by timestamp.

Parameters:

  • list (TopicPartitionList) —

    The TopicPartitionList with timestamps instead of offsets

  • timeout_ms (Integer) (defaults to: Defaults::CONSUMER_OFFSETS_FOR_TIMES_TIMEOUT_MS) —

    timeout in milliseconds for the operation

Returns:

Raises:

  • (RdKafkaError) —

    When the OffsetForTimes lookup fails



521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
 
# File 'lib/rdkafka/consumer.rb', line 521

def offsets_for_times(list, timeout_ms = Defaults::CONSUMER_OFFSETS_FOR_TIMES_TIMEOUT_MS)
  closed_consumer_check(__method__)

  if !list.is_a?(TopicPartitionList)
    raise TypeError.new("list has to be a TopicPartitionList")
  end

  tpl = list.to_native_tpl

  @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_offsets_for_times(
      inner,
      tpl,
      timeout_ms # timeout
    )
    Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
  end

  TopicPartitionList.from_native_tpl(tpl)
ensure
  Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(tpl) if tpl
end

#pause(list) ⇒ nil

Pause producing or consumption for the provided list of partitions

Parameters:

Returns:

  • (nil)

Raises:



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/rdkafka/consumer.rb', line 146

def pause(list)
  closed_consumer_check(__method__)

  unless list.is_a?(TopicPartitionList)
    raise TypeError.new("list has to be a TopicPartitionList")
  end

  tpl = list.to_native_tpl

  begin
    response = @native_kafka.with_inner do |inner|
      Rdkafka::Bindings.rd_kafka_pause_partitions(inner, tpl)
    end

    if response != Rdkafka::Bindings::RD_KAFKA_RESP_ERR_NO_ERROR
      list = TopicPartitionList.from_native_tpl(tpl)
      raise Rdkafka::RdkafkaTopicPartitionListError.new(response, list, "Error pausing '#{list.to_h}'")
    end
  ensure
    Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(tpl)
  end
end

#poll(timeout_ms) ⇒ Message?

Poll for the next message on one of the subscribed topics

Parameters:

  • timeout_ms (Integer) —

    Timeout of this poll

Returns:

  • (Message, nil) —

    A message or nil if there was no new message within the timeout

Raises:



583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
 
# File 'lib/rdkafka/consumer.rb', line 583

def poll(timeout_ms)
  closed_consumer_check(__method__)

  message_ptr = @native_kafka.with_inner do |inner|
    Rdkafka::Bindings.rd_kafka_consumer_poll(inner, timeout_ms)
  end

  return nil if message_ptr.null?

  # Create struct wrapper
  native_message = Rdkafka::Bindings::Message.new(message_ptr)

  # Create a message to pass out
  return Rdkafka::Consumer::Message.new(native_message) if native_message[:err] == Rdkafka::Bindings::RD_KAFKA_RESP_ERR_NO_ERROR

  # Raise error if needed - wrap just the validate! call with client access
  @native_kafka.with_inner do |inner|
    Rdkafka::RdkafkaError.validate!(native_message, client_ptr: inner)
  end
ensure
  # Clean up rdkafka message if there is one
  if message_ptr && !message_ptr.null?
    Rdkafka::Bindings.rd_kafka_message_destroy(message_ptr)
  end
end

#poll_nb(timeout_ms = 0) ⇒ Message?

Poll for the next message without releasing the GVL (Global VM Lock).

This is more efficient than regular polling for non-blocking poll(0) calls, particularly useful in fiber scheduler contexts where GVL release/reacquire overhead is wasteful since we don’t expect to wait.

Examples:

Using with fiber scheduler

# After receiving IO notification that messages are available
while msg = consumer.poll_nb
  process(msg)
end

Parameters:

  • timeout_ms (Integer) (defaults to: 0) —

    Timeout of this poll (default: 0 for non-blocking)

Returns:

  • (Message, nil) —

    A message or nil if there was no new message within the timeout

Raises:



624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
 
# File 'lib/rdkafka/consumer.rb', line 624

def poll_nb(timeout_ms = 0)
  closed_consumer_check(__method__)

  message_ptr = @native_kafka.with_inner do |inner|
    Rdkafka::Bindings.rd_kafka_consumer_poll_nb(inner, timeout_ms)
  end

  if message_ptr.null?
    nil
  else
    # Create struct wrapper
    native_message = Rdkafka::Bindings::Message.new(message_ptr)
    # Raise error if needed
    if native_message[:err] != Rdkafka::Bindings::RD_KAFKA_RESP_ERR_NO_ERROR
      raise Rdkafka::RdkafkaError.new(native_message[:err])
    end
    # Create a message to pass out
    Rdkafka::Consumer::Message.new(native_message)
  end
ensure
  # Clean up rdkafka message if there is one
  if message_ptr && !message_ptr.null?
    Rdkafka::Bindings.rd_kafka_message_destroy(message_ptr)
  end
end

#position(list = nil) ⇒ TopicPartitionList

Return the current positions (offsets) for topics and partitions. The offset field of each requested partition will be set to the offset of the last consumed message + 1, or nil in case there was no previous message.

Parameters:

  • list (TopicPartitionList, nil) (defaults to: nil) —

    The topic with partitions to get the offsets for or nil to use the current subscription.

Returns:

Raises:



315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
 
# File 'lib/rdkafka/consumer.rb', line 315

def position(list = nil)
  if list.nil?
    list = assignment
  elsif !list.is_a?(TopicPartitionList)
    raise TypeError.new("list has to be nil or a TopicPartitionList")
  end

  tpl = list.to_native_tpl

  @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_position(inner, tpl)
    Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
  end

  TopicPartitionList.from_native_tpl(tpl)
end

#query_watermark_offsets(topic, partition, timeout_ms = Defaults::CONSUMER_QUERY_WATERMARK_TIMEOUT_MS) ⇒ Integer

Query broker for low (oldest/beginning) and high (newest/end) offsets for a partition.

Parameters:

  • topic (String) —

    The topic to query

  • partition (Integer) —

    The partition to query

  • timeout_ms (Integer) (defaults to: Defaults::CONSUMER_QUERY_WATERMARK_TIMEOUT_MS) —

    The timeout for querying the broker

Returns:

  • (Integer) —

    The low and high watermark

Raises:



339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
 
# File 'lib/rdkafka/consumer.rb', line 339

def query_watermark_offsets(topic, partition, timeout_ms = Defaults::CONSUMER_QUERY_WATERMARK_TIMEOUT_MS)
  closed_consumer_check(__method__)

  low = FFI::MemoryPointer.new(:int64, 1)
  high = FFI::MemoryPointer.new(:int64, 1)

  @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_query_watermark_offsets(
      inner,
      topic,
      partition,
      low,
      high,
      timeout_ms
    )
    Rdkafka::RdkafkaError.validate!(response, "Error querying watermark offsets for partition #{partition} of #{topic}", client_ptr: inner)
  end

  [low.read_array_of_int64(1).first, high.read_array_of_int64(1).first]
ensure
  low&.free
  high&.free
end

#resume(list) ⇒ nil

Resumes producing consumption for the provided list of partitions

Parameters:

Returns:

  • (nil)

Raises:



174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/rdkafka/consumer.rb', line 174

def resume(list)
  closed_consumer_check(__method__)

  unless list.is_a?(TopicPartitionList)
    raise TypeError.new("list has to be a TopicPartitionList")
  end

  tpl = list.to_native_tpl

  begin
    @native_kafka.with_inner do |inner|
      response = Rdkafka::Bindings.rd_kafka_resume_partitions(inner, tpl)
      Rdkafka::RdkafkaError.validate!(response, "Error resume '#{list.to_h}'", client_ptr: inner)
    end

    nil
  ensure
    Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(tpl)
  end
end

#seek(message) ⇒ nil

Seek to a particular message. The next poll on the topic/partition will return the message at the given offset.

Parameters:

Returns:

  • (nil)

Raises:



471
472
473
 
# File 'lib/rdkafka/consumer.rb', line 471

def seek(message)
  seek_by(message.topic, message.partition, message.offset)
end

#seek_by(topic, partition, offset) ⇒ nil

Seek to a particular message by providing the topic, partition and offset. The next poll on the topic/partition will return the message at the given offset.

Parameters:

  • topic (String) —

    The topic in which to seek

  • partition (Integer) —

    The partition number to seek

  • offset (Integer) —

    The partition offset to seek

Returns:

  • (nil)

Raises:



484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
 
# File 'lib/rdkafka/consumer.rb', line 484

def seek_by(topic, partition, offset)
  closed_consumer_check(__method__)

  # rd_kafka_offset_store is one of the few calls that does not support
  # a string as the topic, so create a native topic for it.
  native_topic = @native_kafka.with_inner do |inner|
    Rdkafka::Bindings.rd_kafka_topic_new(
      inner,
      topic,
      nil
    )
  end

  response = Rdkafka::Bindings.rd_kafka_seek(
    native_topic,
    partition,
    offset,
    Defaults::CONSUMER_SEEK_TIMEOUT_MS
  )

  return nil if response == Rdkafka::Bindings::RD_KAFKA_RESP_ERR_NO_ERROR

  @native_kafka.with_inner do |inner|
    Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
  end
ensure
  if native_topic && !native_topic.null?
    Rdkafka::Bindings.rd_kafka_topic_destroy(native_topic)
  end
end

#start ⇒ Object

Note:

Not needed to run unless explicit start was disabled

Starts the native Kafka polling thread and kicks off the init polling



26
27
28
 
# File 'lib/rdkafka/consumer.rb', line 26

def start
  @native_kafka.start
end

#store_offset(message, metadata = nil) ⇒ nil

Store offset of a message to be used in the next commit of this consumer

When using this ‘enable.auto.offset.store` should be set to `false` in the config.

Parameters:

  • message (Rdkafka::Consumer::Message) —

    The message which offset will be stored

  • metadata (String, nil) (defaults to: nil) —

    commit metadata string or nil if none

Returns:

  • (nil)

Raises:



424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
 
# File 'lib/rdkafka/consumer.rb', line 424

def store_offset(message, metadata = nil)
  closed_consumer_check(__method__)

  list = TopicPartitionList.new

  # For metadata aware commits we build the partition reference directly to save on
  # objects allocations
  if metadata
    list.add_topic_and_partitions_with_offsets(
      message.topic,
      [
        Consumer::Partition.new(
          message.partition,
          message.offset + 1,
          0,
          metadata
        )
      ]
    )
  else
    list.add_topic_and_partitions_with_offsets(
      message.topic,
      message.partition => message.offset + 1
    )
  end

  tpl = list.to_native_tpl

  @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_offsets_store(
      inner,
      tpl
    )
    Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
  end

  nil
ensure
  Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(tpl) if tpl
end

#subscribe(*topics) ⇒ nil

Subscribes to one or more topics letting Kafka handle partition assignments.

Parameters:

  • topics (Array<String>) —

    One or more topic names

Returns:

  • (nil)

Raises:



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/rdkafka/consumer.rb', line 107

def subscribe(*topics)
  closed_consumer_check(__method__)

  # Create topic partition list with topics and no partition set
  tpl = Rdkafka::Bindings.rd_kafka_topic_partition_list_new(topics.length)

  topics.each do |topic|
    Rdkafka::Bindings.rd_kafka_topic_partition_list_add(tpl, topic, Rdkafka::Bindings::RD_KAFKA_PARTITION_UA)
  end

  # Subscribe to topic partition list and check this was successful
  @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_subscribe(inner, tpl)
    Rdkafka::RdkafkaError.validate!(response, "Error subscribing to '#{topics.join(", ")}'", client_ptr: inner)
  end
ensure
  Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(tpl) unless tpl.nil?
end

#subscription ⇒ TopicPartitionList

Returns the current subscription to topics and partitions

Returns:

Raises:



199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/rdkafka/consumer.rb', line 199

def subscription
  closed_consumer_check(__method__)

  ptr = FFI::MemoryPointer.new(:pointer)
  @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_subscription(inner, ptr)
    Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
  end

  native = ptr.read_pointer

  begin
    Rdkafka::Consumer::TopicPartitionList.from_native_tpl(native)
  ensure
    Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(native)
  end
end

#unsubscribe ⇒ nil

Unsubscribe from all subscribed topics.

Returns:

  • (nil)

Raises:



130
131
132
133
134
135
136
137
138
139
 
# File 'lib/rdkafka/consumer.rb', line 130

def unsubscribe
  closed_consumer_check(__method__)

  @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_unsubscribe(inner)
    Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
  end

  nil
end