Fix for an output plugin which has #format

example/in_dummy_with_compression.conf

@@ -0,0 +1,23 @@
+<source>
+  @type dummy
+  @label @main
+  tag "test.data"
+  size 2
+  rate 10
+  dummy {"message":"yaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaay"}
+  auto_increment_key number
+</source>
+
+<label @main>
+  <match test.data>
+    @type buffered_stdout
+    <buffer>
+      @type file
+      path "#{Dir.pwd}/compressed_buffers"
+      flush_at_shutdown false
+      chunk_limit_size 1m
+      flush_interval 10s
+      compress gzip
+    </buffer>
+  </match>
+</label>

lib/fluent/plugin/buffer.rb

@@ -461,7 +461,7 @@ def write_once(metadata, data, format: nil, size: nil, &block)
               serialized = format.call(data)
               chunk.concat(serialized, size ? size.call : data.size)
             else
-              chunk.append(data)
+              chunk.append(data, compress: @compress)
             end
             adding_bytesize = chunk.bytesize - original_bytesize
 
@@ -561,7 +561,7 @@ def write_step_by_step(metadata, data, format, splits_count, &block)
                 if format
                   chunk.concat(format.call(split), split.size)
                 else
-                  chunk.append(split)
+                  chunk.append(split, compress: @compress)
                 end
 
                 if chunk_size_over?(chunk) # split size is larger than difference between size_full? and size_over?

lib/fluent/plugin/buffer/chunk.rb

@@ -21,6 +21,7 @@
 
 require 'monitor'
 require 'tempfile'
+require 'zlib'
 
 module Fluent
   module Plugin
@@ -66,7 +67,8 @@ def initialize(metadata, compress: :text)
         attr_reader :unique_id, :metadata, :created_at, :modified_at, :state
 
         # data is array of formatted record string
-        def append(data)
+        def append(data, **kwargs)
+          raise ArgumentError, '`compress: gzip` can be used for Compressable module' if kwargs[:compress] == :gzip
           adding = ''.b
           data.each do |d|
             adding << d.b
@@ -165,6 +167,20 @@ def write_to(io, **kwargs)
         module Decompressable
           include Fluent::Plugin::Compressable
 
+          def append(data, **kwargs)
+            if kwargs[:compress] == :gzip
+              io = StringIO.new
+              Zlib::GzipWriter.wrap(io) do |gz|
+                data.each do |d|
+                  gz.write d
+                end
+              end
+              concat(io.string, data.size)
+            else
+              super
+            end
+          end
+
           def open(**kwargs, &block)
             if kwargs[:compressed] == :gzip
               super

test/plugin/test_buffer_chunk.rb

@@ -53,6 +53,7 @@ class BufferChunkTest < Test::Unit::TestCase
       assert_raise(ArgumentError){ chunk.read(compressed: :gzip) }
       assert_raise(ArgumentError){ chunk.open(compressed: :gzip){} }
       assert_raise(ArgumentError){ chunk.write_to(nil, compressed: :gzip) }
+      assert_raise(ArgumentError){ chunk.append(nil, compress: :gzip) }
     end
   end
 

test/plugin/test_buffer_file_chunk.rb

@@ -778,6 +778,17 @@ def gen_chunk_path(prefix, unique_id)
       @gzipped_src = compress(@src)
     end
 
+    test '#append with compress option writes  compressed data to chunk when compress is gzip' do
+      c = @klass.new(gen_metadata, File.join(@chunkdir,'test.*.log'), :create, compress: :gzip)
+      c.append([@src, @src], compress: :gzip)
+      c.commit
+
+      # check chunk is compressed
+      assert c.read(compressed: :gzip).size < [@src, @src].join("").size
+
+      assert_equal @src + @src, c.read
+    end
+
     test '#open 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)

test/plugin/test_buffer_memory_chunk.rb

@@ -272,6 +272,17 @@ class BufferMemoryChunkTest < Test::Unit::TestCase
       @gzipped_src = compress(@src)
     end
 
+    test '#append with compress option writes  compressed data to chunk when compress is gzip' do
+      c = Fluent::Plugin::Buffer::MemoryChunk.new(Object.new, compress: :gzip)
+      c.append([@src, @src], compress: :gzip)
+      c.commit
+
+      # check chunk is compressed
+      assert c.read(compressed: :gzip).size < [@src, @src].join("").size
+
+      assert_equal @src + @src, c.read
+    end
+
     test '#open passes io object having decompressed data to a block when compress is gzip' do
       c = Fluent::Plugin::Buffer::MemoryChunk.new(Object.new, compress: :gzip)
       c.concat(@gzipped_src, @src.size)