Method: Fluent::Plugin::Output#try_rollback_write

Defined in:

lib/fluent/plugin/output.rb

#try_rollback_write ⇒ Object

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

def try_rollback_write
  @dequeued_chunks_mutex.synchronize do
    while @dequeued_chunks.first && @dequeued_chunks.first.expired?
      info = @dequeued_chunks.shift
      if @buffer.takeback_chunk(info.chunk_id)
        @rollback_count_metrics.inc
        log.warn "failed to flush the buffer chunk, timeout to commit.", chunk_id: dump_unique_id_hex(info.chunk_id), flushed_at: info.time
        primary = @as_secondary ? @primary_instance : self
        primary.update_retry_state(info.chunk_id, @as_secondary)
      end
    end
  end
end