Backport(v1.19): chunk: ensure to close the Tempfile for decompressed data (#5480)

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

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)