lib/kafka/client.rb
# Initializes a new Kafka producer.
#
# @param ack_timeout [Integer] The number of seconds a broker can wait for
#   replicas to acknowledge a write before responding with a timeout.
#
# @param required_acks [Integer, Symbol] The number of replicas that must acknowledge
#   a write, or `:all` if all in-sync replicas must acknowledge.
#
# @param max_retries [Integer] the number of retries that should be attempted
#   before giving up sending messages to the cluster. Does not include the
#   original attempt.
#
# @param retry_backoff [Integer] the number of seconds to wait between retries.
#
# @param max_buffer_size [Integer] the number of messages allowed in the buffer
#   before new writes will raise {BufferOverflow} exceptions.
#
# @param max_buffer_bytesize [Integer] the maximum size of the buffer in bytes.
#   attempting to produce messages when the buffer reaches this size will
#   result in {BufferOverflow} being raised.
#
# @param compression_codec [Symbol, nil] the name of the compression codec to
#   use, or nil if no compression should be performed. Valid codecs: `:snappy`,
#   `:gzip`, `:lz4`, `:zstd`
#
# @param compression_threshold [Integer] the number of messages that needs to
#   be in a message set before it should be compressed. Note that message sets
#   are per-partition rather than per-topic or per-producer.
#
# @param interceptors [Array<Object>] a list of producer interceptors the implement
#   `call(Kafka::PendingMessage)`.
#
# @return [Kafka::Producer] the Kafka producer.
Initializes a new Kafka producer.
Initializes a new Kafka producer .
def producer(
      compression_codec: nil,
      compression_threshold: 1,
      ack_timeout: 5,
      required_acks: :all,
      max_retries: 2,
      retry_backoff: 1,
      max_buffer_size: 1000,
      max_buffer_bytesize: 10_000_000,
      idempotent: false,
      transactional: false,
      transactional_id: nil,
      transactional_timeout: 60,
      interceptors: []
    )
      cluster = initialize_cluster
      compressor = Compressor.new(
        codec_name: compression_codec,
        threshold: compression_threshold,
        instrumenter: @instrumenter,
      )
      transaction_manager = TransactionManager.new(
        cluster: cluster,
        logger: @logger,
        idempotent: idempotent,
        transactional: transactional,
        transactional_id: transactional_id,
        transactional_timeout: transactional_timeout,
      )
      Producer.new(
        cluster: cluster,
        transaction_manager: transaction_manager,
        logger: @logger,
        instrumenter: @instrumenter,
        compressor: compressor,
        ack_timeout: ack_timeout,
        required_acks: required_acks,
        max_retries: max_retries,
        retry_backoff: retry_backoff,
        max_buffer_size: max_buffer_size,
        max_buffer_bytesize: max_buffer_bytesize,
        partitioner: @partitioner,
        interceptors: interceptors
      )
    end
def producer ( compression_codec : nil , compression_threshold : 1 , ack_timeout : 5 , required_acks : :all , max_retries : 2 , retry_backoff : 1 , max_buffer_size : 1000 , max_buffer_bytesize : 10_000_000 , idempotent : false , transactional : false , transactional_id : nil , transactional_timeout : 60 , interceptors : [ ] ) cluster = initialize_cluster compressor = Compressor . new ( codec_name : compression_codec , threshold : compression_threshold , instrumenter : @instrumenter , ) transaction_manager = TransactionManager . new ( cluster : cluster , logger : @logger , idempotent : idempotent , transactional : transactional , transactional_id : transactional_id , transactional_timeout : transactional_timeout , ) Producer . new ( cluster : cluster , transaction_manager : transaction_manager , logger : @logger , instrumenter : @instrumenter , compressor : compressor , ack_timeout : ack_timeout , required_acks : required_acks , max_retries : max_retries , retry_backoff : retry_backoff , max_buffer_size : max_buffer_size , max_buffer_bytesize : max_buffer_bytesize , partitioner : @partitioner , interceptors : interceptors ) end
Initializes a new Kafka producer.
Initializes a new Kafka producer .
the Kafka producer.
the Kafka producer .
Kafka::Producer
compression_codec
the name of the compression codec to use, or nil if no compression should be performed.
the name of the compression codec to use or nil if no compression should be performed .
compression_threshold
the number of messages that needs to be in a message set before it should be compressed. Note that message sets are per-partition rather than per-topic or per-producer.
the number of messages that needs to be in a message set before it should be compressed . Note that message sets are per - partition rather than per - topic or per - producer .
ack_timeout
The number of seconds a broker can wait for replicas to acknowledge a write before responding with a timeout.
The number of seconds a broker can wait for replicas to acknowledge a write before responding with a timeout .
required_acks
The number of replicas that must acknowledge a write, or `:all` if all in-sync replicas must acknowledge.
The number of replicas that must acknowledge a write or ` : all ` if all in - sync replicas must acknowledge .
max_retries
the number of retries that should be attempted before giving up sending messages to the cluster. Does not include the original attempt.
the number of retries that should be attempted before giving up sending messages to the cluster . Does not include the original attempt .
retry_backoff
the number of seconds to wait between retries.
the number of seconds to wait between retries .
max_buffer_size
the number of messages allowed in the buffer before new writes will raise {BufferOverflow} exceptions.
the number of messages allowed in the buffer before new writes will raise { BufferOverflow } exceptions .
max_buffer_bytesize
the maximum size of the buffer in bytes. attempting to produce messages when the buffer reaches this size will result in {BufferOverflow} being raised.
the maximum size of the buffer in bytes . attempting to produce messages when the buffer reaches this size will result in { BufferOverflow } being raised .
idempotent
<not_specific>
transactional
<not_specific>
transactional_id
<not_specific>
transactional_timeout
<not_specific>
interceptors
a list of producer interceptors the implement `call(Kafka::PendingMessage)`.
a list of producer interceptors the implement ` call ( Kafka :: PendingMessage ) ` .