diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index 9607d0e60e..ea50eb3ecb 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/lib/fluent/plugin/buffer.rb @@ -850,7 +850,12 @@ def write_step_by_step(metadata, data, format, splits_count, &block) # As already processed content is kept after rollback, then unstaged chunk should be queued. # After that, re-process current split again. # New chunk should be allocated, to do it, modify @stage and so on. - synchronize { @stage.delete(modified_metadata) } + synchronize do + @stage.delete(modified_metadata) + # Subtract this chunk's already-counted staged bytes here; + # otherwise @stage_size leaks upward over the buffer's lifetime. + @stage_size_metrics.sub(original_bytesize) if chunk.staged? + end staged_chunk_used = false chunk.unstaged! break diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index 6c62c3a542..9cb803f2f7 100644 --- a/test/plugin/test_buffer.rb +++ b/test/plugin/test_buffer.rb @@ -1339,6 +1339,23 @@ def create_chunk_es(metadata, es) c3 = create_chunk(m, ["a" * 128] * 6 + ["a" * 64]) assert !@p.chunk_size_full?(c3) end + + test '#write does not leak stage_size when a staged chunk is unstaged in write_step_by_step' do + # dm2 is staged with 896 bytes (7*128); chunk_limit_size is 1024. + # First record (64B) is appended to the 896B staged chunk (-> 960B), then + # the second record (200B) would exceed chunk_limit_size (1024), so + # write_step_by_step deletes the chunk from @stage, marks it unstaged and + # enqueues it via enqueue_unstaged_chunk. stage_size must drop by exactly + # the chunk's already-counted bytes (its 896B before this write), not by + # its current bytesize. + assert_equal 128*7 + 128*5, @p.stage_size + + dm2 = @p.metadata(timekey: @dm2.timekey) + @p.write({dm2 => ["x" * 64, "x" * 200]}) + + # Invariant: stage_size equals the real total size of the staged chunks. + assert_equal @p.stage.values.map(&:bytesize).sum, @p.stage_size + end end sub_test_case 'with configuration includes chunk_limit_records' do