diff --git a/lib/fluent/config.rb b/lib/fluent/config.rb
index 4c321ff9bd..b2411853e1 100644
--- a/lib/fluent/config.rb
+++ b/lib/fluent/config.rb
@@ -20,6 +20,25 @@
 
 module Fluent
   module Config
+    # @param config_path [String] config file path
+    # @param encoding [String] encoding of config file
+    # @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)
+      config_fname = File.basename(config_path)
+      config_basedir = File.dirname(config_path)
+      config_data = File.open(config_path, "r:#{encoding}:utf-8") do |f|
+        s = f.read
+        if additional_config
+          c = additional_config.gsub("\\n", "\n")
+          s += "\n#{c}"
+        end
+        s
+      end
+      Fluent::Config.parse(config_data, config_fname, config_basedir, use_v1_config)
+    end
+
     def self.parse(str, fname, basepath = Dir.pwd, v1_config = nil, syntax: :v1)
       parser = if fname =~ /\.rb$/ || syntax == :ruby
                  :ruby
diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb
index b03ee0fb9d..c4202c41d8 100644
--- a/lib/fluent/engine.rb
+++ b/lib/fluent/engine.rb
@@ -23,6 +23,7 @@
 require 'fluent/system_config'
 require 'fluent/plugin'
 require 'fluent/fluent_log_event_router'
+require 'fluent/static_config_analysis'
 
 module Fluent
   class EngineClass
@@ -157,14 +158,47 @@ def run
         raise
       end
 
-      unless @log_event_verbose
-        $log.enable_event(false)
-        @fluent_log_event_router.graceful_stop
+      stop_phase(@root_agent)
+    end
+
+    # @param conf [Fluent::Config]
+    # @param supervisor [Bool]
+    # @reutrn nil
+    def reload_config(conf, supervisor: false)
+      # configure first to reduce down time while restarting
+      new_agent = RootAgent.new(log: log, system_config: @system_config)
+      ret = Fluent::StaticConfigAnalysis.call(conf, workers: system_config.workers)
+
+      ret.all_plugins.each do |plugin|
+        if plugin.respond_to?(:reloadable_plugin?) && !plugin.reloadable_plugin?
+          raise Fluent::ConfigError, "Unreloadable plugin plugin: #{Fluent::Plugin.lookup_type_from_class(plugin.class)}, plugin_id: #{plugin.plugin_id}, class_name: #{plugin.class})"
+        end
       end
-      $log.info "shutting down fluentd worker", worker: worker_id
-      shutdown
 
-      @fluent_log_event_router.stop
+      # Assign @root_agent to new root_agent
+      # for https://github.com/fluent/fluentd/blob/fcef949ce40472547fde295ddd2cfe297e1eddd6/lib/fluent/plugin_helper/event_emitter.rb#L50
+      old_agent, @root_agent = @root_agent, new_agent
+      begin
+        @root_agent.configure(conf)
+      rescue
+        @root_agent = old_agent
+        raise
+      end
+
+      unless @suppress_config_dump
+        $log.info :supervisor, "using configuration file: #{conf.to_s.rstrip}"
+      end
+
+      # supervisor doesn't handle actual data. so the following code is unnecessary.
+      if supervisor
+        old_agent.shutdown      # to close thread created in #configure
+        return
+      end
+
+      stop_phase(old_agent)
+
+      $log.info 'restart fluentd worker', worker: worker_id
+      start_phase(new_agent)
     end
 
     def stop
@@ -189,12 +223,29 @@ def worker_id
     end
 
     private
-    def start
+
+    def stop_phase(root_agent)
+      unless @log_event_verbose
+        $log.enable_event(false)
+        @fluent_log_event_router.graceful_stop
+      end
+      $log.info 'shutting down fluentd worker', worker: worker_id
+      root_agent.shutdown
+
+      @fluent_log_event_router.stop
+    end
+
+    def start_phase(root_agent)
+      @fluent_log_event_router = FluentLogEventRouter.build(root_agent)
+      if @fluent_log_event_router.emittable?
+        $log.enable_event(true)
+      end
+
       @root_agent.start
     end
 
-    def shutdown
-      @root_agent.shutdown
+    def start
+      @root_agent.start
     end
   end
 
diff --git a/lib/fluent/plugin/base.rb b/lib/fluent/plugin/base.rb
index e8e444f33d..64ec5d0967 100644
--- a/lib/fluent/plugin/base.rb
+++ b/lib/fluent/plugin/base.rb
@@ -187,6 +187,11 @@ def inspect
         #   https://github.com/ruby/ruby/blob/trunk/gc.c#L788
         "#<%s:%014x>" % [self.class.name, '0x%014x' % (__id__ << 1)]
       end
+
+      def reloadable_plugin?
+        # Engine can't capture all class variables. so it's forbbiden to use class variables in each plugins if enabling reload.
+        self.class.class_variables.empty?
+      end
     end
   end
 end
diff --git a/lib/fluent/plugin/buf_file.rb b/lib/fluent/plugin/buf_file.rb
index 459b183ce2..75e93dc8ec 100644
--- a/lib/fluent/plugin/buf_file.rb
+++ b/lib/fluent/plugin/buf_file.rb
@@ -19,6 +19,7 @@
 require 'fluent/plugin/buffer'
 require 'fluent/plugin/buffer/file_chunk'
 require 'fluent/system_config'
+require 'fluent/variable_store'
 
 module Fluent
   module Plugin
@@ -43,19 +44,20 @@ class FileBuffer < Fluent::Plugin::Buffer
       config_param :file_permission, :string, default: nil # '0644'
       config_param :dir_permission,  :string, default: nil # '0755'
 
-      @@buffer_paths = {}
-
       def initialize
         super
         @symlink_path = nil
         @multi_workers_available = false
         @additional_resume_path = nil
         @buffer_path = nil
+        @variable_store = nil
       end
 
       def configure(conf)
         super
 
+        @variable_store = Fluent::VariableStore.fetch_or_build(:buf_file)
+
         multi_workers_configured = owner.system_config.workers > 1 ? true : false
 
         using_plugin_root_dir = false
@@ -69,13 +71,13 @@ def configure(conf)
         end
 
         type_of_owner = Plugin.lookup_type_from_class(@_owner.class)
-        if @@buffer_paths.has_key?(@path) && !called_in_test?
-          type_using_this_path = @@buffer_paths[@path]
+        if @variable_store.has_key?(@path) && !called_in_test?
+          type_using_this_path = @variable_store[@path]
           raise ConfigError, "Other '#{type_using_this_path}' plugin already use same buffer path: type = #{type_of_owner}, buffer path = #{@path}"
         end
 
         @buffer_path = @path
-        @@buffer_paths[@buffer_path] = type_of_owner
+        @variable_store[@buffer_path] = type_of_owner
 
         specified_directory_exists = File.exist?(@path) && File.directory?(@path)
         unexisting_path_for_directory = !File.exist?(@path) && !@path.include?('.*')
@@ -125,7 +127,9 @@ def start
       end
 
       def stop
-        @@buffer_paths.delete(@buffer_path)
+        if @variable_store
+          @variable_store.delete(@buffer_path)
+        end
 
         super
       end
diff --git a/lib/fluent/plugin/buf_file_single.rb b/lib/fluent/plugin/buf_file_single.rb
index 7116c951d8..e15d74a90a 100644
--- a/lib/fluent/plugin/buf_file_single.rb
+++ b/lib/fluent/plugin/buf_file_single.rb
@@ -19,6 +19,7 @@
 require 'fluent/plugin/buffer'
 require 'fluent/plugin/buffer/file_single_chunk'
 require 'fluent/system_config'
+require 'fluent/variable_store'
 
 module Fluent
   module Plugin
@@ -48,18 +49,19 @@ class FileSingleBuffer < Fluent::Plugin::Buffer
       desc 'The permission of chunk directory. If no specified, <system> setting or 0755 is used'
       config_param :dir_permission, :string, default: nil
 
-      @@buffer_paths = {}
-
       def initialize
         super
 
         @multi_workers_available = false
         @additional_resume_path = nil
+        @variable_store = nil
       end
 
       def configure(conf)
         super
 
+        @variable_store = Fluent::VariableStore.fetch_or_build(:buf_file_single)
+
         if @chunk_format == :auto
           @chunk_format = owner.formatted_to_msgpack_binary? ? :msgpack : :text
         end
@@ -117,12 +119,12 @@ def configure(conf)
         end
 
         type_of_owner = Plugin.lookup_type_from_class(@_owner.class)
-        if @@buffer_paths.has_key?(@path) && !called_in_test?
-          type_using_this_path = @@buffer_paths[@path]
+        if @variable_store.has_key?(@path) && !called_in_test?
+          type_using_this_path = @variable_store[@path]
           raise Fluent::ConfigError, "Other '#{type_using_this_path}' plugin already uses same buffer path: type = #{type_of_owner}, buffer path = #{@path}"
         end
 
-        @@buffer_paths[@path] = type_of_owner
+        @variable_store[@path] = type_of_owner
         @dir_permission = if @dir_permission
                             @dir_permission.to_i(8)
                           else
@@ -145,7 +147,9 @@ def start
       end
 
       def stop
-        @@buffer_paths.delete(@path)
+        if @variable_store
+          @variable_store.delete(@path)
+        end
 
         super
       end
diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb
index 6a226a544d..f8bb655700 100644
--- a/lib/fluent/plugin/in_tail.rb
+++ b/lib/fluent/plugin/in_tail.rb
@@ -21,6 +21,7 @@
 require 'fluent/event'
 require 'fluent/plugin/buffer'
 require 'fluent/plugin/parser_multiline'
+require 'fluent/variable_store'
 
 if Fluent.windows?
   require_relative 'file_wrapper'
@@ -103,9 +104,8 @@ def initialize
 
     attr_reader :paths
 
-    @@pos_file_paths = {}
-
     def configure(conf)
+      @variable_store = Fluent::VariableStore.fetch_or_build(:in_tail)
       compat_parameters_convert(conf, :parser)
       parser_config = conf.elements('parse').first
       unless parser_config
@@ -132,11 +132,11 @@ def configure(conf)
 
       # TODO: Use plugin_root_dir and storage plugin to store positions if available
       if @pos_file
-        if @@pos_file_paths.has_key?(@pos_file) && !called_in_test?
-          plugin_id_using_this_path = @@pos_file_paths[@pos_file]
+        if @variable_store.key?(@pos_file) && !called_in_test?
+          plugin_id_using_this_path = @variable_store[@pos_file]
           raise Fluent::ConfigError, "Other 'in_tail' plugin already use same pos_file path: plugin_id = #{plugin_id_using_this_path}, pos_file path = #{@pos_file}"
         end
-        @@pos_file_paths[@pos_file] = self.plugin_id
+        @variable_store[@pos_file] = self.plugin_id
       else
         $log.warn "'pos_file PATH' parameter is not set to a 'tail' source."
         $log.warn "this parameter is highly recommended to save the position to resume tailing."
@@ -205,7 +205,9 @@ def start
     end
 
     def stop
-      @@pos_file_paths.delete(@pos_file)
+      if @variable_store
+        @variable_store.delete(@pos_file)
+      end
 
       super
     end
diff --git a/lib/fluent/plugin_id.rb b/lib/fluent/plugin_id.rb
index 0441387ad5..1a4fe4065d 100644
--- a/lib/fluent/plugin_id.rb
+++ b/lib/fluent/plugin_id.rb
@@ -15,26 +15,29 @@
 #
 
 require 'set'
+require 'fluent/variable_store'
 
 module Fluent
   module PluginId
-    @@configured_ids = Set.new
 
     def initialize
       super
+
+      @_plugin_id_variable_store = nil
       @_plugin_root_dir = nil
       @id = nil
     end
 
     def configure(conf)
+      @_plugin_id_variable_store = Fluent::VariableStore.fetch_or_build(:pluing_id, default_value: Set.new)
       @id = conf['@id']
       @_id_configured = !!@id # plugin id is explicitly configured by users (or not)
       if @id
         @id = @id.to_s
-        if @@configured_ids.include?(@id) && !plugin_id_for_test?
+        if @_plugin_id_variable_store.include?(@id) && !plugin_id_for_test?
           raise Fluent::ConfigError, "Duplicated plugin id `#{@id}`. Check whole configuration and fix it."
         end
-        @@configured_ids.add(@id)
+        @_plugin_id_variable_store.add(@id)
       end
 
       super
@@ -79,7 +82,9 @@ def plugin_root_dir
     end
 
     def stop
-      @@configured_ids.delete(@id)
+      if @_plugin_id_variable_store
+        @_plugin_id_variable_store.delete(@id)
+      end
 
       super
     end
diff --git a/lib/fluent/static_config_analysis.rb b/lib/fluent/static_config_analysis.rb
new file mode 100644
index 0000000000..5aabdba970
--- /dev/null
+++ b/lib/fluent/static_config_analysis.rb
@@ -0,0 +1,194 @@
+#
+# 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'
+require 'fluent/plugin'
+
+module Fluent
+  # Static Analysis means analysing all plugins and Fluent::Element without invokeing Plugin#configure
+  class StaticConfigAnalysis
+    module Elem
+      Input = Struct.new(:plugin, :config)
+      Output = Struct.new(:plugin, :config)
+      Filter = Struct.new(:plugin, :config)
+      Label = Struct.new(:name, :config, :nodes)
+      Worker = Struct.new(:ids, :config, :nodes)
+    end
+
+    Result = Struct.new(:tree, :outputs, :inputs, :filters, :labels) do
+      def all_plugins
+        (outputs + inputs + filters).map(&:plugin)
+      end
+    end
+
+    # @param workers [Integer] Number of workers
+    # @return [Fluent::StaticConfigAnalysis::Result]
+    def self.call(conf, workers: 1)
+      new(workers).call(conf)
+    end
+
+    def initialize(workers)
+      @workers = workers
+
+      reset
+    end
+
+    def call(config)
+      reset
+
+      tree = [
+        static_worker_analyse(config),
+        static_label_analyse(config),
+        static_filter_and_output_analyse(config),
+        static_input_analyse(config),
+      ].flatten
+
+      Result.new(tree, @outputs, @inputs, @filters, @labels.values)
+    end
+
+    private
+
+    def reset
+      @outputs = []
+      @inputs = []
+      @filters = []
+      @labels = {}
+    end
+
+    def static_worker_analyse(conf)
+      available_worker_ids = [*0...@workers]
+
+      ret = []
+      conf.elements(name: 'worker').each do |config|
+        ids = parse_worker_id(config)
+        ids.each do |id|
+          if available_worker_ids.include?(id)
+            available_worker_ids.delete(id)
+          else
+            raise Fluent::ConfigError, "specified worker_id<#{id}> collisions is detected on <worker> directive. Available worker id(s): #{available_worker_ids}"
+          end
+        end
+
+        config.elements.each do |elem|
+          unless %w[source match filter label].include?(elem.name)
+            raise Fluent::ConfigError, "<worker> section cannot have <#{elem.name}> directive"
+          end
+        end
+
+        nodes = [
+          static_label_analyse(config),
+          static_filter_and_output_analyse(config),
+          static_input_analyse(config),
+        ].flatten
+        ret << Elem::Worker.new(ids, config, nodes)
+      end
+
+      ret
+    end
+
+    def parse_worker_id(conf)
+      worker_id_str = conf.arg
+
+      if worker_id_str.empty?
+        raise Fluent::ConfigError, 'Missing worker id on <worker> directive'
+      end
+
+      l, r =
+         begin
+           worker_id_str.split('-', 2).map { |v| Integer(v) }
+         rescue TypeError, ArgumentError
+           raise Fluent::ConfigError, "worker id should be integer: #{worker_id_str}"
+         end
+
+      if l < 0 || l >= @workers
+        raise Fluent::ConfigError, "worker id #{l} specified by <worker> directive is not allowed. Available worker id is between 0 and #{@workers-1}"
+      end
+
+      # e.g. specified one worker id like `<worker 0>`
+      if r.nil?
+        return [l]
+      end
+
+      if r < 0 || r >= @workers
+        raise Fluent::ConfigError, "worker id #{r} specified by <worker> directive is not allowed. Available worker id is between 0 and #{@workers-1}"
+      end
+
+      if l > r
+        raise Fluent::ConfigError, "greater first_worker_id<#{l}> than last_worker_id<#{r}> specified by <worker> directive is not allowed. Available multi worker assign syntax is <smaller_worker_id>-<greater_worker_id>"
+      end
+
+      [l, r]
+    end
+
+    def static_label_analyse(conf)
+      ret = []
+      conf.elements(name: 'label').each do |e|
+        name = e.arg
+        if name.empty?
+          raise ConfigError, 'Missing symbol argument on <label> directive'
+        end
+
+        if @labels[name]
+          raise ConfigError, "Section <label #{name}> appears twice"
+        end
+
+        l = Elem::Label.new(name, e, static_filter_and_output_analyse(e))
+        ret << l
+        @labels[name] = l
+      end
+
+      ret
+    end
+
+    def static_filter_and_output_analyse(conf)
+      ret = []
+      conf.elements('filter', 'match').each do |e|
+        type = e['@type']
+        if type.nil? || type.empty?
+          raise Fluent::ConfigError, "Missing '@type' parameter on <#{e.name}> directive"
+        end
+
+        if e.name == 'filter'
+          f = Elem::Filter.new(Fluent::Plugin.new_filter(type), e)
+          ret << f
+          @filters << f
+        else
+          o = Elem::Output.new(Fluent::Plugin.new_output(type), e)
+          ret << o
+          @outputs << o
+        end
+      end
+
+      ret
+    end
+
+    def static_input_analyse(conf)
+      ret = []
+      conf.elements(name: 'source').each do |e|
+        type = e['@type']
+        if type.nil? || type.empty?
+          raise Fluent::ConfigError, "Missing '@type' parameter on <#{e.name}> directive"
+        end
+
+        i = Elem::Input.new(Fluent::Plugin.new_input(type), e)
+        @inputs << i
+        ret << i
+      end
+
+      ret
+    end
+  end
+end
diff --git a/lib/fluent/supervisor.rb b/lib/fluent/supervisor.rb
index 1c34256a9e..aced0cc3b7 100644
--- a/lib/fluent/supervisor.rb
+++ b/lib/fluent/supervisor.rb
@@ -26,6 +26,7 @@
 require 'fluent/rpc'
 require 'fluent/system_config'
 require 'fluent/msgpack_factory'
+require 'fluent/variable_store'
 require 'serverengine'
 
 if Fluent.windows?
@@ -42,6 +43,7 @@
 module Fluent
   module ServerModule
     def before_run
+      @fluentd_conf = config[:fluentd_conf]
       @rpc_server = nil
       @counter = nil
 
@@ -120,6 +122,15 @@ def run_rpc_server
         nil
       }
 
+      @rpc_server.mount_proc('/api/config.gracefulReload') { |req, res|
+        $log.debug "fluentd RPC got /api/config.gracefulReload request"
+        unless Fluent.windows?
+          Process.kill :USR2, $$
+        end
+
+        nil
+      }
+
       @rpc_server.mount_proc('/api/config.getDump') { |req, res|
         $log.debug "fluentd RPC got /api/config.getDump request"
         $log.info "get dump in-memory config via HTTP"
@@ -156,6 +167,11 @@ def install_supervisor_signal_handlers
         $log.debug "fluentd supervisor process get SIGUSR1"
         supervisor_sigusr1_handler
       end unless Fluent.windows?
+
+      trap :USR2 do
+        $log.debug 'fluentd supervisor process got SIGUSR2'
+        supervisor_sigusr2_handler
+      end unless Fluent.windows?
     end
 
     def install_windows_event_handler
@@ -179,20 +195,36 @@ def supervisor_sighup_handler
     end
 
     def supervisor_sigusr1_handler
-      if log = config[:logger_initializer]
-        # Creating new thread due to mutex can't lock
-        # in main thread during trap context
-        Thread.new do
-          log.reopen!
-        end
-      end
+      reopen_log
+      send_signal_to_workers(:USR1)
+    end
 
-      if config[:worker_pid]
-        config[:worker_pid].each_value do |pid|
-          Process.kill(:USR1, pid)
-          # don't rescue Errno::ESRCH here (invalid status)
+    def supervisor_sigusr2_handler
+      conf = nil
+      t = Thread.new do
+        $log.info 'Reloading new config'
+
+        # Validate that loading config is valid at first
+        conf = Fluent::Config.build(
+          config_path: config[:config_path],
+          encoding: config[:conf_encoding],
+          additional_config: config[:inline_config],
+          use_v1_config: config[:use_v1_config],
+        )
+
+        Fluent::VariableStore.try_to_reset do
+          Fluent::Engine.reload_config(conf, supervisor: true)
         end
       end
+
+      t.report_on_exception = false # Error is handled by myself
+      t.join
+
+      reopen_log
+      send_signal_to_workers(:USR2)
+      @fluentd_conf = conf.to_s
+    rescue => e
+      $log.error "Failed to reload config file: #{e}"
     end
 
     def kill_worker
@@ -210,11 +242,32 @@ def kill_worker
     end
 
     def supervisor_dump_config_handler
-      $log.info config[:fluentd_conf]
+      $log.info @fluentd_conf
     end
 
     def supervisor_get_dump_config_handler
-      {conf: config[:fluentd_conf]}
+      { conf: @fluentd_conf }
+    end
+
+    private
+
+    def reopen_log
+      if (log = config[:logger_initializer])
+        # Creating new thread due to mutex can't lock
+        # in main thread during trap context
+        Thread.new do
+          log.reopen!
+        end
+      end
+    end
+
+    def send_signal_to_workers(signal)
+      return unless config[:worker_pid]
+
+      config[:worker_pid].each_value do |pid|
+        # don't rescue Errno::ESRCH here (invalid status)
+        Process.kill(signal, pid)
+      end
     end
   end
 
@@ -302,6 +355,9 @@ def self.load_config(path, params = {})
                                  JSON.dump(params)],
         command_sender: command_sender,
         fluentd_conf: params['fluentd_conf'],
+        conf_encoding: params['conf_encoding'],
+        inline_config: params['inline_config'],
+        config_path: path,
         main_cmd: params['main_cmd'],
         signame: params['signame'],
       }
@@ -557,8 +613,7 @@ def configure(supervisor: false)
       if @inline_config == '-'
         @inline_config = STDIN.read
       end
-
-      @conf = read_config
+      @conf = Fluent::Config.build(config_path: @config_path, encoding: @conf_encoding, additional_config: @inline_config, use_v1_config: @use_v1_config)
       @system_config = build_system_config(@conf)
 
       @log.level = @system_config.log_level
@@ -673,6 +728,10 @@ def install_main_process_signal_handlers
         flush_buffer
       end unless Fluent.windows?
 
+      trap :USR2 do
+        reload_config
+      end unless Fluent.windows?
+
       if Fluent.windows?
         command_pipe = STDIN.dup
         STDIN.reopen(File::NULL, "rb")
@@ -713,6 +772,32 @@ def flush_buffer
       end
     end
 
+    def reload_config
+      Thread.new do
+        $log.debug('worker got SIGUSR2')
+
+        begin
+          conf = Fluent::Config.build(
+            config_path: @config_path,
+            encoding: @conf_encoding,
+            additional_config: @inline_config,
+            use_v1_config: @use_v1_config,
+          )
+
+          Fluent::VariableStore.try_to_reset do
+            Fluent::Engine.reload_config(conf)
+          end
+        rescue => e
+          # it is guranteed that config file is valid by supervisor side. but it's not atomic becuase of using signals to commnicate between worker and super
+          # So need this rescue code
+          $log.error("failed to reload config: #{e}")
+          next
+        end
+
+        @conf = conf
+      end
+    end
+
     def logging_with_console_output
       yield $log
       unless @log.stdout?
@@ -769,16 +854,6 @@ def main_process(&block)
       exit!(unrecoverable_error ? 2 : 1)
     end
 
-    def read_config
-      config_fname = File.basename(@config_path)
-      config_basedir = File.dirname(@config_path)
-      config_data = File.open(@config_path, "r:#{@conf_encoding}:utf-8") {|f| f.read }
-      if @inline_config
-        config_data << "\n" << @inline_config.gsub("\\n", "\n")
-      end
-      Fluent::Config.parse(config_data, config_fname, config_basedir, @use_v1_config)
-    end
-
     def build_system_config(conf)
       system_config = SystemConfig.create(conf, @cl_opt[:strict_config_value])
       opt = {}
diff --git a/lib/fluent/variable_store.rb b/lib/fluent/variable_store.rb
new file mode 100644
index 0000000000..ec6a6dadff
--- /dev/null
+++ b/lib/fluent/variable_store.rb
@@ -0,0 +1,40 @@
+#
+# 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
+  # VariableStore provides all pluigns with the way to shared variable without using class variable
+  # it's for safe reloading mechanism
+  class VariableStore
+    @data = {}
+
+    class << self
+      def fetch_or_build(namespace, default_value: {})
+        @data[namespace] ||= default_value
+      end
+
+      def try_to_reset
+        @data, old = {}, @data
+
+        begin
+          yield
+        rescue
+          @data = old
+          raise
+        end
+      end
+    end
+  end
+end
diff --git a/test/test_config.rb b/test/test_config.rb
index 2382f7d8bc..1efb79edd2 100644
--- a/test/test_config.rb
+++ b/test/test_config.rb
@@ -151,16 +151,16 @@ def test_check_not_fetchd
     assert_equal before_size, match_conf.unused.size
   end
 
-  def write_config(path, data)
+  def write_config(path, data, encoding: 'utf-8')
     FileUtils.mkdir_p(File.dirname(path))
-    File.open(path, "w") {|f| f.write data }
+    File.open(path, "w:#{encoding}:utf-8") {|f| f.write data }
   end
 
   def test_inline
     prepare_config
     opts = {
-        :config_path => "#{TMP_DIR}/config_test_1.conf",
-        :inline_config => "<source>\n  type http\n  port 2222\n </source>"
+      :config_path => "#{TMP_DIR}/config_test_1.conf",
+      :inline_config => "<source>\n  type http\n  port 2222\n </source>"
     }
     assert_nothing_raised do
       Fluent::Supervisor.new(opts)
@@ -175,5 +175,27 @@ def create_warn_dummy_logger
     logger = ServerEngine::DaemonLogger.new(logdev, dl_opts)
     $log = Fluent::Log.new(logger)
   end
-end
 
+  sub_test_case '.build' do
+    test 'read config' do
+      write_config("#{TMP_DIR}/build/config_build.conf", 'key value')
+      c = Fluent::Config.build(config_path: "#{TMP_DIR}/build/config_build.conf")
+      assert_equal('value', c['key'])
+    end
+
+    test 'read config with encoding' do
+      write_config("#{TMP_DIR}/build/config_build2.conf", "#てすと\nkey value", encoding: 'shift_jis')
+
+      c = Fluent::Config.build(config_path: "#{TMP_DIR}/build/config_build2.conf", encoding: 'shift_jis')
+      assert_equal('value', c['key'])
+    end
+
+    test 'read config with additional_config' do
+      write_config("#{TMP_DIR}/build/config_build2.conf", "key value")
+
+      c = Fluent::Config.build(config_path: "#{TMP_DIR}/build/config_build2.conf", additional_config: 'key2 value2')
+      assert_equal('value', c['key'])
+      assert_equal('value2', c['key2'])
+    end
+  end
+end
diff --git a/test/test_engine.rb b/test/test_engine.rb
new file mode 100644
index 0000000000..a28135341c
--- /dev/null
+++ b/test/test_engine.rb
@@ -0,0 +1,203 @@
+require_relative 'helper'
+require 'fluent/engine'
+require 'fluent/config'
+require 'fluent/input'
+require 'fluent/system_config'
+
+class EngineTest < ::Test::Unit::TestCase
+  class DummyEngineTestOutput < Fluent::Plugin::Output
+    Fluent::Plugin.register_output('dummy_engine_test', self)
+    def write(chunk); end
+  end
+
+  class DummyEngineTest2Output < Fluent::Plugin::Output
+    Fluent::Plugin.register_output('dummy_engine_test2', self)
+    def write(chunk); end
+  end
+
+  class DummyEngineTestInput < Fluent::Plugin::Input
+    Fluent::Plugin.register_input('dummy_engine_test', self)
+    def multi_workers_ready?; true; end
+  end
+
+  class DummyEngineTest2Input < Fluent::Plugin::Input
+    Fluent::Plugin.register_input('dummy_engine_test2', self)
+    def multi_workers_ready?; true; end
+  end
+
+  class DummyEngineClassVarTestInput < Fluent::Plugin::Input
+    Fluent::Plugin.register_input('dummy_engine_class_var_test', self)
+    @@test = nil
+    def multi_workers_ready?; true; end
+  end
+
+  sub_test_case '#reload_config' do
+    test 'reload new configuration' do
+      conf_data = <<-CONF
+        <source>
+          @type dummy_engine_test
+        </source>
+        <match>
+          @type dummy_engine_test
+        </match>
+      CONF
+
+      conf = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+      system_config = Fluent::SystemConfig.create(conf)
+
+      engine = Fluent::EngineClass.new
+      engine.init(system_config)
+      engine.configure(conf)
+
+      assert_kind_of DummyEngineTestInput, engine.root_agent.inputs[0]
+      assert_kind_of DummyEngineTestOutput, engine.root_agent.outputs[0]
+
+      new_conf_data = <<-CONF
+        <source>
+          @type dummy_engine_test2
+        </source>
+        <match>
+          @type dummy_engine_test2
+        </match>
+      CONF
+
+      new_conf = Fluent::Config.parse(new_conf_data, '(test)', '(test_dir)', true)
+
+      agent = Fluent::RootAgent.new(log: $log, system_config: system_config)
+      stub(Fluent::RootAgent).new do
+        stub(agent).start.once
+        agent
+      end
+
+      engine.reload_config(new_conf)
+
+      assert_kind_of DummyEngineTest2Input, engine.root_agent.inputs[0]
+      assert_kind_of DummyEngineTest2Output, engine.root_agent.outputs[0]
+    end
+
+    test "doesn't start RootAgent when supervisor is true" do
+      conf_data = <<-CONF
+        <source>
+          @type dummy_engine_test
+        </source>
+        <match>
+          @type dummy_engine_test
+        </match>
+      CONF
+
+      conf = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+      system_config = Fluent::SystemConfig.create(conf)
+
+      engine = Fluent::EngineClass.new
+      engine.init(system_config)
+      engine.configure(conf)
+
+      assert_kind_of DummyEngineTestInput, engine.root_agent.inputs[0]
+      assert_kind_of DummyEngineTestOutput, engine.root_agent.outputs[0]
+
+      new_conf_data = <<-CONF
+        <source>
+          @type dummy_engine_test2
+        </source>
+        <match>
+          @type dummy_engine_test2
+        </match>
+      CONF
+
+      new_conf = Fluent::Config.parse(new_conf_data, '(test)', '(test_dir)', true)
+
+      agent = Fluent::RootAgent.new(log: $log, system_config: system_config)
+      stub(Fluent::RootAgent).new do
+        stub(agent).start.never
+        agent
+      end
+
+      engine.reload_config(new_conf, supervisor: true)
+
+      assert_kind_of DummyEngineTest2Input, engine.root_agent.inputs[0]
+      assert_kind_of DummyEngineTest2Output, engine.root_agent.outputs[0]
+    end
+
+    test 'raise an error when conf is invalid' do
+      conf_data = <<-CONF
+        <source>
+          @type dummy_engine_test
+        </source>
+        <match>
+          @type dummy_engine_test
+        </match>
+      CONF
+
+      conf = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+      system_config = Fluent::SystemConfig.create(conf)
+
+      engine = Fluent::EngineClass.new
+      engine.init(system_config)
+      engine.configure(conf)
+
+      assert_kind_of DummyEngineTestInput, engine.root_agent.inputs[0]
+      assert_kind_of DummyEngineTestOutput, engine.root_agent.outputs[0]
+
+      new_conf_data = <<-CONF
+        <source>
+          @type
+        </source>
+      CONF
+
+      new_conf = Fluent::Config.parse(new_conf_data, '(test)', '(test_dir)', true)
+
+      agent = Fluent::RootAgent.new(log: $log, system_config: system_config)
+      stub(Fluent::RootAgent).new do
+        stub(agent).start.never
+        agent
+      end
+
+      assert_raise(Fluent::ConfigError.new("Missing '@type' parameter on <source> directive")) do
+        engine.reload_config(new_conf)
+      end
+
+      assert_kind_of DummyEngineTestInput, engine.root_agent.inputs[0]
+      assert_kind_of DummyEngineTestOutput, engine.root_agent.outputs[0]
+    end
+
+    test 'raise an error when unreloadable exists' do
+      conf_data = <<-CONF
+        <source>
+          @type dummy_engine_test
+        </source>
+        <match>
+          @type dummy_engine_test
+        </match>
+      CONF
+
+      conf = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+      system_config = Fluent::SystemConfig.create(conf)
+
+      engine = Fluent::EngineClass.new
+      engine.init(system_config)
+      engine.configure(conf)
+
+      assert_kind_of DummyEngineTestInput, engine.root_agent.inputs[0]
+      assert_kind_of DummyEngineTestOutput, engine.root_agent.outputs[0]
+
+      conf_data = <<-CONF
+        <source>
+          @type dummy_engine_class_var_test
+        </source>
+        <match>
+          @type dummy_engine_test
+        </match>
+      CONF
+
+      new_conf = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+
+      e = assert_raise(Fluent::ConfigError) do
+        engine.reload_config(new_conf)
+      end
+      assert e.message.match?('Unreloadable plugin plugin: dummy_engine_class_var_test')
+
+      assert_kind_of DummyEngineTestInput, engine.root_agent.inputs[0]
+      assert_kind_of DummyEngineTestOutput, engine.root_agent.outputs[0]
+    end
+  end
+end
diff --git a/test/test_static_config_analysis.rb b/test/test_static_config_analysis.rb
new file mode 100644
index 0000000000..50477db2f2
--- /dev/null
+++ b/test/test_static_config_analysis.rb
@@ -0,0 +1,177 @@
+require_relative 'helper'
+
+require 'fluent/config'
+require 'fluent/static_config_analysis'
+require 'fluent/plugin/out_forward'
+require 'fluent/plugin/out_stdout'
+require 'fluent/plugin/out_exec'
+require 'fluent/plugin/in_forward'
+require 'fluent/plugin/in_dummy'
+require 'fluent/plugin/filter_grep'
+require 'fluent/plugin/filter_stdout'
+require 'fluent/plugin/filter_parser'
+
+class StaticConfigAnalysisTest < ::Test::Unit::TestCase
+  sub_test_case '.call' do
+    test 'returns outputs, inputs and filters' do
+      conf_data = <<-CONF
+<source>
+  @type forward
+</source>
+<filter>
+  @type grep
+</filter>
+<match>
+  @type forward
+</match>
+      CONF
+
+      c = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+      ret = Fluent::StaticConfigAnalysis.call(c)
+      assert_equal 1, ret.outputs.size
+      assert_kind_of Fluent::Plugin::ForwardOutput, ret.outputs[0].plugin
+      assert_equal 1, ret.inputs.size
+      assert_kind_of Fluent::Plugin::ForwardInput, ret.inputs[0].plugin
+      assert_equal 1, ret.filters.size
+      assert_kind_of Fluent::Plugin::GrepFilter, ret.filters[0].plugin
+      assert_empty ret.labels
+
+      assert_equal [Fluent::Plugin::ForwardOutput, Fluent::Plugin::ForwardInput, Fluent::Plugin::GrepFilter], ret.all_plugins.map(&:class)
+    end
+
+    test 'returns wrapped element with worker and label section' do
+      conf_data = <<-CONF
+<source>
+  @type forward
+</source>
+<filter>
+  @type grep
+</filter>
+<match>
+  @type forward
+</match>
+<worker 0>
+  <source>
+    @type dummy
+  </source>
+  <filter>
+    @type parser
+  </filter>
+  <match>
+    @type exec
+  </match>
+</worker>
+<label @test>
+  <filter>
+    @type stdout
+  </filter>
+  <match>
+    @type stdout
+  </match>
+</label>
+      CONF
+
+      c = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+      ret = Fluent::StaticConfigAnalysis.call(c)
+      assert_equal [Fluent::Plugin::ExecOutput, Fluent::Plugin::StdoutOutput, Fluent::Plugin::ForwardOutput], ret.outputs.map(&:plugin).map(&:class)
+      assert_equal [Fluent::Plugin::DummyInput, Fluent::Plugin::ForwardInput], ret.inputs.map(&:plugin).map(&:class)
+      assert_equal [Fluent::Plugin::ParserFilter, Fluent::Plugin::StdoutFilter, Fluent::Plugin::GrepFilter], ret.filters.map(&:plugin).map(&:class)
+      assert_equal 1, ret.labels.size
+      assert_equal '@test', ret.labels[0].name
+    end
+
+    sub_test_case 'raises config error' do
+      data(
+        'empty' => ['', 'Missing worker id on <worker> directive'],
+        'invalid number' => ['a', 'worker id should be integer: a'],
+        'worker id is negative' => ['-1', 'worker id should be integer: -1'],
+        'min worker id is less than 0' => ['-1-1', 'worker id should be integer: -1-1'],
+        'max worker id is less than 0' => ['1--1', 'worker id -1 specified by <worker> directive is not allowed. Available worker id is between 0 and 1'],
+        'min worker id is greater than workers' => ['0-2', 'worker id 2 specified by <worker> directive is not allowed. Available worker id is between 0 and 1'],
+        'max worker is less than min worker' => ['1-0', "greater first_worker_id<1> than last_worker_id<0> specified by <worker> directive is not allowed. Available multi worker assign syntax is <smaller_worker_id>-<greater_worker_id>"],
+      )
+      test 'when worker number is invalid' do |v|
+        val, msg = v
+        conf_data = <<-CONF
+<worker #{val}>
+</worker>
+CONF
+
+        c = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+        assert_raise(Fluent::ConfigError.new(msg)) do
+          Fluent::StaticConfigAnalysis.call(c, workers: 2)
+        end
+      end
+
+      test 'when worker number is duplicated' do
+        conf_data = <<-CONF
+<worker 0-1>
+</worker>
+<worker 0-1>
+</worker>
+CONF
+
+        c = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+        assert_raise(Fluent::ConfigError.new("specified worker_id<0> collisions is detected on <worker> directive. Available worker id(s): []")) do
+          Fluent::StaticConfigAnalysis.call(c, workers: 2)
+        end
+      end
+
+      test 'duplicated label exits' do
+        conf_data = <<-CONF
+<label @dup>
+</label>
+<label @dup>
+</label>
+CONF
+
+        c = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+        assert_raise(Fluent::ConfigError.new('Section <label @dup> appears twice')) do
+          Fluent::StaticConfigAnalysis.call(c, workers: 2)
+        end
+      end
+
+      test 'empty label' do
+        conf_data = <<-CONF
+<label>
+</label>
+CONF
+
+        c = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+        assert_raise(Fluent::ConfigError.new('Missing symbol argument on <label> directive')) do
+          Fluent::StaticConfigAnalysis.call(c, workers: 2)
+        end
+      end
+
+      data(
+        'in filter' => 'filter',
+        'in source' => 'source',
+        'in match' => 'match',
+      )
+      test 'when @type is missing' do |name|
+        conf_data = <<-CONF
+<#{name}>
+  @type
+</#{name}>
+CONF
+        c = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+        assert_raise(Fluent::ConfigError.new("Missing '@type' parameter on <#{name}> directive")) do
+          Fluent::StaticConfigAnalysis.call(c)
+        end
+      end
+
+      test 'when worker has worker section' do
+        conf_data = <<-CONF
+<worker 0>
+  <worker 0>
+  </worker>
+</worker>
+CONF
+        c = Fluent::Config.parse(conf_data, '(test)', '(test_dir)', true)
+        assert_raise(Fluent::ConfigError.new("<worker> section cannot have <worker> directive")) do
+          Fluent::StaticConfigAnalysis.call(c)
+        end
+      end
+    end
+  end
+end
diff --git a/test/test_supervisor.rb b/test/test_supervisor.rb
index 7808cca6b4..f128e19353 100644
--- a/test/test_supervisor.rb
+++ b/test/test_supervisor.rb
@@ -30,81 +30,6 @@ def write_config(path, data)
     File.open(path, "w") {|f| f.write data }
   end
 
-  def test_read_config
-    create_info_dummy_logger
-
-    tmp_dir = "#{TMP_DIR}/dir/test_read_config.conf"
-    conf_str = %[
-<source>
-  @type forward
-  @id forward_input
-</source>
-<match debug.**>
-  @type stdout
-  @id stdout_output
-</match>
-]
-    write_config tmp_dir, conf_str
-    opts = Fluent::Supervisor.default_options
-    sv = Fluent::Supervisor.new(opts)
-
-    use_v1_config = {}
-    use_v1_config['use_v1_config'] = true
-
-    sv.instance_variable_set(:@config_path, tmp_dir)
-    sv.instance_variable_set(:@use_v1_config, use_v1_config)
-
-    conf = sv.__send__(:read_config)
-    elem = conf.elements.find { |e| e.name == 'source' }
-    assert_equal "forward", elem['@type']
-    assert_equal "forward_input", elem['@id']
-
-    elem = conf.elements.find { |e| e.name == 'match' }
-    assert_equal "debug.**", elem.arg
-    assert_equal "stdout", elem['@type']
-    assert_equal "stdout_output", elem['@id']
-
-    $log.out.reset
-  end
-
-  def test_read_config_with_multibyte_string
-    tmp_path = "#{TMP_DIR}/dir/test_multibyte_config.conf"
-    conf_str = %[
-<source>
-  @type forward
-  @id forward_input
-  @label @INPUT
-</source>
-<label @INPUT>
-  <filter>
-    @type record_transformer
-    <record>
-      message こんにちは. ${record["name"]} has made a order of ${record["item"]} just now.
-    </record>
-  </filter>
-  <match>
-    @type stdout
-  </match>
-</label>
-]
-    FileUtils.mkdir_p(File.dirname(tmp_path))
-    File.open(tmp_path, "w:utf-8") {|file| file.write(conf_str) }
-
-    opts = Fluent::Supervisor.default_options
-    sv = Fluent::Supervisor.new(opts)
-
-    use_v1_config = {}
-    use_v1_config['use_v1_config'] = true
-
-    sv.instance_variable_set(:@config_path, tmp_path)
-    sv.instance_variable_set(:@use_v1_config, use_v1_config)
-
-    conf = sv.__send__(:read_config)
-    label = conf.elements.detect {|e| e.name == "label" }
-    filter = label.elements.detect {|e| e.name == "filter" }
-    record_transformer = filter.elements.detect {|e| e.name = "record_transformer" }
-    assert_equal(Encoding::UTF_8, record_transformer["message"].encoding)
-  end
 
   def test_system_config
     opts = Fluent::Supervisor.default_options
@@ -416,7 +341,7 @@ def test_inline_config
 
     inline_config = '<match *>\n@type stdout\n</match>'
     stub(STDIN).read { inline_config }
-    stub(sv).read_config        # to skip
+    stub(Fluent::Config).build                                # to skip
     stub(sv).build_system_config { Fluent::SystemConfig.new } # to skip
 
     sv.configure
@@ -428,7 +353,7 @@ def test_log_level_affects
     sv = Fluent::Supervisor.new(opts)
 
     c = Fluent::Config::Element.new('system', '', { 'log_level' => 'error' }, [])
-    stub(sv).read_config { config_element('ROOT', '', {}, [c]) }
+    stub(Fluent::Config).build { config_element('ROOT', '', {}, [c]) }
 
     sv.configure
     assert_equal Fluent::Log::LEVEL_ERROR, $log.level
diff --git a/test/test_variable_store.rb b/test/test_variable_store.rb
new file mode 100644
index 0000000000..08dcfacde6
--- /dev/null
+++ b/test/test_variable_store.rb
@@ -0,0 +1,65 @@
+require_relative 'helper'
+require 'fluent/variable_store'
+
+class VariableStoreTest < Test::Unit::TestCase
+  def setup
+  end
+
+  def teardown
+    Fluent::VariableStore.try_to_reset do
+      # nothing
+    end
+  end
+
+  sub_test_case '#fetch_or_build' do
+    test 'fetch same object when the same key is passed' do
+      c1 = Fluent::VariableStore.fetch_or_build(:test)
+      c2 = Fluent::VariableStore.fetch_or_build(:test)
+
+      assert_equal c1, c2
+      assert_equal c1.object_id, c2.object_id
+
+      c3 = Fluent::VariableStore.fetch_or_build(:test2)
+      assert_not_equal c1.object_id, c3.object_id
+    end
+
+    test 'can be passed a default value' do
+      c1 = Fluent::VariableStore.fetch_or_build(:test, default_value: Set.new)
+      c2 = Fluent::VariableStore.fetch_or_build(:test)
+
+      assert_kind_of Set, c1
+      assert_equal c1, c2
+      assert_equal c1.object_id, c2.object_id
+    end
+  end
+
+  sub_test_case '#try_to_reset' do
+    test 'reset all values' do
+      c1 = Fluent::VariableStore.fetch_or_build(:test)
+      c1[:k1] = 1
+      assert_equal 1, c1[:k1]
+
+      Fluent::VariableStore.try_to_reset do
+        # nothing
+      end
+
+      c1 = Fluent::VariableStore.fetch_or_build(:test)
+      assert_nil c1[:k1]
+    end
+
+    test 'rollback resetting if error raised' do
+      c1 = Fluent::VariableStore.fetch_or_build(:test)
+      c1[:k1] = 1
+      assert_equal 1, c1[:k1]
+
+      assert_raise(RuntimeError.new('pass')) do
+        Fluent::VariableStore.try_to_reset do
+          raise 'pass'
+        end
+      end
+
+      c1 = Fluent::VariableStore.fetch_or_build(:test)
+      assert_equal 1, c1[:k1]
+    end
+  end
+end