Class: Rdkafka::Producer

Inherits:
Object
  • Object
show all
Includes:
Helpers::Metadata, Helpers::OAuth, Helpers::Time
Defined in:
lib/rdkafka/producer.rb,
lib/rdkafka/producer/delivery_handle.rb,
lib/rdkafka/producer/delivery_report.rb,
lib/rdkafka/producer/partitions_count_cache.rb

Overview

A producer for Kafka messages. To create a producer set up a Config and call producer on that.

Defined Under Namespace

Classes: DeliveryHandle, DeliveryReport, PartitionsCountCache, TopicHandleCreationError

Constant Summary collapse

@@partitions_count_cache = 
PartitionsCountCache.new

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Helpers::Metadata

#metadata

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, partitioner) ⇒ Producer

Returns a new instance of Producer.

Parameters:

  • native_kafka (NativeKafka)
  • partitioner (String, nil) —

    name of the partitioner we want to use or nil to use the "consistent_random" default



56
57
58
59
60
61
62
63
64
 
# File 'lib/rdkafka/producer.rb', line 56

def initialize(native_kafka, partitioner)
  @topics_refs_map = {}
  @topics_configs = {}
  @native_kafka = native_kafka
  @partitioner = partitioner || "consistent_random"

  # Makes sure, that native kafka gets closed before it gets GCed by Ruby
  ObjectSpace.define_finalizer(self, native_kafka.finalizer)
end

Instance Attribute Details

#delivery_callback ⇒ Proc?

Returns the current delivery callback, by default this is nil.

Returns:

  • (Proc, nil) 


44
45
46
 
# File 'lib/rdkafka/producer.rb', line 44

def delivery_callback
  @delivery_callback
end

#delivery_callback_arity ⇒ Integer? (readonly)

Returns the number of arguments accepted by the callback, by default this is nil.

Returns:

  • (Integer, nil) 


50
51
52
 
# File 'lib/rdkafka/producer.rb', line 50

def delivery_callback_arity
  @delivery_callback_arity
end

Class Method Details

.partitions_count_cache ⇒ Rdkafka::Producer::PartitionsCountCache

Note:

It is critical to remember, that not all users may have statistics callbacks enabled, hence we should not make assumption that this cache is always updated from the stats.

Global (process wide) partitions cache. We use it to store number of topics partitions, either from the librdkafka statistics (if enabled) or via direct inline calls every now and then. Since the partitions count can only grow and should be same for all consumers and producers, we can use a global cache as long as we ensure that updates only move up.

Returns:



21
22
23
 
# File 'lib/rdkafka/producer.rb', line 21

def self.partitions_count_cache
  @@partitions_count_cache
end

.partitions_count_cache=(partitions_count_cache) ⇒ Object

Parameters:



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

def self.partitions_count_cache=(partitions_count_cache)
  @@partitions_count_cache = partitions_count_cache
end

Instance Method Details

#abort_transaction(timeout_ms = -1)) ⇒ true

Abort the current transaction

Parameters:

  • timeout_ms (Integer) (defaults to: -1)) —

    Timeout in milliseconds (-1 for infinite)

Returns:

  • (true) —

    Returns true on success

Raises:



213
214
215
216
217
218
219
220
 
# File 'lib/rdkafka/producer.rb', line 213

def abort_transaction(timeout_ms = -1)
  closed_producer_check(__method__)

  @native_kafka.with_inner do |inner|
    response_ptr = Rdkafka::Bindings.rd_kafka_abort_transaction(inner, timeout_ms)
    Rdkafka::RdkafkaError.validate!(response_ptr, client_ptr: inner) || true
  end
end

#arity(callback) ⇒ Integer

Figures out the arity of a given block/method

Parameters:

  • callback (#call, Proc)

Returns:

  • (Integer) —

    arity of the provided block/method



615
616
617
618
619
 
# File 'lib/rdkafka/producer.rb', line 615

def arity(callback)
  return callback.arity if callback.respond_to?(:arity)

  callback.method(:call).arity
end

#begin_transaction ⇒ true

Begin a new transaction Requires #init_transactions to have been called first

Returns:

  • (true) —

    Returns true on success

Raises:



183
184
185
186
187
188
189
190
191
 
# File 'lib/rdkafka/producer.rb', line 183

def begin_transaction
  closed_producer_check(__method__)

  @native_kafka.with_inner do |inner|
    response_ptr = Rdkafka::Bindings.rd_kafka_begin_transaction(inner)

    Rdkafka::RdkafkaError.validate!(response_ptr, client_ptr: inner) || true
  end
end

#call_delivery_callback(delivery_report, delivery_handle) ⇒ Object

Calls (if registered) the delivery callback

Parameters:



598
599
600
601
602
603
604
605
606
607
608
609
 
# File 'lib/rdkafka/producer.rb', line 598

def call_delivery_callback(delivery_report, delivery_handle)
  return unless @delivery_callback

  case @delivery_callback_arity
  when 0
    @delivery_callback.call
  when 1
    @delivery_callback.call(delivery_report)
  else
    @delivery_callback.call(delivery_report, delivery_handle)
  end
end

#close ⇒ Object

Close this producer and wait for the internal poll queue to empty.



250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
 
# File 'lib/rdkafka/producer.rb', line 250

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

  @native_kafka.close do
    # We need to remove the topics references objects before we destroy the producer,
    # otherwise they would leak out
    @topics_refs_map.each_value do |refs|
      refs.each_value do |ref|
        Rdkafka::Bindings.rd_kafka_topic_destroy(ref)
      end
    end
  end

  @topics_refs_map.clear
end

#closed? ⇒ Boolean

Whether this producer has closed

Returns:

  • (Boolean) 


268
269
270
 
# File 'lib/rdkafka/producer.rb', line 268

def closed?
  @native_kafka.closed?
end

#commit_transaction(timeout_ms = -1)) ⇒ true

Commit the current transaction

Parameters:

  • timeout_ms (Integer) (defaults to: -1)) —

    Timeout in milliseconds (-1 for infinite)

Returns:

  • (true) —

    Returns true on success

Raises:



198
199
200
201
202
203
204
205
206
 
# File 'lib/rdkafka/producer.rb', line 198

def commit_transaction(timeout_ms = -1)
  closed_producer_check(__method__)

  @native_kafka.with_inner do |inner|
    response_ptr = Rdkafka::Bindings.rd_kafka_commit_transaction(inner, timeout_ms)

    Rdkafka::RdkafkaError.validate!(response_ptr, client_ptr: inner) || true
  end
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:



148
149
150
 
# File 'lib/rdkafka/producer.rb', line 148

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 delivery confirmations arrive, librdkafka will write to your FD

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:



139
140
141
 
# File 'lib/rdkafka/producer.rb', line 139

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

#events_poll_nb_each {|count| ... } ⇒ nil

Note:

This method holds the inner lock until the queue is empty or :stop is returned. Other producer operations (produce, close, etc.) will wait until this method returns.

Note:

This method is thread-safe as it uses @native_kafka.with_inner synchronization

Polls for events in a non-blocking loop, yielding the count after each iteration.

This method processes delivery callbacks in a single GVL/mutex session, which is more efficient than repeated individual polls. It uses non-blocking polls internally (no GVL release between polls).

Yields the count of events processed after each poll iteration, allowing the caller to implement timeout or other termination logic by returning :stop.

Examples:

Drain all pending callbacks

producer.events_poll_nb_each { |_count| }

With timeout control

deadline = monotonic_now + timeout_ms
producer.events_poll_nb_each do |_count|
  :stop if monotonic_now >= deadline
end

Yields:

  • (count) —

    Called after each poll iteration

Yield Parameters:

  • count (Integer) —

    Number of events processed in this iteration

Yield Returns:

  • (Symbol, Object) —

    Return :stop to break the loop, any other value continues

Returns:

  • (nil)

Raises:



375
376
377
378
379
380
381
382
383
384
385
 
# File 'lib/rdkafka/producer.rb', line 375

def events_poll_nb_each
  closed_producer_check(__method__)

  @native_kafka.with_inner do |inner|
    loop do
      count = Rdkafka::Bindings.rd_kafka_poll_nb(inner, 0)
      break if count.zero?
      break if yield(count) == :stop
    end
  end
end

#flush(timeout_ms = Defaults::PRODUCER_FLUSH_TIMEOUT_MS) ⇒ Boolean

Note:

We raise an exception for other errors because based on the librdkafka docs, there should be no other errors.

Note:

For timed_out we do not raise an error to keep it backwards compatible

Wait until all outstanding producer requests are completed, with the given timeout in seconds. Call this before closing a producer to ensure delivery of all messages.

Parameters:

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

    how long should we wait for flush of all messages

Returns:

  • (Boolean) —

    true if no more data and all was flushed, false in case there are still outgoing messages after the timeout



283
284
285
286
287
288
289
290
291
292
293
294
295
296
 
# File 'lib/rdkafka/producer.rb', line 283

def flush(timeout_ms = Defaults::PRODUCER_FLUSH_TIMEOUT_MS)
  closed_producer_check(__method__)

  error = @native_kafka.with_inner do |inner|
    response = Rdkafka::Bindings.rd_kafka_flush(inner, timeout_ms)
    Rdkafka::RdkafkaError.build(response)
  end

  # Early skip not to build the error message
  return true unless error
  return false if error.code == :timed_out

  raise(error)
end

#init_transactions ⇒ true

Initialize transactions for the producer Must be called once before any transactional operations

Returns:

  • (true) —

    Returns true on success

Raises:



168
169
170
171
172
173
174
175
176
 
# File 'lib/rdkafka/producer.rb', line 168

def init_transactions
  closed_producer_check(__method__)

  @native_kafka.with_inner do |inner|
    response_ptr = Rdkafka::Bindings.rd_kafka_init_transactions(inner, -1)

    Rdkafka::RdkafkaError.validate!(response_ptr, client_ptr: inner) || true
  end
end

#name ⇒ String

Returns producer name.

Returns:

  • (String) —

    producer name



126
127
128
129
130
 
# File 'lib/rdkafka/producer.rb', line 126

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

#partition_count(topic) ⇒ Integer

Note:

If 'allow.auto.create.topics' is set to true in the broker, the topic will be auto-created after returning nil.

Note:

We cache the partition count for a given topic for given time. If statistics are enabled for any producer or consumer, it will take precedence over per instance fetching.

This prevents us, in case someone uses partition_key, from querying for the count with each message. A missing topic (unknown_topic_or_part) is cached as RD_KAFKA_PARTITION_UA too, so producing to a not-yet-created topic does not run a blocking metadata query on every message; the cache TTL still bounds how long that result is reused before we look again.

Partition count for a given topic.

Parameters:

  • topic (String) —

    The topic name.

Returns:

  • (Integer) —

    partition count for a given topic or RD_KAFKA_PARTITION_UA (-1) if it could not be obtained.



403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
 
# File 'lib/rdkafka/producer.rb', line 403

def partition_count(topic)
  closed_producer_check(__method__)

  self.class.partitions_count_cache.get(topic) do
    topic_metadata = nil

    @native_kafka.with_inner do |inner|
      topic_metadata = ::Rdkafka::Metadata.new(inner, topic).topics&.first
    end

    topic_metadata ? topic_metadata[:partition_count] : Rdkafka::Bindings::RD_KAFKA_PARTITION_UA
  rescue Rdkafka::RdkafkaError => e
    # A missing topic is an expected, cacheable outcome (the topic may be auto-created on
    # produce, or simply not exist yet). Returning RD_KAFKA_PARTITION_UA from inside the block
    # caches it, so we don't re-run a blocking metadata query on every produce(partition_key:)
    # to that topic. Anything else is a real error and is re-raised.
    raise(e) unless e.code == :unknown_topic_or_part

    Rdkafka::Bindings::RD_KAFKA_PARTITION_UA
  end
end

#produce(topic:, payload: nil, key: nil, partition: nil, partition_key: nil, timestamp: nil, headers: nil, label: nil, topic_config: EMPTY_HASH, partitioner: @partitioner) ⇒ DeliveryHandle

Produces a message to a Kafka topic. The message is added to rdkafka's queue, call wait on the returned delivery handle to make sure it is delivered.

When no partition is specified the underlying Kafka library picks a partition based on the key. If no key is specified, a random partition will be used. When a timestamp is provided this is used instead of the auto-generated timestamp.

Parameters:

  • topic (String) —

    The topic to produce to

  • payload (String, nil) (defaults to: nil)
  • key (String, nil) (defaults to: nil)
  • partition (Integer, nil) (defaults to: nil) —

    Optional partition to produce to

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

    Optional partition key based on which partition assignment can happen

  • timestamp (Time, Integer, nil) (defaults to: nil) —

    Optional timestamp of this message. Integer timestamp is in milliseconds since Jan 1 1970.

  • headers (Hash{String => String, Array<String>}) (defaults to: nil) —

    Optional message headers. Values can be either a single string or an array of strings to support duplicate headers per KIP-82

  • label (Object, nil) (defaults to: nil) —

    a label that can be assigned when producing a message that will be part of the delivery handle and the delivery report

  • topic_config (Hash) (defaults to: EMPTY_HASH) —

    topic config for given message dispatch. Allows to send messages to topics with different configuration

  • partitioner (String) (defaults to: @partitioner) —

    name of the partitioner to use

Returns:

  • (DeliveryHandle) —

    Delivery handle that can be used to wait for the result of producing this message

Raises:

  • (RdkafkaError) —

    When adding the message to rdkafka's queue failed



444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
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
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
 
# File 'lib/rdkafka/producer.rb', line 444

def produce(
  topic:,
  payload: nil,
  key: nil,
  partition: nil,
  partition_key: nil,
  timestamp: nil,
  headers: nil,
  label: nil,
  topic_config: EMPTY_HASH,
  partitioner: @partitioner
)
  closed_producer_check(__method__)

  # Start by checking and converting the input

  # Get payload length
  payload_size = if payload.nil?
    0
  else
    payload.bytesize
  end

  # Get key length
  key_size = if key.nil?
    0
  else
    key.bytesize
  end

  topic_config_hash = topic_config.hash

  # Checks if we have the rdkafka topic reference object ready. It saves us on object
  # allocation and allows to use custom config on demand.
  set_topic_config(topic, topic_config, topic_config_hash) unless @topics_refs_map.dig(topic, topic_config_hash)
  topic_ref = @topics_refs_map.dig(topic, topic_config_hash)

  if partition_key
    partition_count = partition_count(topic)

    # Check if there are no overrides for the partitioner and use the default one only when
    # no per-topic is present.
    selected_partitioner = @topics_configs.dig(topic, topic_config_hash, :partitioner) || partitioner

    # If the topic is not present, set to -1
    if partition_count.positive?
      partition = Rdkafka::Bindings.partitioner(
        topic_ref,
        partition_key,
        partition_count,
        selected_partitioner
      )
    end
  end

  # If partition is nil, use RD_KAFKA_PARTITION_UA to let librdafka set the partition randomly or
  # based on the key when present.
  partition ||= Rdkafka::Bindings::RD_KAFKA_PARTITION_UA

  # If timestamp is nil use 0 and let Kafka set one. If an integer or time
  # use it.
  raw_timestamp = if timestamp.nil?
    0
  elsif timestamp.is_a?(Integer)
    timestamp
  elsif timestamp.is_a?(Time)
    (timestamp.to_i * 1000) + (timestamp.usec / 1000)
  else
    raise TypeError.new("Timestamp has to be nil, an Integer or a Time")
  end

  delivery_handle = DeliveryHandle.new
  delivery_handle.label = label
  delivery_handle.topic = topic
  delivery_handle[:pending] = true
  delivery_handle[:response] = Rdkafka::Bindings::RD_KAFKA_PARTITION_UA
  delivery_handle[:partition] = Rdkafka::Bindings::RD_KAFKA_PARTITION_UA
  delivery_handle[:offset] = Rdkafka::Bindings::RD_KAFKA_PARTITION_UA
  DeliveryHandle.register(delivery_handle)

  # Everything between registration and the successful return must remove the handle from the
  # process-global registry if it raises. Otherwise a failure after registration (a concurrent
  # close making `with_inner` raise `ClosedInnerError`, or a header value whose `#to_s` raises)
  # would orphan the handle in the registry forever, since nothing else will ever resolve it.
  begin
    args = [
      :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_RKT, :pointer, topic_ref,
      :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_MSGFLAGS, :int, Rdkafka::Bindings::RD_KAFKA_MSG_F_COPY,
      :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_VALUE, :buffer_in, payload, :size_t, payload_size,
      :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_KEY, :buffer_in, key, :size_t, key_size,
      :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_PARTITION, :int32, partition,
      :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_TIMESTAMP, :int64, raw_timestamp,
      :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_OPAQUE, :pointer, delivery_handle
    ]

    if headers && !headers.empty?
      headers.each do |key0, value0|
        key = key0.to_s
        case value0
        when Array
          # Handle array of values per KIP-82
          value0.each do |v|
            value = v.to_s
            args.push(
              :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_HEADER,
              :string, key,
              :pointer, value,
              :size_t, value.bytesize
            )
          end
        else
          # Handle single value
          value = value0.to_s
          args.push(
            :int, Rdkafka::Bindings::RD_KAFKA_VTYPE_HEADER,
            :string, key,
            :pointer, value,
            :size_t, value.bytesize
          )
        end
      end
    end

    args.push(:int, Rdkafka::Bindings::RD_KAFKA_VTYPE_END)

    # Produce the message
    response = @native_kafka.with_inner do |inner|
      Rdkafka::Bindings.rd_kafka_producev(
        inner,
        *args
      )
    end

    # Raise error if the produce call was not successful. On a successful enqueue the handle
    # stays registered on purpose: the delivery callback resolves and removes it once librdkafka
    # reports the delivery.
    if response != Rdkafka::Bindings::RD_KAFKA_RESP_ERR_NO_ERROR
      @native_kafka.with_inner do |inner|
        Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
      end
    end
  rescue Exception
    DeliveryHandle.remove(delivery_handle.to_ptr.address)

    raise
  end

  delivery_handle
end

#purge ⇒ Object

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.



304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
 
# File 'lib/rdkafka/producer.rb', line 304

def purge
  closed_producer_check(__method__)

  @native_kafka.with_inner do |inner|
    response = Bindings.rd_kafka_purge(
      inner,
      Bindings::RD_KAFKA_PURGE_F_QUEUE | Bindings::RD_KAFKA_PURGE_F_INFLIGHT
    )

    Rdkafka::RdkafkaError.validate!(response, client_ptr: inner)
  end

  # Wait for the purge to affect everything
  sleep(Defaults::PRODUCER_PURGE_SLEEP_INTERVAL_MS / 1_000.0) until flush(Defaults::PRODUCER_PURGE_FLUSH_TIMEOUT_MS)

  true
end

#queue_size ⇒ Integer Also known as: queue_length

Note:

This method is thread-safe as it uses the @native_kafka.with_inner synchronization

Returns the number of messages and requests waiting to be sent to the broker as well as delivery reports queued for the application.

This provides visibility into the producer's internal queue depth, useful for:

  • Monitoring producer backpressure
  • Implementing custom flow control
  • Debugging message delivery issues
  • Graceful shutdown logic (wait until queue is empty)

Examples:

producer.queue_size #=> 42

Returns:

  • (Integer) —

    the number of messages in the queue

Raises:



338
339
340
341
342
343
344
 
# File 'lib/rdkafka/producer.rb', line 338

def queue_size
  closed_producer_check(__method__)

  @native_kafka.with_inner do |inner|
    Rdkafka::Bindings.rd_kafka_outq_len(inner)
  end
end

#send_offsets_to_transaction(consumer, tpl, timeout_ms = Defaults::PRODUCER_SEND_OFFSETS_TIMEOUT_MS) ⇒ Object

Note:

Use only in the context of an active transaction

Sends provided offsets of a consumer to the transaction for collective commit

Parameters:

  • consumer (Consumer) —

    consumer that owns the given tpls

  • tpl (Consumer::TopicPartitionList)
  • timeout_ms (Integer) (defaults to: Defaults::PRODUCER_SEND_OFFSETS_TIMEOUT_MS) —

    offsets send timeout



228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/rdkafka/producer.rb', line 228

def send_offsets_to_transaction(consumer, tpl, timeout_ms = Defaults::PRODUCER_SEND_OFFSETS_TIMEOUT_MS)
  closed_producer_check(__method__)

  return if tpl.empty?

  cgmetadata = consumer.consumer_group_metadata_pointer
  native_tpl = tpl.to_native_tpl

  @native_kafka.with_inner do |inner|
    response_ptr = Bindings.rd_kafka_send_offsets_to_transaction(inner, native_tpl, cgmetadata, timeout_ms)

    Rdkafka::RdkafkaError.validate!(response_ptr, client_ptr: inner)
  end
ensure
  if cgmetadata && !cgmetadata.null?
    Bindings.rd_kafka_consumer_group_metadata_destroy(cgmetadata)
  end

  Rdkafka::Bindings.rd_kafka_topic_partition_list_destroy(native_tpl) unless native_tpl.nil?
end

#set_topic_config(topic, config, config_hash) ⇒ Object

Note:

It is not allowed to re-set the same topic config twice because of the underlying librdkafka caching

Sets alternative set of configuration details that can be set per topic

Parameters:

  • topic (String) —

    The topic name

  • config (Hash) —

    config we want to use per topic basis

  • config_hash (Integer) —

    hash of the config. We expect it here instead of computing it, because it is already computed during the retrieval attempt in the #produce flow.



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/rdkafka/producer.rb', line 74

def set_topic_config(topic, config, config_hash)
  # Ensure lock on topic reference just in case
  @native_kafka.with_inner do |inner|
    @topics_refs_map[topic] ||= {}
    @topics_configs[topic] ||= {}

    return if @topics_configs[topic].key?(config_hash)

    # If config is empty, we create an empty reference that will be used with defaults
    rd_topic_config = if config.empty?
      nil
    else
      Rdkafka::Bindings.rd_kafka_topic_conf_new.tap do |topic_config|
        config.each do |key, value|
          error_buffer = FFI::MemoryPointer.new(:char, 256)
          result = Rdkafka::Bindings.rd_kafka_topic_conf_set(
            topic_config,
            key.to_s,
            value.to_s,
            error_buffer,
            256
          )

          unless result == :config_ok
            # rd_kafka_topic_new has not been called yet, so librdkafka has not taken
            # ownership of the config. Destroy it ourselves before raising, otherwise the
            # partially built rd_kafka_topic_conf_t leaks. (On the rd_kafka_topic_new path
            # librdkafka frees the conf on both success and failure, so we never destroy it
            # there.)
            Rdkafka::Bindings.rd_kafka_topic_conf_destroy(topic_config)
            raise Config::ConfigError.new(error_buffer.read_string)
          end
        end
      end
    end

    topic_handle = Bindings.rd_kafka_topic_new(inner, topic, rd_topic_config)

    raise TopicHandleCreationError.new("Error creating topic handle for topic #{topic}") if topic_handle.null?

    @topics_configs[topic][config_hash] = config
    @topics_refs_map[topic][config_hash] = topic_handle
  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



121
122
123
 
# File 'lib/rdkafka/producer.rb', line 121

def start
  @native_kafka.start
end