fluent/fluentd · error · NotImplementedError

Optional API #parse_io is not implemented

Error message

Optional API #parse_io is not implemented

What it means

Optional parser API guard (lib/fluent/plugin/parser.rb:174): the base Parser#parse_io raises NotImplementedError unless the concrete parser overrides it (json and msgpack parsers do). parse_io lets a parser stream-parse an IO object instead of a whole string; in_exec/out_exec_filter call it only after implement?(:parse_io) returns true. Hitting the raise means the base implementation was invoked directly, bypassing or defeating that guard.

Source

Thrown at lib/fluent/plugin/parser.rb:174

      def call(*a, &b)
        # Keep backward compatibility for existing plugins
        # TODO: warn when deprecated
        parse(*a, &b)
      end

      def implement?(feature)
        methods_of_plugin = self.class.instance_methods(false)
        case feature
        when :parse_io then methods_of_plugin.include?(:parse_io)
        when :parse_partial_data then methods_of_plugin.include?(:parse_partial_data)
        else
          raise ArgumentError, "Unknown feature for parser plugin: #{feature}"
        end
      end

      def parse_io(io, &block)
        raise NotImplementedError, "Optional API #parse_io is not implemented"
      end

      def parse_partial_data(data, &block)
        raise NotImplementedError, "Optional API #parse_partial_data is not implemented"
      end

      def parse_time(record)
        if @time_key && record.respond_to?(:has_key?) && record.has_key?(@time_key)
          src = if @keep_time_key
                  record[@time_key]
                else
                  record.delete(@time_key)
                end
          @time_parser.parse(src)
        elsif @estimate_current_event
          Fluent::EventTime.now
        else
          nil

View on GitHub (pinned to dd45c6e18d)

Solutions

1. Always gate the call: unless parser.implement?(:parse_io), fall back to parser.parse(io.read).
2. If your custom parser declares parse_io in its method table, actually define def parse_io(io, &block) instead of aliasing the base.
3. Do not override implement? manually; let the default instance_methods(false) detection work.
4. Switch to a parser type that supports IO streaming (@type json or @type msgpack) when streaming parse is required.

Example fix

# before
@parser.parse_io(io) { |t, r| on_record(t, r) } # raises if not implemented
# after
if @parser.implement?(:parse_io)
  @parser.parse_io(io) { |t, r| on_record(t, r) }
else
  @parser.parse(io.read) { |t, r| on_record(t, r) }
end

if @parser.implement?(:parse_io)
  @parser.parse_io(io, &callback)
else
  @parser.parse(io.read, &callback)
end

def parse_io_capable?(parser)
  parser.class.instance_methods(false).include?(:parse_io)
end

begin
  @parser.parse_io(io, &callback)
rescue NotImplementedError
  @parser.parse(io.read, &callback)
end

When it happens

Trigger: Calling parser.parse_io(io) on a parser of a type that does not define it (e.g. @type regexp, csv, syslog); a custom parser subclass whose implement? logic was overridden to return true without defining parse_io; aliasing/removing methods so instance_methods(false) includes :parse_io but it delegates to super.

Common situations: Custom input plugin authors copying the in_exec dispatch block but hardcoding the parse_io branch; monkey-patched parsers (e.g. by compat shims like parser_compat) that alter the method table; plugins that define parse_io on the class but an ancestor cache or typo makes the base method win.

Related errors

• Optional API #parse_partial_data is not implemented
• Implement this method in child class
• Unknown feature for parser plugin: #{feature}
• unknown value conversion for key:'#{field_name}', type:'#{ty
• Implement this method in child class

AI-assisted analysis of fluent/fluentd@dd45c6e18d (2026-08-21). Data as JSON: /api/errors/3a52b107251d27ab. Report an issue: GitHub.