def deliver_message(value, key: nil, headers: {}, topic:, partition: nil, partition_key: nil, retries: 1)
      create_time = Time.now

      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.partition_for_key(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
        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

      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.partition_for_key(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
        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
Initializes a new Kafka client.

 @param seed_brokers [Array<String>, String] the list of brokers used to initialize
   the client. Either an Array of connections, or a comma separated string of connections.
   A connection can either be a string of "host:port" or a full URI with a scheme.
   If there's a scheme it's ignored and only host/port are used.

 @param client_id [String] the identifier for this application.

 @param logger [Logger] the logger that should be used by the client.

 @param connect_timeout [Integer, nil] the timeout setting for connecting
   to brokers. See {BrokerPool#initialize}.

 @param socket_timeout [Integer, nil] the timeout setting for socket
   connections. See {BrokerPool#initialize}.

 @param ssl_ca_cert [String, Array<String>, nil] a PEM encoded CA cert, or an Array of
   PEM encoded CA certs, to use with an SSL connection.

 @param ssl_ca_cert_file_path [String, nil] a path on the filesystem to a PEM encoded CA cert
   to use with an SSL connection.

 @param ssl_client_cert [String, nil] a PEM encoded client cert to use with an
   SSL connection. Must be used in combination with ssl_client_cert_key.

 @param ssl_client_cert_key [String, nil] a PEM encoded client cert key to use with an
   SSL connection. Must be used in combination with ssl_client_cert.

 @param ssl_client_cert_key_password [String, nil] the password required to read the
   ssl_client_cert_key. Must be used in combination with ssl_client_cert_key.

 @param sasl_gssapi_principal [String, nil] a KRB5 principal

 @param sasl_gssapi_keytab [String, nil] a KRB5 keytab filepath

 @param sasl_scram_username [String, nil] SCRAM username

 @param sasl_scram_password [String, nil] SCRAM password

 @param sasl_scram_mechanism [String, nil] Scram mechanism, either "sha256" or "sha512"

 @param sasl_over_ssl [Boolean] whether to enforce SSL with SASL

 @param sasl_oauth_token_provider [Object, nil] OAuthBearer Token Provider instance that
   implements method token. See {Sasl::OAuth#initialize}

 @return [Client]
 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]