diff --git a/lib/fluent/plugin/buffer/chunk.rb b/lib/fluent/plugin/buffer/chunk.rb index c5b8f4a64d..5cc98ec346 100644 --- a/lib/fluent/plugin/buffer/chunk.rb +++ b/lib/fluent/plugin/buffer/chunk.rb @@ -220,9 +220,13 @@ def open(**kwargs, &block) Tempfile.new('decompressed-data') end output_io.binmode if output_io.is_a?(Tempfile) - decompress(input_io: chunk_io, output_io: output_io) - output_io.seek(0, IO::SEEK_SET) - yield output_io + begin + decompress(input_io: chunk_io, output_io: output_io) + output_io.seek(0, IO::SEEK_SET) + yield output_io + ensure + output_io.close! if output_io.is_a?(Tempfile) + end end end end @@ -274,9 +278,13 @@ def open(**kwargs, &block) Tempfile.new('decompressed-data') end output_io.binmode if output_io.is_a?(Tempfile) - decompress(input_io: chunk_io, output_io: output_io, type: :zstd) - output_io.seek(0, IO::SEEK_SET) - yield output_io + begin + decompress(input_io: chunk_io, output_io: output_io, type: :zstd) + output_io.seek(0, IO::SEEK_SET) + yield output_io + ensure + output_io.close! if output_io.is_a?(Tempfile) + end end end end diff --git a/test/plugin/test_buffer_file_chunk.rb b/test/plugin/test_buffer_file_chunk.rb index 755869a7ae..87125433bf 100644 --- a/test/plugin/test_buffer_file_chunk.rb +++ b/test/plugin/test_buffer_file_chunk.rb @@ -829,6 +829,42 @@ def gen_chunk_path(prefix, unique_id) assert_equal @src, decomressed_data end + test '#open closes the temporary file for decompressed data when compress is gzip' do + c = @klass.new(gen_metadata, File.join(@chunkdir,'test.*.log'), :create, compress: :gzip) + c.concat(@gzipped_src, @src.size) + c.commit + + decompressed_io = nil + path = nil + c.open do |io| + decompressed_io = io + path = io.path + assert_true File.exist?(path) + end + + assert_true decompressed_io.closed? + assert_false File.exist?(path) + end + + test '#open closes the temporary file for decompressed data even when the block raises and compress is gzip' do + c = @klass.new(gen_metadata, File.join(@chunkdir,'test.*.log'), :create, compress: :gzip) + c.concat(@gzipped_src, @src.size) + c.commit + + decompressed_io = nil + path = nil + assert_raise RuntimeError.new('failed to consume decompressed data') do + c.open do |io| + decompressed_io = io + path = io.path + raise 'failed to consume decompressed data' + end + end + + assert_true decompressed_io.closed? + assert_false File.exist?(path) + end + test '#open with compressed option passes io object having decompressed data to a block when compress is gzip' do c = @klass.new(gen_metadata, File.join(@chunkdir,'test.*.log'), :create, compress: :gzip) c.concat(@gzipped_src, @src.size) @@ -893,6 +929,42 @@ def gen_chunk_path(prefix, unique_id) assert_equal @src, decomressed_data end + test '#open closes the temporary file for decompressed data when compress is zstd' do + c = @klass.new(gen_metadata, File.join(@chunkdir,'test.*.log'), :create, compress: :zstd) + c.concat(@zstded_src, @src.size) + c.commit + + decompressed_io = nil + path = nil + c.open do |io| + decompressed_io = io + path = io.path + assert_true File.exist?(path) + end + + assert_true decompressed_io.closed? + assert_false File.exist?(path) + end + + test '#open closes the temporary file for decompressed data even when the block raises and compress is zstd' do + c = @klass.new(gen_metadata, File.join(@chunkdir,'test.*.log'), :create, compress: :zstd) + c.concat(@zstded_src, @src.size) + c.commit + + decompressed_io = nil + path = nil + assert_raise RuntimeError.new('failed to consume decompressed data') do + c.open do |io| + decompressed_io = io + path = io.path + raise 'failed to consume decompressed data' + end + end + + assert_true decompressed_io.closed? + assert_false File.exist?(path) + end + test '#open with compressed option passes io object having decompressed data to a block when compress is zstd' do c = @klass.new(gen_metadata, File.join(@chunkdir,'test.*.log'), :create, compress: :zstd) c.concat(@zstded_src, @src.size)