formatter_csv: use cache only when owner is provided

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

lib/fluent/compat/formatter.rb

@@ -112,6 +112,10 @@ def csv_for_thread
         def csv_thread_key
           raise NotImplementedError, "Compat CsvFormatter does not support CSV cache. Do not use this method."
         end
+
+        def csv_cacheable?
+          false
+        end
       end
 
       class SingleValueFormatter < Fluent::Plugin::SingleValueFormatter

lib/fluent/plugin/formatter_csv.rb

@@ -35,12 +35,20 @@ class CsvFormatter < Formatter
       config_param :fields, :array, value_type: :string
       config_param :add_newline, :bool, default: true
 
+      def csv_cacheable?
+        owner ? true : false
+      end
+
       def csv_thread_key
-        "#{owner.plugin_id}_csv_formatter_#{@usage}_csv"
+        csv_cacheable? ? "#{owner.plugin_id}_csv_formatter_#{@usage}_csv" : nil
       end
 
       def csv_for_thread
-        Thread.current[csv_thread_key] ||= CSV.new("".force_encoding(Encoding::ASCII_8BIT), **@generate_opts)
+        if csv_cacheable?
+          Thread.current[csv_thread_key] ||= CSV.new("".force_encoding(Encoding::ASCII_8BIT), **@generate_opts)
+        else
+          CSV.new("".force_encoding(Encoding::ASCII_8BIT), **@generate_opts)
+        end
       end
 
       def configure(conf)