Method: Fluent::Plugin::Output#update_retry_state

Defined in:

lib/fluent/plugin/output.rb

#update_retry_state(chunk_id, using_secondary, error = nil) ⇒ Object

# File 'lib/fluent/plugin/output.rb', line 1309

def update_retry_state(chunk_id, using_secondary, error = nil)
  @retry_mutex.synchronize do
    @num_errors_metrics.inc
    chunk_id_hex = dump_unique_id_hex(chunk_id)

    unless @retry
      @retry = retry_state(@buffer_config.retry_randomize)

      if @retry.limit?
        handle_limit_reached(error)
      elsif error
        log_retry_error(error, chunk_id_hex, using_secondary)
      end

      return
    end

    # @retry exists

    # Ensure that the current time is greater than or equal to @retry.next_time to avoid the situation when
    # @retry.step is called almost as many times as the number of flush threads in a short time.
    if Time.now >= @retry.next_time
      @retry.step
    else
      @retry.recalc_next_time # to prevent all flush threads from retrying at the same time
    end

    if @retry.limit?
      handle_limit_reached(error)
    elsif error
      log_retry_error(error, chunk_id_hex, using_secondary)
    end
  end
end