class Fluent::Plugin::Output

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)
        @counters_monitor.synchronize{ @rollback_count += 1 }
        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