diff --git a/lib/fluent/command/fluentd.rb b/lib/fluent/command/fluentd.rb
index 2b2d229a01..2150ece521 100644
--- a/lib/fluent/command/fluentd.rb
+++ b/lib/fluent/command/fluentd.rb
@@ -126,6 +126,16 @@
   opts[:without_source] = b
 }
 
+op.on('--config-file-type VALU', 'guessing file type of fluentd configuration. yaml/yml or guess') { |s|
+  if (s == 'yaml') || (s == 'yml')
+    opts[:config_file_type] = s.to_sym
+  elsif (s == 'guess')
+    opts[:config_file_type] = s.to_sym
+  else
+    usage '--config-file-type accepts yaml/yml or guess'
+  end
+}
+
 op.on('--use-v1-config', "Use v1 configuration format (default)", TrueClass) {|b|
   opts[:use_v1_config] = b
 }
diff --git a/lib/fluent/config.rb b/lib/fluent/config.rb
index 8ca5cc6618..d471114830 100644
--- a/lib/fluent/config.rb
+++ b/lib/fluent/config.rb
@@ -17,6 +17,7 @@
 require 'fluent/config/error'
 require 'fluent/config/element'
 require 'fluent/configurable'
+require 'fluent/config/yaml_parser'
 
 module Fluent
   module Config
@@ -25,7 +26,18 @@ module Config
     # @param additional_config [String] config which is added to last of config body
     # @param use_v1_config [Bool] config is formatted with v1 or not
     # @return [Fluent::Config]
-    def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_config: true)
+    def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_config: true, type: nil)
+      if type == :guess
+        config_file_ext = File.extname(config_path)
+        if config_file_ext == '.yaml' || config_file_ext == '.yml'
+          type = :yaml
+        end
+      end
+
+      if type == :yaml || type == :yml
+        return Fluent::Config::YamlParser.parse(config_path)
+      end
+
       config_fname = File.basename(config_path)
       config_basedir = File.dirname(config_path)
       config_data = File.open(config_path, "r:#{encoding}:utf-8") do |f|
@@ -36,6 +48,7 @@ def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_c
         end
         s
       end
+
       Fluent::Config.parse(config_data, config_fname, config_basedir, use_v1_config)
     end
 
diff --git a/lib/fluent/config/yaml_parser.rb b/lib/fluent/config/yaml_parser.rb
new file mode 100644
index 0000000000..e000bfa446
--- /dev/null
+++ b/lib/fluent/config/yaml_parser.rb
@@ -0,0 +1,56 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'fluent/config/yaml_parser/loader'
+require 'fluent/config/yaml_parser/parser'
+require 'pathname'
+
+module Fluent
+  module Config
+    module YamlParser
+      def self.parse(path)
+        context = Kernel.binding
+
+        unless context.respond_to?(:use_nil)
+          context.define_singleton_method(:use_nil) do
+            raise Fluent::SetNil
+          end
+        end
+
+        unless context.respond_to?(:use_default)
+          context.define_singleton_method(:use_default) do
+            raise Fluent::SetDefault
+          end
+        end
+
+        unless context.respond_to?(:hostname)
+          context.define_singleton_method(:hostname) do
+            Socket.gethostname
+          end
+        end
+
+        unless context.respond_to?(:worker_id)
+          context.define_singleton_method(:worker_id) do
+            ENV['SERVERENGINE_WORKER_ID'] || ''
+          end
+        end
+
+        s = Fluent::Config::YamlParser::Loader.new(context).load(Pathname.new(path))
+        Fluent::Config::YamlParser::Parser.new(s).build.to_element
+      end
+    end
+  end
+end
diff --git a/lib/fluent/config/yaml_parser/fluent_value.rb b/lib/fluent/config/yaml_parser/fluent_value.rb
new file mode 100644
index 0000000000..69108d8bd1
--- /dev/null
+++ b/lib/fluent/config/yaml_parser/fluent_value.rb
@@ -0,0 +1,47 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+module Fluent
+  module Config
+    module YamlParser
+      module FluentValue
+        JsonValue = Struct.new(:val) do
+          def to_s
+            val.to_json
+          end
+
+          def to_element
+            to_s
+          end
+        end
+
+        StringValue = Struct.new(:val, :context) do
+          def to_s
+            context.instance_eval("\"#{val}\"")
+          rescue Fluent::SetNil => _
+            ''
+          rescue Fluent::SetDefault => _
+            ':default'
+          end
+
+          def to_element
+            to_s
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/fluent/config/yaml_parser/loader.rb b/lib/fluent/config/yaml_parser/loader.rb
new file mode 100644
index 0000000000..22f32885e4
--- /dev/null
+++ b/lib/fluent/config/yaml_parser/loader.rb
@@ -0,0 +1,91 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'psych'
+require 'json'
+require 'fluent/config/error'
+require 'fluent/config/yaml_parser/fluent_value'
+
+# Based on https://github.com/eagletmt/hako/blob/34cdde06fe8f3aeafd794be830180c3cedfbb4dc/lib/hako/yaml_loader.rb
+
+module Fluent
+  module Config
+    module YamlParser
+      class Loader
+        INCLUDE_TAG = 'tag:include'.freeze
+        FLUENT_JSON_TAG = 'tag:fluent/json'.freeze
+        FLUENT_STR_TAG = 'tag:fluent/s'.freeze
+        SHOVEL = '<<'.freeze
+
+        def initialize(context = Kernel.binding)
+          @context = context
+          @current_path = nil
+        end
+
+        # @param [String] path
+        # @return [Hash]
+        def load(path)
+          class_loader = Psych::ClassLoader.new
+          scanner = Psych::ScalarScanner.new(class_loader)
+
+          visitor = Visitor.new(scanner, class_loader)
+
+          visitor._register_domain(INCLUDE_TAG) do |_, val|
+            load(path.parent.join(val))
+          end
+
+          visitor._register_domain(FLUENT_JSON_TAG) do |_, val|
+            Fluent::Config::YamlParser::FluentValue::JsonValue.new(val)
+          end
+
+          visitor._register_domain(FLUENT_STR_TAG) do |_, val|
+            Fluent::Config::YamlParser::FluentValue::StringValue.new(val, @context)
+          end
+
+          path.open do |f|
+            visitor.accept(Psych.parse(f))
+          end
+        end
+
+        class Visitor < Psych::Visitors::ToRuby
+          def initialize(scanner, class_loader)
+            super(scanner, class_loader)
+          end
+
+          def _register_domain(name, &block)
+            @domain_types.merge!({ name => [name, block] })
+          end
+
+          def revive_hash(hash, o)
+            super(hash, o).tap do |r|
+              if r[SHOVEL].is_a?(Hash)
+                h2 = {}
+                r.each do |k, v|
+                  if k == SHOVEL
+                    h2.merge!(v)
+                  else
+                    h2[k] = v
+                  end
+                end
+                r.replace(h2)
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/fluent/config/yaml_parser/parser.rb b/lib/fluent/config/yaml_parser/parser.rb
new file mode 100644
index 0000000000..dc1caaf46d
--- /dev/null
+++ b/lib/fluent/config/yaml_parser/parser.rb
@@ -0,0 +1,166 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'fluent/config/yaml_parser/section_builder'
+
+module Fluent
+  module Config
+    module YamlParser
+      class Parser
+        def initialize(config, indent: 2)
+          @base_indent = indent
+          @config = config
+        end
+
+        def build
+          s = @config['system'] && system_config_build(@config['system'])
+          c = @config['config'] && config_build(@config['config'], root: true)
+          RootBuilder.new(s, c)
+        end
+
+        private
+
+        def system_config_build(config)
+          section_build('system', config)
+        end
+
+        def config_build(config, indent: 0, root: false)
+          sb = SectionBodyBuilder.new(indent, root: root)
+          config.each do |c|
+            if (lc = c.delete('label'))
+              sb.add_section(label_build(lc, indent: indent))
+            end
+
+            if (sc = c.delete('source'))
+              sb.add_section(source_build(sc, indent: indent))
+            end
+
+            if (fc = c.delete('filter'))
+              sb.add_section(filter_build(fc, indent: indent))
+            end
+
+            if (mc = c.delete('match'))
+              sb.add_section(match_build(mc, indent: indent))
+            end
+
+            if (wc = c.delete('worker'))
+              sb.add_section(worker_build(wc, indent: indent))
+            end
+
+            included_sections_build(c, sb, indent: indent)
+          end
+
+          sb
+        end
+
+        def label_build(config, indent: 0)
+          config = config.dup
+          name = config.delete('$name')
+          c = config.delete('config')
+          SectionBuilder.new('label', config_build(c, indent: indent + @base_indent), indent, name)
+        end
+
+        def worker_build(config, indent: 0)
+          config = config.dup
+          num = config.delete('$arg')
+          c = config.delete('config')
+          SectionBuilder.new('worker', config_build(c, indent: indent + @base_indent), indent, num)
+        end
+
+        def source_build(config, indent: 0)
+          section_build('source', config, indent: indent)
+        end
+
+        def filter_build(config, indent: 0)
+          config = config.dup
+          tag = config.delete('$tag')
+          if tag.is_a?(Array)
+            section_build('filter', config, indent: indent, arg: tag&.join(','))
+          else
+            section_build('filter', config, indent: indent, arg: tag)
+          end
+        end
+
+        def match_build(config, indent: 0)
+          config = config.dup
+          tag = config.delete('$tag')
+          if tag.is_a?(Array)
+            section_build('match', config, indent: indent, arg: tag&.join(','))
+          else
+            section_build('match', config, indent: indent, arg: tag)
+          end
+        end
+
+        def included_sections_build(config, section_builder, indent: 0)
+          config.each_entry do |e|
+            k = e.keys.first
+            cc = e.delete(k)
+            case k
+            when 'label'
+              section_builder.add_section(label_build(cc, indent: indent))
+            when 'worker'
+              section_builder.add_section(worker_build(cc, indent: indent))
+            when 'source'
+              section_builder.add_section(source_build(cc, indent: indent))
+            when 'filter'
+              section_builder.add_section(filter_build(cc, indent: indent))
+            when 'match'
+              section_builder.add_section(match_build(cc, indent: indent))
+            end
+          end
+        end
+
+        def section_build(name, config, indent: 0, arg: nil)
+          sb = SectionBodyBuilder.new(indent + @base_indent)
+
+          if (v = config.delete('$type'))
+            sb.add_line('@type', v)
+          end
+
+          if (v = config.delete('$label'))
+            sb.add_line('@label', v)
+          end
+
+          if (v = config.delete('$id'))
+            sb.add_line('@id', v)
+          end
+
+          config.each do |key, val|
+            if val.is_a?(Array)
+              val.each do |v|
+                sb.add_section(section_build(key, v, indent: indent + @base_indent))
+              end
+            elsif val.is_a?(Hash)
+              harg = val.delete('$arg')
+              if harg.is_a?(Array)
+                # To prevent to generate invalid configuration for arg.
+                # "arg" should be String object and concatenated by ","
+                # when two or more objects are specified there.
+                sb.add_section(section_build(key, val, indent: indent + @base_indent, arg: harg&.join(',')))
+              else
+                sb.add_section(section_build(key, val, indent: indent + @base_indent, arg: harg))
+              end
+            else
+              sb.add_line(key, val)
+            end
+          end
+
+          SectionBuilder.new(name, sb, indent, arg)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/fluent/config/yaml_parser/section_builder.rb b/lib/fluent/config/yaml_parser/section_builder.rb
new file mode 100644
index 0000000000..58a89e6cd9
--- /dev/null
+++ b/lib/fluent/config/yaml_parser/section_builder.rb
@@ -0,0 +1,107 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+module Fluent
+  module Config
+    module YamlParser
+      SectionBuilder = Struct.new(:name, :body, :indent_size, :arg) do
+        def to_s
+          indent = ' ' * indent_size
+
+          if arg && !arg.to_s.empty?
+            "#{indent}<#{name} #{arg}>\n#{body}\n#{indent}</#{name}>"
+          else
+            "#{indent}<#{name}>\n#{body}\n#{indent}</#{name}>"
+          end
+        end
+
+        def to_element
+          elem = body.to_element
+          elem.name = name
+          elem.arg = arg.to_s if arg
+          elem.v1_config = true
+          elem
+        end
+      end
+
+      class RootBuilder
+        def initialize(system, conf)
+          @system = system
+          @conf = conf
+        end
+
+        attr_reader :system, :conf
+
+        def to_element
+          Fluent::Config::Element.new('ROOT', '', {}, [@system, @conf].compact.map(&:to_element).flatten)
+        end
+
+        def to_s
+          s = StringIO.new(+'')
+          s.puts(@system.to_s) if @system
+          s.puts(@conf.to_s) if @conf
+
+          s.string
+        end
+      end
+
+      class SectionBodyBuilder
+        Row = Struct.new(:key, :value, :indent) do
+          def to_s
+            "#{indent}#{key} #{value}"
+          end
+        end
+
+        def initialize(indent, root: false)
+          @indent = ' ' * indent
+          @bodies = []
+          @root = root
+        end
+
+        def add_line(k, v)
+          @bodies << Row.new(k, v, @indent)
+        end
+
+        def add_section(section)
+          @bodies << section
+        end
+
+        def to_element
+          if @root
+            return @bodies.map(&:to_element)
+          end
+
+          not_section, section = @bodies.partition { |e| e.is_a?(Row) }
+          r = {}
+          not_section.each do |e|
+            v = e.value
+            r[e.key] = v.respond_to?(:to_element) ? v.to_element : v
+          end
+
+          if @root
+            section.map(&:to_element)
+          else
+            Fluent::Config::Element.new('', '', r, section.map(&:to_element))
+          end
+        end
+
+        def to_s
+          @bodies.map(&:to_s).join("\n")
+        end
+      end
+    end
+  end
+end
diff --git a/lib/fluent/supervisor.rb b/lib/fluent/supervisor.rb
index 4c57048809..1df23053d6 100644
--- a/lib/fluent/supervisor.rb
+++ b/lib/fluent/supervisor.rb
@@ -580,7 +580,8 @@ def self.default_options
         standalone_worker: false,
         signame: nil,
         conf_encoding: 'utf-8',
-        disable_shared_socket: nil
+        disable_shared_socket: nil,
+        config_file_type: :guess,
       }
     end
 
@@ -593,6 +594,7 @@ def self.cleanup_resources
     end
 
     def initialize(opt)
+      @config_file_type = opt[:config_file_type]
       @daemonize = opt[:daemonize]
       @standalone_worker= opt[:standalone_worker]
       @config_path = opt[:config_path]
@@ -618,7 +620,9 @@ def initialize(opt)
         @conf = Fluent::Config.build(config_path: @config_path,
                                      encoding: @conf_encoding ? @conf_encoding : 'utf-8',
                                      additional_config: @inline_config ? @inline_config : nil,
-                                     use_v1_config: !!@use_v1_config)
+                                     use_v1_config: !!@use_v1_config,
+                                     type: @config_file_type,
+                                    )
         @system_config = build_system_config(@conf)
         if @system_config.log
           @log_rotate_age ||= @system_config.log.rotate_age
@@ -744,7 +748,13 @@ def configure(supervisor: false)
         $log.warn('the value "-" for `inline_config` is deprecated. See https://github.com/fluent/fluentd/issues/2711')
         @inline_config = STDIN.read
       end
-      @conf = Fluent::Config.build(config_path: @config_path, encoding: @conf_encoding, additional_config: @inline_config, use_v1_config: @use_v1_config)
+      @conf = Fluent::Config.build(
+        config_path: @config_path,
+        encoding: @conf_encoding,
+        additional_config: @inline_config,
+        use_v1_config: @use_v1_config,
+        type: @config_file_type,
+      )
       @system_config = build_system_config(@conf)
 
       @log.level = @system_config.log_level
@@ -929,6 +939,7 @@ def reload_config
             encoding: @conf_encoding,
             additional_config: @inline_config,
             use_v1_config: @use_v1_config,
+            type: @config_file_type,
           )
 
           Fluent::VariableStore.try_to_reset do
diff --git a/test/command/test_fluentd.rb b/test/command/test_fluentd.rb
index 643588ffdb..2ad023bf90 100644
--- a/test/command/test_fluentd.rb
+++ b/test/command/test_fluentd.rb
@@ -589,6 +589,39 @@ def assert_fluentd_fails_to_start(cmdline, *pattern_list, timeout: 10)
       )
     end
 
+    sub_test_case "YAML config format" do
+      test 'success to start the number of workers specified in configuration' do
+        conf = <<'CONF'
+        system:
+          workers: 2
+          root_dir: "#{@root_path}"
+        config:
+          - source:
+              $type: dummy
+              $id: !fluent/s "dummy.#{worker_id}" # check worker_id works or not with actual command
+              $label: '@dummydata'
+              tag: dummy
+              dummy: !fluent/json {"message": !fluent/s "yay from #{hostname}!"}
+
+          - label:
+              $name: '@dummydata'
+              config:
+               - match:
+                  $tag: dummy
+                  $type: "null"
+                  $id: blackhole
+CONF
+        conf_path = create_conf_file('workers1.yaml', conf)
+        assert Dir.exist?(@root_path)
+
+        assert_log_matches(
+          create_cmdline(conf_path),
+          "#0 fluentd worker is now running worker=0",
+          "#1 fluentd worker is now running worker=1"
+        )
+      end
+    end
+
     test 'success to start the number of workers specified by command line option' do
       conf = <<CONF
 <system>
diff --git a/test/test_config.rb b/test/test_config.rb
index e9d1005647..74b4dc5d7f 100644
--- a/test/test_config.rb
+++ b/test/test_config.rb
@@ -10,11 +10,18 @@ class ConfigTest < Test::Unit::TestCase
 
   TMP_DIR = File.dirname(__FILE__) + "/tmp/config#{ENV['TEST_ENV_NUMBER']}"
 
-  def read_config(path)
+  def read_config(path, use_yaml: false)
     path = File.expand_path(path)
-    File.open(path) { |io|
-      Fluent::Config::Parser.parse(io, File.basename(path), File.dirname(path))
-    }
+    if use_yaml
+      context = Kernel.binding
+
+      s = Fluent::Config::YamlParser::Loader.new(context).load(Pathname.new(path))
+      Fluent::Config::YamlParser::Parser.new(s).build.to_element
+    else
+      File.open(path) { |io|
+        Fluent::Config::Parser.parse(io, File.basename(path), File.dirname(path))
+      }
+    end
   end
 
   def prepare_config
@@ -151,6 +158,130 @@ def test_check_not_fetchd
     assert_equal before_size, match_conf.unused.size
   end
 
+  sub_test_case "yaml config" do
+    def test_included
+      write_config "#{TMP_DIR}/config_test_not_fetched.yaml", <<-EOS
+      config:
+        - source:
+            $type: dummy
+            tag: tag.dummy
+        - source:
+            $type: tcp
+            tag: tag.tcp
+            parse:
+              $arg:
+                - why.parse.section.doesnot.have.arg
+                - huh
+              $type: none
+        - match:
+            $tag: tag.*
+            $type: stdout
+            buffer:
+              $type: memory
+              flush_interval: 1s
+        - !include fluent-included.yaml
+      EOS
+      write_config "#{TMP_DIR}/fluent-included.yaml", <<-EOS
+      - label:
+          $name: '@FLUENT_LOG'
+          config:
+            - match:
+                $type: "null"
+                $tag: "**"
+                buffer:
+                  $type: memory
+                  flush_mode: interval
+                  flush_interval: 1s
+      EOS
+      root_conf  = read_config("#{TMP_DIR}/config_test_not_fetched.yaml", use_yaml: true)
+      dummy_source_conf = root_conf.elements.first
+      tcp_source_conf = root_conf.elements[1]
+      parse_tcp_conf = tcp_source_conf.elements.first
+      match_conf = root_conf.elements[2]
+      label_conf = root_conf.elements[3]
+      fluent_log_conf = label_conf.elements.first
+      fluent_log_buffer_conf = fluent_log_conf.elements.first
+
+      assert_equal(
+        [
+          'dummy',
+          'tag.dummy',
+          'tcp',
+          'tag.tcp',
+          'none',
+          'why.parse.section.doesnot.have.arg,huh',
+          'stdout',
+          'tag.*',
+          'null',
+          '**',
+          '@FLUENT_LOG',
+          'memory',
+          'interval',
+          '1s',
+        ],
+        [
+          dummy_source_conf['@type'],
+          dummy_source_conf['tag'],
+          tcp_source_conf['@type'],
+          tcp_source_conf['tag'],
+          parse_tcp_conf['@type'],
+          parse_tcp_conf.arg,
+          match_conf['@type'],
+          match_conf.arg,
+          fluent_log_conf['@type'],
+          fluent_log_conf.arg,
+          label_conf.arg,
+          fluent_log_buffer_conf['@type'],
+          fluent_log_buffer_conf['flush_mode'],
+          fluent_log_buffer_conf['flush_interval'],
+        ])
+    end
+
+    def test_check_not_fetchd
+      write_config "#{TMP_DIR}/config_test_not_fetched.yaml", <<-EOS
+      config:
+        - match:
+            $arg: dummy
+            $type: rewrite
+            add_prefix:    filtered
+            rule:
+              key:     path
+              pattern: "^[A-Z]+"
+              replace: true
+      EOS
+      root_conf  = read_config("#{TMP_DIR}/config_test_not_fetched.yaml", use_yaml: true)
+      match_conf = root_conf.elements.first
+      rule_conf  = match_conf.elements.first
+
+      not_fetched = []; root_conf.check_not_fetched {|key, e| not_fetched << key }
+      assert_equal %w[@type $arg add_prefix key pattern replace], not_fetched
+
+      not_fetched = []; match_conf.check_not_fetched {|key, e| not_fetched << key }
+      assert_equal %w[@type $arg add_prefix key pattern replace], not_fetched
+
+      not_fetched = []; rule_conf.check_not_fetched {|key, e| not_fetched << key }
+      assert_equal %w[key pattern replace], not_fetched
+
+      # accessing should delete
+      match_conf['type']
+      rule_conf['key']
+
+      not_fetched = []; root_conf.check_not_fetched {|key, e| not_fetched << key }
+      assert_equal %w[@type $arg add_prefix pattern replace], not_fetched
+
+      not_fetched = []; match_conf.check_not_fetched {|key, e| not_fetched << key }
+      assert_equal %w[@type $arg add_prefix pattern replace], not_fetched
+
+      not_fetched = []; rule_conf.check_not_fetched {|key, e| not_fetched << key }
+      assert_equal %w[pattern replace], not_fetched
+
+      # repeatedly accessing should not grow memory usage
+      before_size = match_conf.unused.size
+      10.times { match_conf['type'] }
+      assert_equal before_size, match_conf.unused.size
+    end
+  end
+
   def write_config(path, data, encoding: 'utf-8')
     FileUtils.mkdir_p(File.dirname(path))
     File.open(path, "w:#{encoding}:utf-8") {|f| f.write data }
diff --git a/test/test_supervisor.rb b/test/test_supervisor.rb
index 9b6ca82195..86159c2cb7 100644
--- a/test/test_supervisor.rb
+++ b/test/test_supervisor.rb
@@ -90,6 +90,93 @@ def test_system_config
     assert_equal 2, counter_client.timeout
   end
 
+  sub_test_case "yaml config" do
+    def parse_yaml(yaml)
+      context = Kernel.binding
+
+      config = nil
+      Tempfile.open do |file|
+        file.puts(yaml)
+        file.flush
+        s = Fluent::Config::YamlParser::Loader.new(context).load(Pathname.new(file))
+        config = Fluent::Config::YamlParser::Parser.new(s).build.to_element
+      end
+      config
+    end
+
+    def test_system_config
+      opts = Fluent::Supervisor.default_options
+      sv = Fluent::Supervisor.new(opts)
+      conf_data = <<-EOC
+      system:
+        rpc_endpoint: 127.0.0.1:24445
+        suppress_repeated_stacktrace: true
+        suppress_config_dump: true
+        without_source: true
+        enable_get_dump: true
+        process_name: "process_name"
+        log_level: info
+        root_dir: !fluent/s "#{TMP_ROOT_DIR}"
+        log:
+          format: json
+          time_format: "%Y"
+        counter_server:
+          bind: 127.0.0.1
+          port: 24321
+          scope: server1
+          backup_path: /tmp/backup
+        counter_client:
+          host: 127.0.0.1
+          port: 24321
+          timeout: 2
+      EOC
+      conf = parse_yaml(conf_data)
+      sys_conf = sv.__send__(:build_system_config, conf)
+
+    counter_client = sys_conf.counter_client
+    counter_server = sys_conf.counter_server
+    assert_equal(
+      [
+        '127.0.0.1:24445',
+        true,
+        true,
+        true,
+        true,
+        "process_name",
+        2,
+        TMP_ROOT_DIR,
+        :json,
+        '%Y',
+        '127.0.0.1',
+        24321,
+        'server1',
+        '/tmp/backup',
+        '127.0.0.1',
+        24321,
+        2,
+      ],
+      [
+        sys_conf.rpc_endpoint,
+        sys_conf.suppress_repeated_stacktrace,
+        sys_conf.suppress_config_dump,
+        sys_conf.without_source,
+        sys_conf.enable_get_dump,
+        sys_conf.process_name,
+        sys_conf.log_level,
+        sys_conf.root_dir,
+        sys_conf.log.format,
+        sys_conf.log.time_format,
+        counter_server.bind,
+        counter_server.port,
+        counter_server.scope,
+        counter_server.backup_path,
+        counter_client.host,
+        counter_client.port,
+        counter_client.timeout,
+      ])
+    end
+  end
+
   def test_main_process_signal_handlers
     omit "Windows cannot handle signals" if Fluent.windows?
 
@@ -557,6 +644,31 @@ def test_override_default_log_rotate
                       logger.instance_variable_get(:@rotate_size)])
       end
     end
+
+    def test_override_default_log_rotate_with_yaml_config
+      Tempfile.open do |file|
+        config = <<-EOS
+          system:
+            log:
+              rotate_age: 3
+              rotate_size: 300
+        EOS
+        file.puts(config)
+        file.flush
+        opts = Fluent::Supervisor.default_options.merge(
+          log_path: "#{TMP_DIR}/test.log", config_path: file.path, config_file_type: :yaml,
+        )
+        sv = Fluent::Supervisor.new(opts)
+
+        log = sv.instance_variable_get(:@log)
+        log.init(:standalone, 0)
+        logger = $log.instance_variable_get(:@logger)
+
+        assert_equal([3, 300],
+                     [logger.instance_variable_get(:@rotate_age),
+                      logger.instance_variable_get(:@rotate_size)])
+      end
+    end
   end
 
   def test_inline_config