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
    )
      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,
      )
    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
    )
      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,
      )
    end
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.

 @return [Kafka::Producer] the Kafka producer.