lib/kafka/client.rb
# Delivers a single message to the Kafka cluster.
#
# **Note:** Only use this API for low-throughput scenarios. If you want to deliver
# many messages at a high rate, or if you want to configure the way messages are
# sent, use the {#producer} or {#async_producer} APIs instead.
#
# @param value [String, nil] the message value.
# @param key [String, nil] the message key.
# @param headers [Hash<String, String>] the headers for the message.
# @param topic [String] the topic that the message should be written to.
# @param partition [Integer, nil] the partition that the message should be written
#   to, or `nil` if either `partition_key` is passed or the partition should be
#   chosen at random.
# @param partition_key [String] a value used to deterministically choose a
#   partition to write to.
# @param retries [Integer] the number of times to retry the delivery before giving
#   up.
# @return [nil]
Delivers a single message to the Kafka cluster.
Note:** Only use this API for low-throughput scenarios. If you want to deliver
many messages at a high rate, or if you want to configure the way messages are
sent, use the {#producer} or {#async_producer} APIs instead.
Delivers a single message to the Kafka cluster . Note : ** Only use this API for low - throughput scenarios . If you want to deliver many messages at a high rate or if you want to configure the way messages are sent use the { #producer } or { #async_producer } APIs instead .
def deliver_message(value, key: nil, headers: {}, topic:, partition: nil, partition_key: nil, retries: 1)
      create_time = Time.now
      topic = topic.to_str
      message = PendingMessage.new(
        value: value,
        key: key,
        headers: headers,
        topic: topic,
        partition: partition,
        partition_key: partition_key,
        create_time: create_time
      )
      if partition.nil?
        partition_count = @cluster.partitions_for(topic).count
        partition = @partitioner.call(partition_count, message)
      end
      buffer = MessageBuffer.new
      buffer.write(
        value: message.value,
        key: message.key,
        headers: message.headers,
        topic: message.topic,
        partition: partition,
        create_time: message.create_time,
      )
      @cluster.add_target_topics([topic])
      compressor = Compressor.new(
        instrumenter: @instrumenter,
      )
      transaction_manager = TransactionManager.new(
        cluster: @cluster,
        logger: @logger,
        idempotent: false,
        transactional: false
      )
      operation = ProduceOperation.new(
        cluster: @cluster,
        transaction_manager: transaction_manager,
        buffer: buffer,
        required_acks: 1,
        ack_timeout: 10,
        compressor: compressor,
        logger: @logger,
        instrumenter: @instrumenter,
      )
      attempt = 1
      begin
        @cluster.refresh_metadata_if_necessary!
        operation.execute
        unless buffer.empty?
          raise DeliveryFailed.new(nil, [message])
        end
      rescue Kafka::Error => e
        @cluster.mark_as_stale!
        if attempt >= (retries + 1)
          raise
        else
          attempt += 1
          @logger.warn "Error while delivering message, #{e.class}: #{e.message}; retrying after 1s..."
          sleep 1
          retry
        end
      end
    end
def deliver_message ( value , key : nil , headers : { } , topic : , partition : nil , partition_key : nil , retries : 1 ) create_time = Time . now topic = topic . to_str message = PendingMessage . new ( value : value , key : key , headers : headers , topic : topic , partition : partition , partition_key : partition_key , create_time : create_time ) if partition . nil? partition_count = @cluster . partitions_for ( topic ) . count partition = @partitioner . call ( partition_count , message ) end buffer = MessageBuffer . new buffer . write ( value : message . value , key : message . key , headers : message . headers , topic : message . topic , partition : partition , create_time : message . create_time , ) @cluster . add_target_topics ( [ topic ] ) compressor = Compressor . new ( instrumenter : @instrumenter , ) transaction_manager = TransactionManager . new ( cluster : @cluster , logger : @logger , idempotent : false , transactional : false ) operation = ProduceOperation . new ( cluster : @cluster , transaction_manager : transaction_manager , buffer : buffer , required_acks : 1 , ack_timeout : 10 , compressor : compressor , logger : @logger , instrumenter : @instrumenter , ) attempt = 1 begin @cluster . refresh_metadata_if_necessary! operation . execute unless buffer . empty? raise DeliveryFailed . new ( nil , [ message ] ) end rescue Kafka :: Error => e @cluster . mark_as_stale! if attempt >= ( retries + 1 ) raise else attempt += 1 @logger . warn "Error while delivering message, #{e.class}: #{e.message}; retrying after 1s..." sleep 1 retry end end end
Delivers a single message to the Kafka cluster.
Delivers a single message to the Kafka cluster .
# We want to fail fast if `topic` isn't a String
None
nil
value
the message value.
the message value .
key
the message key.
the message key .
headers
the headers for the message.
the headers for the message .
topic
the topic that the message should be written to.
the topic that the message should be written to .
partition
the partition that the message should be written to, or `nil` if either `partition_key` is passed or the partition should be chosen at random.
the partition that the message should be written to or ` nil ` if either ` partition_key ` is passed or the partition should be chosen at random .
partition_key
a value used to deterministically choose a partition to write to.
a value used to deterministically choose a partition to write to .
retries
the number of times to retry the delivery before giving up.
the number of times to retry the delivery before giving up .