Ensure to remove metadata if #write_metadata raise an error because of disk full

lib/fluent/plugin/buffer/file_chunk.rb

@@ -300,6 +300,13 @@ def create_new_chunk(path, perm)
             # This case is easier than enqueued!. Just removing pre-create buffer file
             @chunk.close rescue nil
             File.unlink(@path) rescue nil
+
+            if @meta
+              # ensure to unlink when #write_metadata fails
+              @meta.close rescue nil
+              File.unlink(@meta_path) rescue nil
+            end
+
             # Same as @chunk case. See above
             raise BufferOverflowError, "can't create buffer metadata for #{path}. Stop creating buffer files: error = #{e}"
           end

test/plugin/test_buffer_file_chunk.rb

@@ -495,6 +495,24 @@ def gen_chunk_path(prefix, unique_id)
     end
   end
 
+  test 'ensure to remove metadata file if #write_metadata raise an error becuase of disk full' do
+    chunk_path = File.join(@chunkdir, 'test.*.log')
+    stub(Fluent::UniqueId).hex(anything) { 'id' } # to fix chunk id
+
+    any_instance_of(Fluent::Plugin::Buffer::FileChunk) do |klass|
+      stub(klass).write_metadata(anything) do |v|
+        raise 'disk full'
+      end
+    end
+
+    err = assert_raise(Fluent::Plugin::Buffer::BufferOverflowError) do
+      Fluent::Plugin::Buffer::FileChunk.new(gen_metadata, chunk_path, :create)
+    end
+
+    assert_false File.exist?(File.join(@chunkdir, 'test.bid.log.meta'))
+    assert_match(/create buffer metadata/, err.message)
+  end
+
   sub_test_case 'chunk with file for staged chunk' do
     setup do
       @chunk_id = gen_test_chunk_id