compressable: fix RuntimeError with large zstd compressed data (#5006)

**Which issue(s) this PR fixes**: 
Fixes #5005

**What this PR does / why we need it**: 
This PR will fix RuntimeError with large zstd compressed data

**Docs Changes**:
Not needed because this feature has not been released yet.

**Release Note**: 
Should be described with #4657

Signed-off-by: Shizuo Fujita <[email protected]>

Author: Watson1978

lib/fluent/plugin/compressable.rb

@@ -79,9 +79,9 @@ def string_decompress_gzip(compressed_data)
 
       def string_decompress_zstd(compressed_data)
         io = StringIO.new(compressed_data)
+        reader = Zstd::StreamReader.new(io)
         out = ''
         loop do
-          reader = Zstd::StreamReader.new(io)
           # Zstd::StreamReader needs to specify the size of the buffer
           out << reader.read(1024)
           # Zstd::StreamReader doesn't provide unused data, so we have to manually adjust the position
@@ -117,8 +117,8 @@ def io_decompress_gzip(input, output)
       end
 
       def io_decompress_zstd(input, output)
+        reader = Zstd::StreamReader.new(input)
         loop do
-          reader = Zstd::StreamReader.new(input)
           # Zstd::StreamReader needs to specify the size of the buffer
           v = reader.read(1024)
           output.write(v)

test/plugin/test_compressable.rb

@@ -83,5 +83,29 @@ def compress_assert_equal(expected, actual)
       assert_equal '', decompress('')
       assert_equal '', decompress('', output_io: StringIO.new)
     end
+
+    test 'decompress large zstd compressed data' do
+      src1 = SecureRandom.random_bytes(1024)
+      src2 = SecureRandom.random_bytes(1024)
+      src3 = SecureRandom.random_bytes(1024)
+
+      zstd_compressed_data = compress(src1, type: :zstd) + compress(src2, type: :zstd) + compress(src3, type: :zstd)
+      assert_equal src1 + src2 + src3, decompress(zstd_compressed_data, type: :zstd)
+    end
+
+    test 'decompress large zstd compressed data with input_io and output_io' do
+      src1 = SecureRandom.random_bytes(1024)
+      src2 = SecureRandom.random_bytes(1024)
+      src3 = SecureRandom.random_bytes(1024)
+
+      zstd_compressed_data = compress(src1, type: :zstd) + compress(src2, type: :zstd) + compress(src3, type: :zstd)
+
+      input_io = StringIO.new(zstd_compressed_data)
+      output_io = StringIO.new
+      output_io.set_encoding(src1.encoding)
+
+      decompress(input_io: input_io, output_io: output_io, type: :zstd)
+      assert_equal src1 + src2 + src3, output_io.string
+    end
   end
 end