diff --git a/.github/ISSUE_TEMPLATE/bug_report.yaml b/.github/ISSUE_TEMPLATE/bug_report.yml
similarity index 100%
rename from .github/ISSUE_TEMPLATE/bug_report.yaml
rename to .github/ISSUE_TEMPLATE/bug_report.yml
diff --git a/.github/ISSUE_TEMPLATE/feature_request.yaml b/.github/ISSUE_TEMPLATE/feature_request.yml
similarity index 100%
rename from .github/ISSUE_TEMPLATE/feature_request.yaml
rename to .github/ISSUE_TEMPLATE/feature_request.yml
diff --git a/.github/workflows/linux-test.yaml b/.github/workflows/linux-test.yaml
deleted file mode 100644
index de2757a508..0000000000
--- a/.github/workflows/linux-test.yaml
+++ /dev/null
@@ -1,36 +0,0 @@
-name: Testing on Ubuntu
-
-on:
-  push:
-    branches: [master]
-  pull_request:
-    branches: [master]
-
-jobs:
-  test:
-    runs-on: ${{ matrix.os }}
-    continue-on-error: ${{ matrix.experimental }}
-    strategy:
-      fail-fast: false
-      matrix:
-        ruby-version: ['3.2', '3.1', '3.0', '2.7']
-        os: [ubuntu-latest]
-        experimental: [false]
-        include:
-          - ruby-version: head
-            os: ubuntu-latest
-            experimental: true
-
-    name: Unit testing with Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }}
-    steps:
-      - uses: actions/checkout@v3
-      - name: Set up Ruby
-        uses: ruby/setup-ruby@v1
-        with:
-          ruby-version: ${{ matrix.ruby-version }}
-      - name: Install addons
-        run: sudo apt-get install libgmp3-dev libcap-ng-dev
-      - name: Install dependencies
-        run: bundle install
-      - name: Run tests
-        run: bundle exec rake test
diff --git a/.github/workflows/macos-test.yaml b/.github/workflows/macos-test.yaml
deleted file mode 100644
index de54469ddd..0000000000
--- a/.github/workflows/macos-test.yaml
+++ /dev/null
@@ -1,34 +0,0 @@
-name: Testing on macOS
-
-on:
-  push:
-    branches: [master]
-  pull_request:
-    branches: [master]
-
-jobs:
-  test:
-    runs-on: ${{ matrix.os }}
-    continue-on-error: ${{ matrix.experimental }}
-    strategy:
-      fail-fast: false
-      matrix:
-        ruby-version: ['3.2', '3.1', '3.0', '2.7']
-        os: [macos-latest]
-        experimental: [true]
-        include:
-          - ruby-version: head
-            os: macos-latest
-            experimental: true
-
-    name: Unit testing with Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }}
-    steps:
-      - uses: actions/checkout@v3
-      - name: Set up Ruby
-        uses: ruby/setup-ruby@v1
-        with:
-          ruby-version: ${{ matrix.ruby-version }}
-      - name: Install dependencies
-        run: bundle install
-      - name: Run tests
-        run: bundle exec rake test
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
new file mode 100644
index 0000000000..2869d32c6d
--- /dev/null
+++ b/.github/workflows/test.yml
@@ -0,0 +1,38 @@
+name: Test
+
+on:
+  push:
+    branches: [v1.16]
+    paths-ignore:
+      - '*.md'
+      - 'lib/fluent/version.rb'
+  pull_request:
+    branches: [v1.16]
+    paths-ignore:
+      - '*.md'
+      - 'lib/fluent/version.rb'
+
+jobs:
+  test:
+    runs-on: ${{ matrix.os }}
+    continue-on-error: false
+    strategy:
+      fail-fast: false
+      matrix:
+        os: ['ubuntu-latest', 'macos-latest', 'windows-latest']
+        ruby-version: ['3.3', '3.2', '3.1', '3.0', '2.7']
+
+    name: Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Ruby
+        uses: ruby/setup-ruby@v1
+        with:
+          ruby-version: ${{ matrix.ruby-version }}
+      - name: Install addons
+        if: ${{ matrix.os == 'ubuntu-latest' }}
+        run: sudo apt-get install libgmp3-dev libcap-ng-dev
+      - name: Install dependencies
+        run: bundle install
+      - name: Run tests
+        run: bundle exec rake test TESTOPTS="-v --no-show-detail-immediately"
diff --git a/.github/workflows/windows-test.yaml b/.github/workflows/windows-test.yaml
deleted file mode 100644
index 10792577b5..0000000000
--- a/.github/workflows/windows-test.yaml
+++ /dev/null
@@ -1,49 +0,0 @@
-name: Testing on Windows
-
-on:
-  push:
-    branches: [master]
-  pull_request:
-    branches: [master]
-
-jobs:
-  test:
-    runs-on: ${{ matrix.os }}
-    continue-on-error: ${{ matrix.experimental }}
-    strategy:
-      fail-fast: false
-      matrix:
-        ruby-version: ['3.2', '3.1', '2.7']
-        os:
-          - windows-latest
-        experimental: [false]
-        include:
-          - ruby-version: head
-            os: windows-latest
-            experimental: true
-          - ruby-version: '3.0.3'
-            os: windows-latest
-            experimental: false
-            # On Ruby 3.0, we need to use fiddle 1.0.8 or later to retrieve correct
-            # error code. In addition, we have to specify the path of fiddle by RUBYLIB
-            # because RubyInstaller loads Ruby's bundled fiddle before initializing gem.
-            # See also:
-            # * https://github.com/ruby/fiddle/issues/72
-            # * https://bugs.ruby-lang.org/issues/17813
-            # * https://github.com/oneclick/rubyinstaller2/blob/8225034c22152d8195bc0aabc42a956c79d6c712/lib/ruby_installer/build/dll_directory.rb
-            ruby-lib-opt: RUBYLIB=%RUNNER_TOOL_CACHE%/Ruby/3.0.3/x64/lib/ruby/gems/3.0.0/gems/fiddle-1.1.0/lib
-
-    name: Unit testing with Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }}
-    steps:
-      - uses: actions/checkout@v3
-      - name: Set up Ruby
-        uses: ruby/setup-ruby@v1
-        with:
-          ruby-version: ${{ matrix.ruby-version }}
-      - name: Add Fiddle 1.1.0
-        if: ${{ matrix.ruby-version == '3.0.3' }}
-        run: gem install fiddle --version 1.1.0
-      - name: Install dependencies
-        run: ridk exec bundle install
-      - name: Run tests
-        run: bundle exec rake test TESTOPTS=-v ${{ matrix.ruby-lib-opt }}
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ca435b2fd6..0f27ba6a60 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,60 @@
 # v1.16
 
+## Release v1.16.6 - 2024/08/16
+
+### Bug Fix
+
+* YAML config syntax: Fix issue where `$log_level` element was not supported correctly
+  https://github.com/fluent/fluentd/pull/4486
+* parser_json: Fix wrong LoadError warning
+  https://github.com/fluent/fluentd/pull/4592
+* `fluentd` command: Fix `--plugin` (`-p`) option not to overwrite default value
+  https://github.com/fluent/fluentd/pull/4605
+
+### Misc
+
+* out_file: Add warn message for symlink_path setting
+  https://github.com/fluent/fluentd/pull/4512
+* Keep console gem v1.23 to avoid LoadError
+  https://github.com/fluent/fluentd/pull/4510
+
+## Release v1.16.5 - 2024/03/27
+
+### Bug Fix
+
+* Buffer: Fix emit error of v1.16.4 sometimes failing to process large data
+  exceeding chunk size limit
+  https://github.com/fluent/fluentd/pull/4447
+
+## Release v1.16.4 - 2024/03/14
+
+### Bug Fix
+
+* Fix to avoid processing discarded chunks in write_step_by_step.
+  It fixes not to raise pile of IOError when many `chunk
+  bytes limit exceeds` errors are occurred.
+  https://github.com/fluent/fluentd/pull/4342
+* in_tail: Fix tail watchers in `rotate_wait` state not being managed.
+  https://github.com/fluent/fluentd/pull/4334
+
+### Misc
+
+* buffer: Avoid unnecessary log processing. It will improve performance.
+  https://github.com/fluent/fluentd/pull/4331
+
+## Release v1.16.3 - 2023/11/14
+
+### Bug Fix
+
+* in_tail: Fix a stall bug on !follow_inode case
+  https://github.com/fluent/fluentd/pull/4327
+* in_tail: add warning for silent stop on !follow_inodes case
+  https://github.com/fluent/fluentd/pull/4339
+* Buffer: Fix NoMethodError with empty unstaged chunk arrays
+  https://github.com/fluent/fluentd/pull/4303
+* Fix for rotate_age where Fluentd passes as Symbol
+  https://github.com/fluent/fluentd/pull/4311
+
 ## Release v1.16.2 - 2023/07/14
 
 ### Bug Fix
diff --git a/fluentd.gemspec b/fluentd.gemspec
index a5b5c805e3..879c26f6ff 100644
--- a/fluentd.gemspec
+++ b/fluentd.gemspec
@@ -29,6 +29,7 @@ Gem::Specification.new do |gem|
   gem.add_runtime_dependency("tzinfo-data", ["~> 1.0"])
   gem.add_runtime_dependency("strptime", [">= 0.2.4", "< 1.0.0"])
   gem.add_runtime_dependency("webrick", ["~> 1.4"])
+  gem.add_runtime_dependency("console", ["< 1.24"])
 
   # build gem for a certain platform. see also Rakefile
   fake_platform = ENV['GEM_BUILD_FAKE_PLATFORM'].to_s
@@ -45,7 +46,9 @@ Gem::Specification.new do |gem|
   gem.add_development_dependency("parallel_tests", ["~> 0.15.3"])
   gem.add_development_dependency("simplecov", ["~> 0.7"])
   gem.add_development_dependency("rr", ["~> 3.0"])
-  gem.add_development_dependency("timecop", ["~> 0.9"])
+  # timecop v0.9.9 supports `Process.clock_gettime`. It breaks some tests.
+  # (https://github.com/fluent/fluentd/pull/4521)
+  gem.add_development_dependency("timecop", ["< 0.9.9"])
   gem.add_development_dependency("test-unit", ["~> 3.3"])
   gem.add_development_dependency("test-unit-rr", ["~> 1.0"])
   gem.add_development_dependency("oj", [">= 2.14", "< 4"])
diff --git a/lib/fluent/command/fluentd.rb b/lib/fluent/command/fluentd.rb
index afa26553e5..da25b74cca 100644
--- a/lib/fluent/command/fluentd.rb
+++ b/lib/fluent/command/fluentd.rb
@@ -46,7 +46,7 @@
 }
 
 op.on('-p', '--plugin DIR', "add plugin directory") {|s|
-  (cmd_opts[:plugin_dirs] ||= []) << s
+  (cmd_opts[:plugin_dirs] ||= default_opts[:plugin_dirs]) << s
 }
 
 op.on('-I PATH', "add library path") {|s|
diff --git a/lib/fluent/config/yaml_parser/parser.rb b/lib/fluent/config/yaml_parser/parser.rb
index dc1caaf46d..862f774802 100644
--- a/lib/fluent/config/yaml_parser/parser.rb
+++ b/lib/fluent/config/yaml_parser/parser.rb
@@ -138,6 +138,10 @@ def section_build(name, config, indent: 0, arg: nil)
             sb.add_line('@id', v)
           end
 
+          if (v = config.delete('$log_level'))
+            sb.add_line('@log_level', v)
+          end
+
           config.each do |key, val|
             if val.is_a?(Array)
               val.each do |v|
diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb
index d04ae08296..80709c12bb 100644
--- a/lib/fluent/plugin/buffer.rb
+++ b/lib/fluent/plugin/buffer.rb
@@ -417,7 +417,7 @@ def write(metadata_and_data, format: nil, size: nil, enqueue: false)
             if c.staged? && (enqueue || chunk_size_full?(c))
               m = c.metadata
               enqueue_chunk(m)
-              if unstaged_chunks[m]
+              if unstaged_chunks[m] && !unstaged_chunks[m].empty?
                 u = unstaged_chunks[m].pop
                 u.synchronize do
                   if u.unstaged? && !chunk_size_full?(u)
@@ -580,7 +580,7 @@ def takeback_chunk(chunk_id)
           chunk = @dequeued.delete(chunk_id)
           return false unless chunk # already purged by other thread
           @queue.unshift(chunk)
-          log.trace "chunk taken back", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: chunk.metadata
+          log.on_trace { log.trace "chunk taken back", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: chunk.metadata }
           @queued_num[chunk.metadata] += 1 # BUG if nil
           @dequeued_num[chunk.metadata] -= 1
         end
@@ -610,7 +610,7 @@ def purge_chunk(chunk_id)
             @queued_num.delete(metadata)
             @dequeued_num.delete(metadata)
           end
-          log.trace "chunk purged", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: metadata
+          log.on_trace { log.trace "chunk purged", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: metadata }
         end
 
         nil
@@ -728,7 +728,6 @@ def write_once(metadata, data, format: nil, size: nil, &block)
 
       def write_step_by_step(metadata, data, format, splits_count, &block)
         splits = []
-        errors = []
         if splits_count > data.size
           splits_count = data.size
         end
@@ -749,16 +748,14 @@ def write_step_by_step(metadata, data, format, splits_count, &block)
         modified_chunks = []
         modified_metadata = metadata
         get_next_chunk = ->(){
-          c = if staged_chunk_used
-                # Staging new chunk here is bad idea:
-                # Recovering whole state including newly staged chunks is much harder than current implementation.
-                modified_metadata = modified_metadata.dup_next
-                generate_chunk(modified_metadata)
-              else
-                synchronize { @stage[modified_metadata] ||= generate_chunk(modified_metadata).staged! }
-              end
-          modified_chunks << c
-          c
+          if staged_chunk_used
+            # Staging new chunk here is bad idea:
+            # Recovering whole state including newly staged chunks is much harder than current implementation.
+            modified_metadata = modified_metadata.dup_next
+            generate_chunk(modified_metadata)
+          else
+            synchronize { @stage[modified_metadata] ||= generate_chunk(modified_metadata).staged! }
+          end
         }
 
         writing_splits_index = 0
@@ -766,104 +763,116 @@ def write_step_by_step(metadata, data, format, splits_count, &block)
 
         while writing_splits_index < splits.size
           chunk = get_next_chunk.call
-          chunk.synchronize do
-            raise ShouldRetry unless chunk.writable?
-            staged_chunk_used = true if chunk.staged?
-
-            original_bytesize = committed_bytesize = chunk.bytesize
-            begin
-              while writing_splits_index < splits.size
-                split = splits[writing_splits_index]
-                formatted_split = format ? format.call(split) : nil
+          errors = []
+          # The chunk must be locked until being passed to &block.
+          chunk.mon_enter
+          modified_chunks << {chunk: chunk, adding_bytesize: 0, errors: errors}
 
-                if split.size == 1 # Check BufferChunkOverflowError
-                  determined_bytesize = nil
-                  if @compress != :text
-                    determined_bytesize = nil
-                  elsif formatted_split
-                    determined_bytesize = formatted_split.bytesize
-                  elsif split.first.respond_to?(:bytesize)
-                    determined_bytesize = split.first.bytesize
-                  end
+          raise ShouldRetry unless chunk.writable?
+          staged_chunk_used = true if chunk.staged?
 
-                  if determined_bytesize && determined_bytesize > @chunk_limit_size
-                    # It is a obvious case that BufferChunkOverflowError should be raised here.
-                    # But if it raises here, already processed 'split' or
-                    # the proceeding 'split' will be lost completely.
-                    # So it is a last resort to delay raising such a exception
-                    errors << "a #{determined_bytesize} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})"
-                    writing_splits_index += 1
-                    next
-                  end
+          original_bytesize = committed_bytesize = chunk.bytesize
+          begin
+            while writing_splits_index < splits.size
+              split = splits[writing_splits_index]
+              formatted_split = format ? format.call(split) : nil
 
-                  if determined_bytesize.nil? || chunk.bytesize + determined_bytesize > @chunk_limit_size
-                    # The split will (might) cause size over so keep already processed
-                    # 'split' content here (allow performance regression a bit).
-                    chunk.commit
-                    committed_bytesize = chunk.bytesize
-                  end
+              if split.size == 1 # Check BufferChunkOverflowError
+                determined_bytesize = nil
+                if @compress != :text
+                  determined_bytesize = nil
+                elsif formatted_split
+                  determined_bytesize = formatted_split.bytesize
+                elsif split.first.respond_to?(:bytesize)
+                  determined_bytesize = split.first.bytesize
                 end
 
-                if format
-                  chunk.concat(formatted_split, split.size)
-                else
-                  chunk.append(split, compress: @compress)
+                if determined_bytesize && determined_bytesize > @chunk_limit_size
+                  # It is a obvious case that BufferChunkOverflowError should be raised here.
+                  # But if it raises here, already processed 'split' or
+                  # the proceeding 'split' will be lost completely.
+                  # So it is a last resort to delay raising such a exception
+                  errors << "a #{determined_bytesize} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})"
+                  writing_splits_index += 1
+                  next
                 end
-                adding_bytes = chunk.bytesize - committed_bytesize
 
-                if chunk_size_over?(chunk) # split size is larger than difference between size_full? and size_over?
-                  chunk.rollback
+                if determined_bytesize.nil? || chunk.bytesize + determined_bytesize > @chunk_limit_size
+                  # The split will (might) cause size over so keep already processed
+                  # 'split' content here (allow performance regression a bit).
+                  chunk.commit
                   committed_bytesize = chunk.bytesize
+                end
+              end
 
-                  if split.size == 1 # Check BufferChunkOverflowError again
-                    if adding_bytes > @chunk_limit_size
-                      errors << "concatenated/appended a #{adding_bytes} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})"
-                      writing_splits_index += 1
-                      next
-                    else
-                      # As already processed content is kept after rollback, then unstaged chunk should be queued.
-                      # After that, re-process current split again.
-                      # New chunk should be allocated, to do it, modify @stage and so on.
-                      synchronize { @stage.delete(modified_metadata) }
-                      staged_chunk_used = false
-                      chunk.unstaged!
-                      break
-                    end
-                  end
+              if format
+                chunk.concat(formatted_split, split.size)
+              else
+                chunk.append(split, compress: @compress)
+              end
+              adding_bytes = chunk.bytesize - committed_bytesize
 
-                  if chunk_size_full?(chunk) || split.size == 1
-                    enqueue_chunk_before_retry = true
+              if chunk_size_over?(chunk) # split size is larger than difference between size_full? and size_over?
+                chunk.rollback
+                committed_bytesize = chunk.bytesize
+
+                if split.size == 1 # Check BufferChunkOverflowError again
+                  if adding_bytes > @chunk_limit_size
+                    errors << "concatenated/appended a #{adding_bytes} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})"
+                    writing_splits_index += 1
+                    next
                   else
-                    splits_count *= 10
+                    # As already processed content is kept after rollback, then unstaged chunk should be queued.
+                    # After that, re-process current split again.
+                    # New chunk should be allocated, to do it, modify @stage and so on.
+                    synchronize { @stage.delete(modified_metadata) }
+                    staged_chunk_used = false
+                    chunk.unstaged!
+                    break
                   end
+                end
 
-                  raise ShouldRetry
+                if chunk_size_full?(chunk) || split.size == 1
+                  enqueue_chunk_before_retry = true
+                else
+                  splits_count *= 10
                 end
 
-                writing_splits_index += 1
+                raise ShouldRetry
+              end
+
+              writing_splits_index += 1
 
-                if chunk_size_full?(chunk)
-                  break
-                end
+              if chunk_size_full?(chunk)
+                break
               end
-            rescue
-              chunk.purge if chunk.unstaged? # unstaged chunk will leak unless purge it
-              raise
             end
-
-            block.call(chunk, chunk.bytesize - original_bytesize, errors)
-            errors = []
+          rescue
+            chunk.purge if chunk.unstaged? # unstaged chunk will leak unless purge it
+            raise
           end
+
+          modified_chunks.last[:adding_bytesize] = chunk.bytesize - original_bytesize
+        end
+        modified_chunks.each do |data|
+          block.call(data[:chunk], data[:adding_bytesize], data[:errors])
         end
       rescue ShouldRetry
-        modified_chunks.each do |mc|
-          mc.rollback rescue nil
-          if mc.unstaged?
-            mc.purge rescue nil
+        modified_chunks.each do |data|
+          chunk = data[:chunk]
+          chunk.rollback rescue nil
+          if chunk.unstaged?
+            chunk.purge rescue nil
           end
+          chunk.mon_exit rescue nil
         end
         enqueue_chunk(metadata) if enqueue_chunk_before_retry
         retry
+      ensure
+        modified_chunks.each do |data|
+          chunk = data[:chunk]
+          chunk.mon_exit
+        end
       end
 
       STATS_KEYS = [
diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb
index 54fd900039..7e0289f65b 100644
--- a/lib/fluent/plugin/in_tail.rb
+++ b/lib/fluent/plugin/in_tail.rb
@@ -52,6 +52,7 @@ def initialize
       super
       @paths = []
       @tails = {}
+      @tails_rotate_wait = {}
       @pf_file = nil
       @pf = nil
       @ignore_list = []
@@ -267,6 +268,9 @@ def shutdown
       @shutdown_start_time = Fluent::Clock.now
       # during shutdown phase, don't close io. It should be done in close after all threads are stopped. See close.
       stop_watchers(existence_path, immediate: true, remove_watcher: false)
+      @tails_rotate_wait.keys.each do |tw|
+        detach_watcher(tw, @tails_rotate_wait[tw][:ino], false)
+      end
       @pf_file.close if @pf_file
 
       super
@@ -275,6 +279,7 @@ def shutdown
     def close
       super
       # close file handles after all threads stopped (in #close of thread plugin helper)
+      # It may be because we need to wait IOHanlder.ready_to_shutdown()
       close_watcher_handles
     end
 
@@ -385,7 +390,7 @@ def refresh_watchers
         # So that inode can't be contained in `removed_hash`, and can't be unwatched by `stop_watchers`.
         #
         # This logic may work for `@follow_inodes false` too.
-        # Just limiting the case to supress the impact to existing logics.
+        # Just limiting the case to suppress the impact to existing logics.
         @pf&.unwatch_removed_targets(target_paths_hash)
         need_unwatch_in_stop_watchers = false
       end
@@ -393,6 +398,28 @@ def refresh_watchers
       removed_hash = existence_paths_hash.reject {|key, value| target_paths_hash.key?(key)}
       added_hash = target_paths_hash.reject {|key, value| existence_paths_hash.key?(key)}
 
+      # If an exisiting TailWatcher already follows a target path with the different inode,
+      # it means that the TailWatcher following the rotated file still exists. In this case,
+      # `refresh_watcher` can't start the new TailWatcher for the new current file. So, we
+      # should output a warning log in order to prevent silent collection stops.
+      # (Such as https://github.com/fluent/fluentd/pull/4327)
+      # (Usually, such a TailWatcher should be removed from `@tails` in `update_watcher`.)
+      # (The similar warning may work for `@follow_inodes true` too. Just limiting the case
+      # to suppress the impact to existing logics.)
+      unless @follow_inodes
+        target_paths_hash.each do |path, target|
+          next unless @tails.key?(path)
+          # We can't use `existence_paths_hash[path].ino` because it is from `TailWatcher.ino`,
+          # which is very unstable parameter. (It can be `nil` or old).
+          # So, we need to use `TailWatcher.pe.read_inode`.
+          existing_watcher_inode = @tails[path].pe.read_inode
+          if existing_watcher_inode != target.ino
+            log.warn "Could not follow a file (inode: #{target.ino}) because an existing watcher for that filepath follows a different inode: #{existing_watcher_inode} (e.g. keeps watching a already rotated file). If you keep getting this message, please restart Fluentd.",
+              filepath: target.path
+          end
+        end
+      end
+
       stop_watchers(removed_hash, unwatched: need_unwatch_in_stop_watchers) unless removed_hash.empty?
       start_watchers(added_hash) unless added_hash.empty?
       @startup = false if @startup
@@ -494,6 +521,9 @@ def close_watcher_handles
           tw.close
         end
       end
+      @tails_rotate_wait.keys.each do |tw|
+        tw.close
+      end
     end
 
     # refresh_watchers calls @tails.keys so we don't use stop_watcher -> start_watcher sequence for safety.
@@ -548,10 +578,6 @@ def update_watcher(tail_watcher, pe, new_inode)
       detach_watcher_after_rotate_wait(tail_watcher, pe.read_inode)
     end
 
-    # TailWatcher#close is called by another thread at shutdown phase.
-    # It causes 'can't modify string; temporarily locked' error in IOHandler
-    # so adding close_io argument to avoid this problem.
-    # At shutdown, IOHandler's io will be released automatically after detached the event loop
     def detach_watcher(tw, ino, close_io = true)
       if @follow_inodes && tw.ino != ino
         log.warn("detach_watcher could be detaching an unexpected tail_watcher with a different ino.",
@@ -564,7 +590,7 @@ def detach_watcher(tw, ino, close_io = true)
 
       tw.close if close_io
 
-      if tw.unwatched && @pf
+      if @pf && tw.unwatched && (@follow_inode || !@tails[tw.path])
         target_info = TargetInfo.new(tw.path, ino)
         @pf.unwatch(target_info)
       end
@@ -582,7 +608,11 @@ def detach_watcher_after_rotate_wait(tw, ino)
       if @open_on_every_update
         # Detach now because it's already closed, waiting it doesn't make sense.
         detach_watcher(tw, ino)
-      elsif throttling_is_enabled?(tw)
+      end
+
+      return if @tails_rotate_wait[tw]
+
+      if throttling_is_enabled?(tw)
         # When the throttling feature is enabled, it might not reach EOF yet.
         # Should ensure to read all contents before closing it, with keeping throttling.
         start_time_to_wait = Fluent::Clock.now
@@ -590,14 +620,18 @@ def detach_watcher_after_rotate_wait(tw, ino)
           elapsed = Fluent::Clock.now - start_time_to_wait
           if tw.eof? && elapsed >= @rotate_wait
             timer.detach
+            @tails_rotate_wait.delete(tw)
             detach_watcher(tw, ino)
           end
         end
+        @tails_rotate_wait[tw] = { ino: ino, timer: timer }
       else
         # when the throttling feature isn't enabled, just wait @rotate_wait
-        timer_execute(:in_tail_close_watcher, @rotate_wait, repeat: false) do
+        timer = timer_execute(:in_tail_close_watcher, @rotate_wait, repeat: false) do
+          @tails_rotate_wait.delete(tw)
           detach_watcher(tw, ino)
         end
+        @tails_rotate_wait[tw] = { ino: ino, timer: timer }
       end
     end
 
diff --git a/lib/fluent/plugin/out_file.rb b/lib/fluent/plugin/out_file.rb
index 7de4b90979..19c0afa8ef 100644
--- a/lib/fluent/plugin/out_file.rb
+++ b/lib/fluent/plugin/out_file.rb
@@ -172,6 +172,14 @@ def configure(conf)
           log.warn "symlink_path is unavailable on Windows platform. disabled."
           @symlink_path = nil
         else
+          placeholder_validators(:symlink_path, @symlink_path).reject{ |v| v.type == :time }.each do |v|
+            begin
+              v.validate!
+            rescue Fluent::ConfigError => e
+              log.warn "#{e}. This means multiple chunks are competing for a single symlink_path, so some logs may not be taken from the symlink."
+            end
+          end
+
           @buffer.extend SymlinkBufferMixin
           @buffer.symlink_path = @symlink_path
           @buffer.output_plugin_for_symlink = self
diff --git a/lib/fluent/plugin/parser_json.rb b/lib/fluent/plugin/parser_json.rb
index 829aa4b72a..79fea5abe9 100644
--- a/lib/fluent/plugin/parser_json.rb
+++ b/lib/fluent/plugin/parser_json.rb
@@ -50,23 +50,15 @@ def configure(conf)
       def configure_json_parser(name)
         case name
         when :oj
-          raise LoadError unless Fluent::OjOptions.available?
-          [Oj.method(:load), Oj::ParseError]
+          return [Oj.method(:load), Oj::ParseError] if Fluent::OjOptions.available?
+
+          log&.info "Oj is not installed, and failing back to Yajl for json parser"
+          configure_json_parser(:yajl)
         when :json then [JSON.method(:load), JSON::ParserError]
         when :yajl then [Yajl.method(:load), Yajl::ParseError]
         else
           raise "BUG: unknown json parser specified: #{name}"
         end
-      rescue LoadError => ex
-        name = :yajl
-        if log
-          if /\boj\z/.match?(ex.message)
-            log.info "Oj is not installed, and failing back to Yajl for json parser"
-          else
-            log.warn ex.message
-          end
-        end
-        retry
       end
 
       def parse(text)
diff --git a/lib/fluent/system_config.rb b/lib/fluent/system_config.rb
index 5fb781d72b..4919294989 100644
--- a/lib/fluent/system_config.rb
+++ b/lib/fluent/system_config.rb
@@ -62,7 +62,7 @@ class SystemConfig
       config_param :time_format, :string, default: '%Y-%m-%d %H:%M:%S %z'
       config_param :rotate_age, default: nil do |v|
         if Fluent::Log::LOG_ROTATE_AGE.include?(v)
-          v.to_sym
+          v
         else
           begin
             Integer(v)
diff --git a/lib/fluent/version.rb b/lib/fluent/version.rb
index 87bcc578d9..6c6fdd7b24 100644
--- a/lib/fluent/version.rb
+++ b/lib/fluent/version.rb
@@ -16,6 +16,6 @@
 
 module Fluent
 
-  VERSION = '1.16.2'
+  VERSION = '1.16.6'
 
 end
diff --git a/test/command/test_cat.rb b/test/command/test_cat.rb
index 51c1a20bbb..40605815bb 100644
--- a/test/command/test_cat.rb
+++ b/test/command/test_cat.rb
@@ -18,7 +18,7 @@ def setup
     @primary = create_primary
     metadata = @primary.buffer.new_metadata
     @chunk = create_chunk(@primary, metadata, @es)
-    @port = unused_port
+    @port = unused_port(protocol: :all)
   end
 
   def teardown
diff --git a/test/command/test_fluentd.rb b/test/command/test_fluentd.rb
index 87c9d28d96..c51f7f9ad4 100644
--- a/test/command/test_fluentd.rb
+++ b/test/command/test_fluentd.rb
@@ -128,11 +128,14 @@ def eager_read(io)
 
   # ATTENTION: This stops taking logs when all `pattern_list` match or timeout,
   # so `patterns_not_match` can test only logs up to that point.
+  # You can pass a block to assert something after log matching.
   def assert_log_matches(cmdline, *pattern_list, patterns_not_match: [], timeout: 20, env: {})
     matched = false
     matched_wrongly = false
-    assert_error_msg = ""
+    error_msg_match = ""
     stdio_buf = ""
+    succeeded_block = true
+    error_msg_block = ""
     begin
       execute_command(cmdline, @tmp_dir, env) do |pid, stdout|
         begin
@@ -163,6 +166,13 @@ def assert_log_matches(cmdline, *pattern_list, patterns_not_match: [], timeout:
               end
             end
           end
+
+          begin
+            yield if block_given?
+          rescue => e
+            succeeded_block = false
+            error_msg_block = "failed block execution after matching: #{e}"
+          end
         ensure
           if SUPERVISOR_PID_PATTERN =~ stdio_buf
             @supervisor_pid = $1.to_i
@@ -173,19 +183,19 @@ def assert_log_matches(cmdline, *pattern_list, patterns_not_match: [], timeout:
         end
       end
     rescue Timeout::Error
-      assert_error_msg = "execution timeout"
+      error_msg_match = "execution timeout"
       # https://github.com/fluent/fluentd/issues/4095
       # On Windows, timeout without `@supervisor_pid` means that the test is invalid,
       # since the supervisor process will survive without being killed correctly.
       flunk("Invalid test: The pid of supervisor could not be taken, which is necessary on Windows.") if Fluent.windows? && @supervisor_pid.nil?
     rescue => e
-      assert_error_msg = "unexpected error in launching fluentd: #{e.inspect}"
+      error_msg_match = "unexpected error in launching fluentd: #{e.inspect}"
     else
-      assert_error_msg = "log doesn't match" unless matched
+      error_msg_match = "log doesn't match" unless matched
     end
 
     if patterns_not_match.empty?
-      assert_error_msg = build_message(assert_error_msg,
+      error_msg_match = build_message(error_msg_match,
                                        "<?>\nwas expected to include:\n<?>",
                                        stdio_buf, pattern_list)
     else
@@ -197,16 +207,17 @@ def assert_log_matches(cmdline, *pattern_list, patterns_not_match: [], timeout:
                             lines.any?{|line| line.include?(ptn) }
                           end
         if matched_wrongly
-          assert_error_msg << "\n" unless assert_error_msg.empty?
-          assert_error_msg << "pattern exists in logs wrongly: #{ptn}"
+          error_msg_match << "\n" unless error_msg_match.empty?
+          error_msg_match << "pattern exists in logs wrongly: #{ptn}"
         end
       end
-      assert_error_msg = build_message(assert_error_msg,
+      error_msg_match = build_message(error_msg_match,
                                        "<?>\nwas expected to include:\n<?>\nand not include:\n<?>",
                                        stdio_buf, pattern_list, patterns_not_match)
     end
 
-    assert matched && !matched_wrongly, assert_error_msg
+    assert matched && !matched_wrongly, error_msg_match
+    assert succeeded_block, error_msg_block if block_given?
   end
 
   def assert_fluentd_fails_to_start(cmdline, *pattern_list, timeout: 20)
@@ -941,7 +952,7 @@ def multi_workers_ready?
       '-external-encoding' => '--external-encoding=utf-8',
       '-internal-encoding' => '--internal-encoding=utf-8',
     )
-    test "-E option is set to RUBYOPT" do |opt|
+    test "-E option is set to RUBYOPT" do |base_opt|
       conf = <<CONF
 <source>
   @type dummy
@@ -952,6 +963,7 @@ def multi_workers_ready?
 </match>
 CONF
       conf_path = create_conf_file('rubyopt_test.conf', conf)
+      opt = base_opt.dup
       opt << " #{ENV['RUBYOPT']}" if ENV['RUBYOPT']
       assert_log_matches(
         create_cmdline(conf_path),
@@ -991,9 +1003,14 @@ def multi_workers_ready?
 </match>
 CONF
       conf_path = create_conf_file('rubyopt_invalid_test.conf', conf)
+      if Gem::Version.create(RUBY_VERSION) >= Gem::Version.create('3.3.0')
+        expected_phrase = 'ruby: invalid switch in RUBYOPT'
+      else
+        expected_phrase = 'Invalid option is passed to RUBYOPT'
+      end
       assert_log_matches(
         create_cmdline(conf_path),
-        'Invalid option is passed to RUBYOPT',
+        expected_phrase,
         env: { 'RUBYOPT' => 'a' },
       )
     end
@@ -1158,7 +1175,7 @@ def multi_workers_ready?; true; end
     end
   end
 
-  sub_test_case 'sahred socket options' do
+  sub_test_case 'shared socket options' do
     test 'enable shared socket by default' do
       conf = ""
       conf_path = create_conf_file('empty.conf', conf)
@@ -1282,4 +1299,40 @@ def multi_workers_ready?; true; end
                          "[debug]")
     end
   end
+
+  sub_test_case "plugin option" do
+    test "should be the default value when not specifying" do
+      conf_path = create_conf_file('test.conf', <<~CONF)
+        <source>
+          @type monitor_agent
+        </source>
+      CONF
+      assert File.exist?(conf_path)
+      cmdline = create_cmdline(conf_path)
+
+      assert_log_matches(cmdline, "fluentd worker is now running") do
+        response = Net::HTTP.get(URI.parse("http://localhost:24220/api/config.json"))
+        actual_conf = JSON.parse(response)
+        assert_equal Fluent::Supervisor.default_options[:plugin_dirs], actual_conf["plugin_dirs"]
+      end
+    end
+
+    data(short: "-p")
+    data(long: "--plugin")
+    test "can be added by specifying the option" do |option_name|
+      conf_path = create_conf_file('test.conf', <<~CONF)
+        <source>
+          @type monitor_agent
+        </source>
+      CONF
+      assert File.exist?(conf_path)
+      cmdline = create_cmdline(conf_path, option_name, @tmp_dir, option_name, @tmp_dir)
+
+      assert_log_matches(cmdline, "fluentd worker is now running") do
+        response = Net::HTTP.get(URI.parse("http://localhost:24220/api/config.json"))
+        actual_conf = JSON.parse(response)
+        assert_equal Fluent::Supervisor.default_options[:plugin_dirs] + [@tmp_dir, @tmp_dir], actual_conf["plugin_dirs"]
+      end
+    end
+  end
 end
diff --git a/test/config/test_system_config.rb b/test/config/test_system_config.rb
index 502fb7d75a..a45fde5f15 100644
--- a/test/config/test_system_config.rb
+++ b/test/config/test_system_config.rb
@@ -151,7 +151,7 @@ def parse_text(text)
       data('daily' => "daily",
            'weekly' => 'weekly',
            'monthly' => 'monthly')
-      test "symbols for rotate_age" do |age|
+      test "strings for rotate_age" do |age|
         conf = parse_text(<<-EOS)
           <system>
             <log>
@@ -160,7 +160,7 @@ def parse_text(text)
           </system>
         EOS
         sc = Fluent::SystemConfig.new(conf)
-        assert_equal(age.to_sym, sc.log.rotate_age)
+        assert_equal(age, sc.log.rotate_age)
       end
 
       test "numeric number for rotate age" do
diff --git a/test/helper.rb b/test/helper.rb
index 9dc8633423..4e9f028a9c 100644
--- a/test/helper.rb
+++ b/test/helper.rb
@@ -71,17 +71,31 @@ class Test::Unit::AssertionFailedError < StandardError
 
 include Fluent::Test::Helpers
 
-def unused_port(num = 1, protocol: :tcp, bind: "0.0.0.0")
+def unused_port(num = 1, protocol:, bind: "0.0.0.0")
   case protocol
-  when :tcp
+  when :tcp, :tls
     unused_port_tcp(num)
   when :udp
     unused_port_udp(num, bind: bind)
+  when :all
+    unused_port_tcp_udp(num)
   else
     raise ArgumentError, "unknown protocol: #{protocol}"
   end
 end
 
+def unused_port_tcp_udp(num = 1)
+  raise "not support num > 1" if num > 1
+
+  # The default maximum number of file descriptors in macOS is 256.
+  # It might need to set num to a smaller value than that.
+  tcp_ports = unused_port_tcp(200)
+  port = unused_port_udp(1, port_list: tcp_ports)
+  raise "can't find unused port" unless port
+
+  port
+end
+
 def unused_port_tcp(num = 1)
   ports = []
   sockets = []
@@ -90,7 +104,7 @@ def unused_port_tcp(num = 1)
     sockets << s
     ports << s.addr[1]
   end
-  sockets.each{|s| s.close }
+  sockets.each(&:close)
   if num == 1
     return ports.first
   else
@@ -100,12 +114,15 @@ def unused_port_tcp(num = 1)
 
 PORT_RANGE_AVAILABLE = (1024...65535)
 
-def unused_port_udp(num = 1, bind: "0.0.0.0")
+def unused_port_udp(num = 1, port_list: [], bind: "0.0.0.0")
   family = IPAddr.new(IPSocket.getaddress(bind)).ipv4? ? ::Socket::AF_INET : ::Socket::AF_INET6
   ports = []
   sockets = []
-  while ports.size < num
-    port = rand(PORT_RANGE_AVAILABLE)
+
+  use_random_port = port_list.empty?
+  i = 0
+  loop do
+    port = use_random_port ? rand(PORT_RANGE_AVAILABLE) : port_list[i]
     u = UDPSocket.new(family)
     if (u.bind(bind, port) rescue nil)
       ports << port
@@ -113,8 +130,11 @@ def unused_port_udp(num = 1, bind: "0.0.0.0")
     else
       u.close
     end
+    i += 1
+    break if ports.size >= num
+    break if !use_random_port && i >= port_list.size
   end
-  sockets.each{|s| s.close }
+  sockets.each(&:close)
   if num == 1
     return ports.first
   else
diff --git a/test/plugin/in_tail/test_io_handler.rb b/test/plugin/in_tail/test_io_handler.rb
index 647c4a55a1..0def36b31c 100644
--- a/test/plugin/in_tail/test_io_handler.rb
+++ b/test/plugin/in_tail/test_io_handler.rb
@@ -5,20 +5,19 @@
 require 'tempfile'
 
 class IntailIOHandlerTest < Test::Unit::TestCase
-  setup do
-    @file = Tempfile.new('intail_io_handler').binmode
-    opened_file_metrics = Fluent::Plugin::LocalMetrics.new
-    opened_file_metrics.configure(config_element('metrics', '', {}))
-    closed_file_metrics = Fluent::Plugin::LocalMetrics.new
-    closed_file_metrics.configure(config_element('metrics', '', {}))
-    rotated_file_metrics = Fluent::Plugin::LocalMetrics.new
-    rotated_file_metrics.configure(config_element('metrics', '', {}))
-    @metrics = Fluent::Plugin::TailInput::MetricsInfo.new(opened_file_metrics, closed_file_metrics, rotated_file_metrics)
-  end
-
-  teardown do
-    @file.close rescue nil
-    @file.unlink rescue nil
+  def setup
+    Tempfile.create('intail_io_handler') do |file|
+      file.binmode
+      @file = file
+      opened_file_metrics = Fluent::Plugin::LocalMetrics.new
+      opened_file_metrics.configure(config_element('metrics', '', {}))
+      closed_file_metrics = Fluent::Plugin::LocalMetrics.new
+      closed_file_metrics.configure(config_element('metrics', '', {}))
+      rotated_file_metrics = Fluent::Plugin::LocalMetrics.new
+      rotated_file_metrics.configure(config_element('metrics', '', {}))
+      @metrics = Fluent::Plugin::TailInput::MetricsInfo.new(opened_file_metrics, closed_file_metrics, rotated_file_metrics)
+      yield
+    end
   end
 
   def create_target_info
diff --git a/test/plugin/in_tail/test_position_file.rb b/test/plugin/in_tail/test_position_file.rb
index af692fdee5..b1957f7353 100644
--- a/test/plugin/in_tail/test_position_file.rb
+++ b/test/plugin/in_tail/test_position_file.rb
@@ -6,13 +6,12 @@
 require 'tempfile'
 
 class IntailPositionFileTest < Test::Unit::TestCase
-  setup do
-    @file = Tempfile.new('intail_position_file_test').binmode
-  end
-
-  teardown do
-    @file.close rescue nil
-    @file.unlink rescue nil
+  def setup
+    Tempfile.create('intail_position_file_test') do |file|
+      file.binmode
+      @file = file
+      yield
+    end
   end
 
   UNWATCHED_STR = '%016x' % Fluent::Plugin::TailInput::PositionFile::UNWATCHED_POSITION
diff --git a/test/plugin/out_forward/test_socket_cache.rb b/test/plugin/out_forward/test_socket_cache.rb
index f7dec5a0f1..1d584d6b4e 100644
--- a/test/plugin/out_forward/test_socket_cache.rb
+++ b/test/plugin/out_forward/test_socket_cache.rb
@@ -17,7 +17,7 @@ class SocketCacheTest < Test::Unit::TestCase
       assert_equal(socket, c.checkout_or('key') { socket })
       c.checkin(socket)
 
-      sock = dont_allow(mock!).open
+      sock = mock!.open.never.subject
       assert_equal(socket, c.checkout_or('key') { sock.open })
     end
 
@@ -130,7 +130,7 @@ def teardown
 
       c = Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log)
       sock = mock!.close { 'closed' }.subject
-      sock2 = dont_allow(mock!).close
+      sock2 = mock!.close.never.subject
       stub(sock).inspect
       stub(sock2).inspect
 
@@ -154,7 +154,7 @@ def teardown
       Timecop.freeze(Time.parse('2016-04-13 14:00:00 +0900'))
 
       c = Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log)
-      sock = dont_allow(mock!).close
+      sock = mock!.close.never.subject
       stub(sock).inspect
       c.checkout_or('key') { sock }
 
diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb
index d35d2d6ce7..14d33af9c7 100644
--- a/test/plugin/test_buffer.rb
+++ b/test/plugin/test_buffer.rb
@@ -850,6 +850,116 @@ def create_chunk_es(metadata, es)
     test '#compress returns :text' do
       assert_equal :text, @p.compress
     end
+
+    # https://github.com/fluent/fluentd/issues/3089
+    test "closed chunk should not be committed" do
+      assert_equal 8 * 1024 * 1024, @p.chunk_limit_size
+      assert_equal 0.95, @p.chunk_full_threshold
+
+      purge_count = 0
+
+      stub.proxy(@p).generate_chunk(anything) do |chunk|
+        stub.proxy(chunk).purge do |result|
+          purge_count += 1
+          result
+        end
+        stub.proxy(chunk).commit do |result|
+          assert_false(chunk.closed?)
+          result
+        end
+        stub.proxy(chunk).rollback do |result|
+          assert_false(chunk.closed?)
+          result
+        end
+        chunk
+      end
+
+      m = @p.metadata(timekey: Time.parse('2016-04-11 16:40:00 +0000').to_i)
+      small_row = "x" * 1024 * 400
+      big_row = "x" * 1024 * 1024 * 8 # just `chunk_size_limit`, it does't cause BufferOverFlowError.
+
+      # Write 42 events in 1 event stream, last one is for triggering `ShouldRetry`
+      @p.write({m => [small_row] * 40 + [big_row] + ["x"]})
+
+      # Above event strem will be splitted twice by `Buffer#write_step_by_step`
+      #
+      # 1. `write_once`: 42 [events] * 1 [stream]
+      # 2. `write_step_by_step`: 4 [events]* 10 [streams] + 2 [events] * 1 [stream]
+      # 3. `write_step_by_step` (by `ShouldRetry`): 1 [event] * 42 [streams]
+      #
+      # The problematic data is built in the 2nd stage.
+      # In the 2nd stage, 5 streams are packed in a chunk.
+      # ((1024 * 400) [bytes] * 4 [events] * 5 [streams] = 8192000 [bytes] < `chunk_limit_size` (8MB)).
+      # So 3 chunks are used to store all data.
+      # The 1st chunk is already staged by `write_once`.
+      # The 2nd & 3rd chunks are newly created as unstaged.
+      # The 3rd chunk is purged before `ShouldRetry`, it's no problem:
+      #   https://github.com/fluent/fluentd/blob/7e9eba736ff40ad985341be800ddc46558be75f2/lib/fluent/plugin/buffer.rb#L850
+      # The 2nd chunk is purged in `rescue ShouldRetry`:
+      #   https://github.com/fluent/fluentd/blob/7e9eba736ff40ad985341be800ddc46558be75f2/lib/fluent/plugin/buffer.rb#L862
+      # It causes the issue described in https://github.com/fluent/fluentd/issues/3089#issuecomment-1811839198
+
+      assert_equal 2, purge_count
+    end
+
+    # https://github.com/fluent/fluentd/issues/4446
+    test "#write_step_by_step keeps chunks kept in locked in entire #write process" do
+      assert_equal 8 * 1024 * 1024, @p.chunk_limit_size
+      assert_equal 0.95, @p.chunk_full_threshold
+
+      mon_enter_counts_by_chunk = {}
+      mon_exit_counts_by_chunk = {}
+
+      stub.proxy(@p).generate_chunk(anything) do |chunk|
+        stub(chunk).mon_enter do
+          enter_count = 1 + mon_enter_counts_by_chunk.fetch(chunk, 0)
+          exit_count = mon_exit_counts_by_chunk.fetch(chunk, 0)
+          mon_enter_counts_by_chunk[chunk] = enter_count
+
+          # Assert that chunk is passed to &block of write_step_by_step before exiting the lock.
+          # (i.e. The lock count must be 2 greater than the exit count).
+          # Since ShouldRetry occurs once, the staged chunk takes the lock 3 times when calling the block.
+          if chunk.staged?
+            lock_in_block = enter_count == 3
+            assert_equal(enter_count - 2, exit_count) if lock_in_block
+          else
+            lock_in_block = enter_count == 2
+            assert_equal(enter_count - 2, exit_count) if lock_in_block
+          end
+        end
+        stub(chunk).mon_exit do
+          exit_count = 1 + mon_exit_counts_by_chunk.fetch(chunk, 0)
+          mon_exit_counts_by_chunk[chunk] = exit_count
+        end
+        chunk
+      end
+
+      m = @p.metadata(timekey: Time.parse('2016-04-11 16:40:00 +0000').to_i)
+      small_row = "x" * 1024 * 400
+      big_row = "x" * 1024 * 1024 * 8 # just `chunk_size_limit`, it does't cause BufferOverFlowError.
+
+      # Write 42 events in 1 event stream, last one is for triggering `ShouldRetry`
+      @p.write({m => [small_row] * 40 + [big_row] + ["x"]})
+
+      # Above event strem will be splitted twice by `Buffer#write_step_by_step`
+      #
+      # 1. `write_once`: 42 [events] * 1 [stream]
+      # 2. `write_step_by_step`: 4 [events]* 10 [streams] + 2 [events] * 1 [stream]
+      # 3. `write_step_by_step` (by `ShouldRetry`): 1 [event] * 42 [streams]
+      # 
+      # Example of staged chunk lock behavior:
+      # 
+      # 1. mon_enter in write_step_by_step
+      # 2. ShouldRetry occurs
+      # 3. mon_exit in write_step_by_step
+      # 4. mon_enter again in write_step_by_step (retry)
+      # 5. passed to &block of write_step_by_step
+      # 6. mon_enter in the block (write)
+      # 7. mon_exit in write_step_by_step
+      # 8. mon_exit in write
+
+      assert_equal(mon_enter_counts_by_chunk.values, mon_exit_counts_by_chunk.values)
+    end
   end
 
   sub_test_case 'standard format with configuration for test with lower chunk limit size' do
diff --git a/test/plugin/test_in_forward.rb b/test/plugin/test_in_forward.rb
index bce0cdc857..63ac8d112c 100644
--- a/test/plugin/test_in_forward.rb
+++ b/test/plugin/test_in_forward.rb
@@ -18,7 +18,8 @@ def setup
     Fluent::Test.setup
     @responses = []  # for testing responses after sending data
     @d = nil
-    @port = unused_port
+    # forward plugin uses TCP and UDP sockets on the same port number
+    @port = unused_port(protocol: :all)
   end
 
   def teardown
diff --git a/test/plugin/test_in_http.rb b/test/plugin/test_in_http.rb
index e66845520a..13cb196241 100644
--- a/test/plugin/test_in_http.rb
+++ b/test/plugin/test_in_http.rb
@@ -18,7 +18,7 @@ def shutdown
 
   def setup
     Fluent::Test.setup
-    @port = unused_port
+    @port = unused_port(protocol: :tcp)
   end
 
   def teardown
diff --git a/test/plugin/test_in_monitor_agent.rb b/test/plugin/test_in_monitor_agent.rb
index 76e0a26a18..eb8d2dea7c 100644
--- a/test/plugin/test_in_monitor_agent.rb
+++ b/test/plugin/test_in_monitor_agent.rb
@@ -392,7 +392,7 @@ def test_enable_input_metrics(with_config)
     end
 
     test "emit" do
-      port = unused_port
+      port = unused_port(protocol: :tcp)
       d = create_driver("
   @type monitor_agent
   bind '127.0.0.1'
@@ -451,7 +451,7 @@ def get(uri, header = {})
 
   sub_test_case "servlets" do
     setup do
-      @port = unused_port
+      @port = unused_port(protocol: :tcp)
       # check @type and type in one configuration
       conf = <<-EOC
 <source>
@@ -759,7 +759,7 @@ def write(chunk)
     end
 
     setup do
-      @port = unused_port
+      @port = unused_port(protocol: :tcp)
       # check @type and type in one configuration
       conf = <<-EOC
 <source>
@@ -840,7 +840,7 @@ def write(chunk)
 
   sub_test_case "check the port number of http server" do
     test "on single worker environment" do
-      port = unused_port
+      port = unused_port(protocol: :tcp)
       d = create_driver("
   @type monitor_agent
   bind '127.0.0.1'
@@ -851,7 +851,7 @@ def write(chunk)
     end
 
     test "worker_id = 2 on multi worker environment" do
-      port = unused_port
+      port = unused_port(protocol: :tcp)
       Fluent::SystemConfig.overwrite_system_config('workers' => 4) do
         d = Fluent::Test::Driver::Input.new(Fluent::Plugin::MonitorAgentInput)
         d.instance.instance_eval{ @_fluentd_worker_id = 2 }
@@ -905,7 +905,7 @@ def filter(tag, time, record)
     end
 
     test "plugins have a variable named buffer does not throws NoMethodError" do
-      port = unused_port
+      port = unused_port(protocol: :tcp)
       d = create_driver("
   @type monitor_agent
   bind '127.0.0.1'
diff --git a/test/plugin/test_in_syslog.rb b/test/plugin/test_in_syslog.rb
index f715e1fef9..868e77bb04 100755
--- a/test/plugin/test_in_syslog.rb
+++ b/test/plugin/test_in_syslog.rb
@@ -5,24 +5,24 @@
 class SyslogInputTest < Test::Unit::TestCase
   def setup
     Fluent::Test.setup
-    @port = unused_port
+    @port = unused_port(protocol: :udp)
   end
 
   def teardown
     @port = nil
   end
 
-  def ipv4_config
+  def ipv4_config(port = @port)
     %[
-      port #{@port}
+      port #{port}
       bind 127.0.0.1
       tag syslog
     ]
   end
 
-  def ipv6_config
+  def ipv6_config(port = @port)
     %[
-      port #{@port}
+      port #{port}
       bind ::1
       tag syslog
     ]
@@ -69,7 +69,8 @@ def test_configure_resolve_hostname(param)
        'Use transport and protocol' => ["protocol_type udp\n<transport tcp>\n </transport>", :udp, :tcp])
   def test_configure_protocol(param)
     conf, proto_type, transport_proto_type = *param
-    d = create_driver([ipv4_config, conf].join("\n"))
+    port = unused_port(protocol: proto_type ? proto_type : transport_proto_type)
+    d = create_driver([ipv4_config(port), conf].join("\n"))
 
     assert_equal(d.instance.protocol_type, proto_type)
     assert_equal(d.instance.transport_config.protocol, transport_proto_type)
@@ -158,12 +159,13 @@ def test_msg_size_udp_for_large_msg
   end
 
   def test_msg_size_with_tcp
-    d = create_driver([ipv4_config, "<transport tcp> \n</transport>"].join("\n"))
+    port = unused_port(protocol: :tcp)
+    d = create_driver([ipv4_config(port), "<transport tcp> \n</transport>"].join("\n"))
     tests = create_test_case
 
     d.run(expect_emits: 2) do
       tests.each {|test|
-        TCPSocket.open('127.0.0.1', @port) do |s|
+        TCPSocket.open('127.0.0.1', port) do |s|
           s.send(test['msg'], 0)
         end
       }
@@ -189,11 +191,12 @@ def test_emit_rfc5452
   end
 
   def test_msg_size_with_same_tcp_connection
-    d = create_driver([ipv4_config, "<transport tcp> \n</transport>"].join("\n"))
+    port = unused_port(protocol: :tcp)
+    d = create_driver([ipv4_config(port), "<transport tcp> \n</transport>"].join("\n"))
     tests = create_test_case
 
     d.run(expect_emits: 2) do
-      TCPSocket.open('127.0.0.1', @port) do |s|
+      TCPSocket.open('127.0.0.1', port) do |s|
         tests.each {|test|
           s.send(test['msg'], 0)
         }
@@ -347,12 +350,13 @@ def compare_test_result(events, tests, options = {})
 
   sub_test_case 'octet counting frame' do
     def test_msg_size_with_tcp
-      d = create_driver([ipv4_config, "<transport tcp> \n</transport>", 'frame_type octet_count'].join("\n"))
+      port = unused_port(protocol: :tcp)
+      d = create_driver([ipv4_config(port), "<transport tcp> \n</transport>", 'frame_type octet_count'].join("\n"))
       tests = create_test_case
 
       d.run(expect_emits: 2) do
         tests.each {|test|
-          TCPSocket.open('127.0.0.1', @port) do |s|
+          TCPSocket.open('127.0.0.1', port) do |s|
             s.send(test['msg'], 0)
           end
         }
@@ -363,11 +367,12 @@ def test_msg_size_with_tcp
     end
 
     def test_msg_size_with_same_tcp_connection
-      d = create_driver([ipv4_config, "<transport tcp> \n</transport>", 'frame_type octet_count'].join("\n"))
+      port = unused_port(protocol: :tcp)
+      d = create_driver([ipv4_config(port), "<transport tcp> \n</transport>", 'frame_type octet_count'].join("\n"))
       tests = create_test_case
 
       d.run(expect_emits: 2) do
-        TCPSocket.open('127.0.0.1', @port) do |s|
+        TCPSocket.open('127.0.0.1', port) do |s|
           tests.each {|test|
             s.send(test['msg'], 0)
           }
@@ -469,7 +474,8 @@ def test_emit_unmatched_lines_with_address
   end
 
   def test_send_keepalive_packet_is_disabled_by_default
-    d = create_driver(ipv4_config + %[
+    port = unused_port(protocol: :tcp)
+    d = create_driver(ipv4_config(port) + %[
       <transport tcp>
       </transport>
       protocol tcp
@@ -479,19 +485,20 @@ def test_send_keepalive_packet_is_disabled_by_default
 
   def test_send_keepalive_packet_can_be_enabled
     addr = "127.0.0.1"
-    d = create_driver(ipv4_config + %[
+    port = unused_port(protocol: :tcp)
+    d = create_driver(ipv4_config(port) + %[
       <transport tcp>
       </transport>
       send_keepalive_packet true
     ])
     assert_true d.instance.send_keepalive_packet
     mock.proxy(d.instance).server_create_connection(
-      :in_syslog_tcp_server, @port,
+      :in_syslog_tcp_server, port,
       bind: addr,
       resolve_name: nil,
       send_keepalive_packet: true)
     d.run do
-      TCPSocket.open(addr, @port)
+      TCPSocket.open(addr, port)
     end
   end
 
diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb
index 1546bf9d4d..58006c0b98 100644
--- a/test/plugin/test_in_tail.rb
+++ b/test/plugin/test_in_tail.rb
@@ -3016,5 +3016,273 @@ def test_path_resurrection
         },
       )
     end
+
+    def test_next_rotation_occurs_very_fast_while_old_TW_still_waiting_rotate_wait
+      config = config_element(
+        "ROOT",
+        "",
+        {
+          "path" => "#{@tmp_dir}/tail.txt*",
+          "pos_file" => "#{@tmp_dir}/tail.pos",
+          "tag" => "t1",
+          "format" => "none",
+          "read_from_head" => "true",
+          "follow_inodes" => "true",
+          "rotate_wait" => "3s",
+          "refresh_interval" => "1h",
+          # stat_watcher often calls `TailWatcher::on_notify` faster than creating a new log file,
+          # so disable it in order to reproduce the same condition stably.
+          "enable_stat_watcher" => "false",
+        }
+      )
+      d = create_driver(config, false)
+
+      tail_watchers = []
+      stub.proxy(d.instance).setup_watcher do |tw|
+        tail_watchers.append(tw)
+        mock.proxy(tw).close.once # Note: Currently, there is no harm in duplicate calls.
+        tw
+      end
+
+      Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file1 log1"}
+
+      d.run(expect_records: 6, timeout: 15) do
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file1 log2"}
+
+        sleep 1.5 # Need to be larger than 1s (the interval of watch_timer)
+
+        FileUtils.move("#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt" + "1")
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file2 log1"}
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file2 log2"}
+
+        sleep 1.5 # Need to be larger than 1s (the interval of watch_timer)
+
+        # Rotate again (Old TailWatcher waiting rotate_wait also calls update_watcher)
+        [1, 0].each do |i|
+          FileUtils.move("#{@tmp_dir}/tail.txt#{i}", "#{@tmp_dir}/tail.txt#{i + 1}")
+        end
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file3 log1"}
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file3 log2"}
+
+        # Wait rotate_wait to confirm that TailWatcher.close is not called in duplicate.
+        # (Note: Currently, there is no harm in duplicate calls)
+        sleep 4
+      end
+
+      inode_0 = tail_watchers[0]&.ino
+      inode_1 = tail_watchers[1]&.ino
+      inode_2 = tail_watchers[2]&.ino
+      record_values = d.events.collect { |event| event[2]["message"] }.sort
+      position_entries = []
+      Fluent::FileWrapper.open("#{@tmp_dir}/tail.pos", "r") do |f|
+        f.readlines(chomp: true).each do |line|
+          values = line.split("\t")
+          position_entries.append([values[0], values[1], values[2].to_i(16)])
+        end
+      end
+
+      assert_equal(
+        {
+          record_values: ["file1 log1", "file1 log2", "file2 log1", "file2 log2", "file3 log1", "file3 log2"],
+          tail_watcher_paths: ["#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0"],
+          tail_watcher_inodes: [inode_0, inode_1, inode_2],
+          tail_watcher_io_handler_opened_statuses: [false, false, false],
+          position_entries: [
+            ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_0],
+            ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_1],
+            ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_2],
+          ],
+        },
+        {
+          record_values: record_values,
+          tail_watcher_paths: tail_watchers.collect { |tw| tw.path },
+          tail_watcher_inodes: tail_watchers.collect { |tw| tw.ino },
+          tail_watcher_io_handler_opened_statuses: tail_watchers.collect { |tw| tw.instance_variable_get(:@io_handler)&.opened? || false },
+          position_entries: position_entries
+        },
+      )
+    end
+  end
+
+  sub_test_case "Update watchers for rotation without follow_inodes" do
+    # The scenario where in_tail wrongly unwatches the PositionEntry.
+    # This is reported in https://github.com/fluent/fluentd/issues/3614.
+    def test_refreshTW_during_rotation
+      config = config_element(
+        "ROOT",
+        "",
+        {
+          "path" => "#{@tmp_dir}/tail.txt0",
+          "pos_file" => "#{@tmp_dir}/tail.pos",
+          "tag" => "t1",
+          "format" => "none",
+          "read_from_head" => "true",
+          # In order to detach the old watcher quickly.
+          "rotate_wait" => "3s",
+          # In order to reproduce the same condition stably, ensure that `refresh_watchers` is not
+          # called by a timer.
+          "refresh_interval" => "1h",
+          # stat_watcher often calls `TailWatcher::on_notify` faster than creating a new log file,
+          # so disable it in order to reproduce the same condition stably.
+          "enable_stat_watcher" => "false",
+        }
+      )
+      d = create_driver(config, false)
+
+      tail_watchers = []
+      stub.proxy(d.instance).setup_watcher do |tw|
+        tail_watchers.append(tw)
+        tw
+      end
+
+      Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file1 log1"}
+
+      d.run(expect_records: 6, timeout: 15) do
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file1 log2"}
+        FileUtils.move("#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt" + "1")
+
+        # This reproduces the following situation:
+        #     `refresh_watchers` is called during the rotation process and it detects the current file being lost.
+        #     Then it stops and unwatches the TailWatcher.
+        d.instance.refresh_watchers
+
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file2 log1"}
+
+        # `watch_timer` calls `TailWatcher::on_notify`, and then `update_watcher` trys to add the new TailWatcher.
+        # After `rotate_wait` interval, the PositionEntry is unwatched.
+        # HOWEVER, the new TailWatcher is still using that PositionEntry, so this breaks the PositionFile!!
+        # That PositionEntry is removed from `PositionFile::map`, but it is still working and remaining in the real pos file.
+        sleep 5
+
+        # Append to the new current log file.
+        # The PositionEntry is updated although it does not exist in `PositionFile::map`.
+        #     `PositionFile::map`: empty
+        #     Real pos file: `.../tail.txt 0000000000000016 (inode)`
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file2 log2"}
+
+        # Rotate again
+        [1, 0].each do |i|
+          FileUtils.move("#{@tmp_dir}/tail.txt#{i}", "#{@tmp_dir}/tail.txt#{i + 1}")
+        end
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file3 log1"}
+
+        # `watch_timer` calls `TailWatcher::on_notify`, and then `update_watcher` trys to update the TailWatcher.
+        sleep 3
+
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file3 log2"}
+      end
+
+      inode_0 = tail_watchers[0]&.ino
+      inode_1 = tail_watchers[1]&.ino
+      inode_2 = tail_watchers[2]&.ino
+      record_values = d.events.collect { |event| event[2]["message"] }.sort
+      position_entries = []
+      Fluent::FileWrapper.open("#{@tmp_dir}/tail.pos", "r") do |f|
+        f.readlines(chomp: true).each do |line|
+          values = line.split("\t")
+          position_entries.append([values[0], values[1], values[2].to_i(16)])
+        end
+      end
+
+      assert_equal(
+        {
+          record_values: ["file1 log1", "file1 log2", "file2 log1", "file2 log2", "file3 log1", "file3 log2"],
+          tail_watcher_paths: ["#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0"],
+          tail_watcher_inodes: [inode_0, inode_1, inode_2],
+          tail_watcher_io_handler_opened_statuses: [false, false, false],
+          position_entries: [
+            # The recorded path is old, but it is no problem. The path is not used when using follow_inodes.
+            ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_2],
+          ],
+        },
+        {
+          record_values: record_values,
+          tail_watcher_paths: tail_watchers.collect { |tw| tw.path },
+          tail_watcher_inodes: tail_watchers.collect { |tw| tw.ino },
+          tail_watcher_io_handler_opened_statuses: tail_watchers.collect { |tw| tw.instance_variable_get(:@io_handler)&.opened? || false },
+          position_entries: position_entries
+        },
+      )
+    end
+
+    def test_next_rotation_occurs_very_fast_while_old_TW_still_waiting_rotate_wait
+      config = config_element(
+        "ROOT",
+        "",
+        {
+          "path" => "#{@tmp_dir}/tail.txt0",
+          "pos_file" => "#{@tmp_dir}/tail.pos",
+          "tag" => "t1",
+          "format" => "none",
+          "read_from_head" => "true",
+          "rotate_wait" => "3s",
+          "refresh_interval" => "1h",
+        }
+      )
+      d = create_driver(config, false)
+
+      tail_watchers = []
+      stub.proxy(d.instance).setup_watcher do |tw|
+        tail_watchers.append(tw)
+        mock.proxy(tw).close.once # Note: Currently, there is no harm in duplicate calls.
+        tw
+      end
+
+      Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file1 log1"}
+
+      d.run(expect_records: 6, timeout: 15) do
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file1 log2"}
+
+        sleep 1.5 # Need to be larger than 1s (the interval of watch_timer)
+
+        FileUtils.move("#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt" + "1")
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file2 log1"}
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file2 log2"}
+
+        sleep 1.5 # Need to be larger than 1s (the interval of watch_timer)
+
+        # Rotate again (Old TailWatcher waiting rotate_wait also calls update_watcher)
+        [1, 0].each do |i|
+          FileUtils.move("#{@tmp_dir}/tail.txt#{i}", "#{@tmp_dir}/tail.txt#{i + 1}")
+        end
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "wb") {|f| f.puts "file3 log1"}
+        Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file3 log2"}
+
+        # Wait rotate_wait to confirm that TailWatcher.close is not called in duplicate.
+        # (Note: Currently, there is no harm in duplicate calls)
+        sleep 4
+      end
+
+      inode_0 = tail_watchers[0]&.ino
+      inode_1 = tail_watchers[1]&.ino
+      inode_2 = tail_watchers[2]&.ino
+      record_values = d.events.collect { |event| event[2]["message"] }.sort
+      position_entries = []
+      Fluent::FileWrapper.open("#{@tmp_dir}/tail.pos", "r") do |f|
+        f.readlines(chomp: true).each do |line|
+          values = line.split("\t")
+          position_entries.append([values[0], values[1], values[2].to_i(16)])
+        end
+      end
+
+      assert_equal(
+        {
+          record_values: ["file1 log1", "file1 log2", "file2 log1", "file2 log2", "file3 log1", "file3 log2"],
+          tail_watcher_paths: ["#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0", "#{@tmp_dir}/tail.txt0"],
+          tail_watcher_inodes: [inode_0, inode_1, inode_2],
+          tail_watcher_io_handler_opened_statuses: [false, false, false],
+          position_entries: [
+            ["#{@tmp_dir}/tail.txt0", "0000000000000016", inode_2],
+          ],
+        },
+        {
+          record_values: record_values,
+          tail_watcher_paths: tail_watchers.collect { |tw| tw.path },
+          tail_watcher_inodes: tail_watchers.collect { |tw| tw.ino },
+          tail_watcher_io_handler_opened_statuses: tail_watchers.collect { |tw| tw.instance_variable_get(:@io_handler)&.opened? || false },
+          position_entries: position_entries
+        },
+      )
+    end
   end
 end
diff --git a/test/plugin/test_in_tcp.rb b/test/plugin/test_in_tcp.rb
index c1d917332e..a27e00bba3 100755
--- a/test/plugin/test_in_tcp.rb
+++ b/test/plugin/test_in_tcp.rb
@@ -5,7 +5,7 @@
 class TcpInputTest < Test::Unit::TestCase
   def setup
     Fluent::Test.setup
-    @port = unused_port
+    @port = unused_port(protocol: :tcp)
   end
 
   def teardown
diff --git a/test/plugin/test_in_udp.rb b/test/plugin/test_in_udp.rb
old mode 100755
new mode 100644
index dcc63e1f8f..34c9c3ac26
--- a/test/plugin/test_in_udp.rb
+++ b/test/plugin/test_in_udp.rb
@@ -5,7 +5,7 @@
 class UdpInputTest < Test::Unit::TestCase
   def setup
     Fluent::Test.setup
-    @port = unused_port
+    @port = unused_port(protocol: :udp)
   end
 
   def teardown
@@ -268,25 +268,31 @@ def create_udp_socket(host, port)
 
   test 'message_length_limit' do
     message_length_limit = 32
+
+    if Fluent.windows?
+      expected_records = ["0" * 30, "4" * 30]
+    else
+      expected_records = 1.upto(3).collect do |i|
+        "#{i}" * message_length_limit
+      end
+      expected_records.prepend("0" * 30)
+      expected_records.append("4" * 30)
+    end
+
     d = create_driver(base_config + %!
       format none
       message_length_limit #{message_length_limit}
     !)
-    d.run(expect_records: 3) do
+    d.run(expect_records: expected_records.size, timeout: 5) do
       create_udp_socket('127.0.0.1', @port) do |u|
-        3.times do |i|
+        u.send("0" * 30 + "\n", 0)
+        1.upto(3) do |i|
           u.send("#{i}" * 40 + "\n", 0)
         end
+        u.send("4" * 30 + "\n", 0)
       end
     end
 
-    if Fluent.windows?
-      expected_records = []
-    else
-      expected_records = 3.times.collect do |i|
-        "#{i}" * message_length_limit
-      end
-    end
     actual_records = d.events.collect do |event|
       event[2]["message"]
     end
diff --git a/test/plugin/test_out_file.rb b/test/plugin/test_out_file.rb
index 5727bc1e69..8b6ab161d0 100644
--- a/test/plugin/test_out_file.rb
+++ b/test/plugin/test_out_file.rb
@@ -130,7 +130,7 @@ def create_driver(conf = CONFIG, opts = {})
           'path' => "#{TMP_DIR}/${tag}/${type}/conf_test.%Y%m%d.%H%M.log",
           'add_path_suffix' => 'false',
           'append' => "true",
-          'symlink_path' => "#{TMP_DIR}/conf_test.current.log",
+          'symlink_path' => "#{TMP_DIR}/${tag}/conf_test.current.log",
           'compress' => 'gzip',
           'recompress' => 'true',
         }, [
@@ -183,6 +183,26 @@ def create_driver(conf = CONFIG, opts = {})
         Fluent::Test::Driver::Output.new(Fluent::Plugin::NullOutput).configure(conf)
       end
     end
+
+    test 'warning for symlink_path not including correct placeholders corresponding to chunk keys' do
+      omit "Windows doesn't support symlink" if Fluent.windows?
+      conf = config_element('match', '**', {
+          'path' => "#{TMP_DIR}/${tag}/${key1}/${key2}/conf_test.%Y%m%d.%H%M.log",
+          'symlink_path' => "#{TMP_DIR}/conf_test.current.log",
+        }, [
+          config_element('buffer', 'time,tag,key1,key2', {
+              '@type' => 'file',
+              'timekey' => '1d',
+              'path' => "#{TMP_DIR}/buf_conf_test",
+          }),
+      ])
+      assert_nothing_raised do
+        d = create_driver(conf)
+        assert do
+          d.logs.count { |log| log.include?("multiple chunks are competing for a single symlink_path") } == 2
+        end
+      end
+    end
   end
 
   sub_test_case 'fully configured output' do
diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb
index 80557c5989..c2442b631a 100644
--- a/test/plugin/test_out_forward.rb
+++ b/test/plugin/test_out_forward.rb
@@ -12,7 +12,8 @@ def setup
     FileUtils.rm_rf(TMP_DIR)
     FileUtils.mkdir_p(TMP_DIR)
     @d = nil
-    @target_port = unused_port
+    # forward plugin uses TCP and UDP sockets on the same port number
+    @target_port = unused_port(protocol: :all)
   end
 
   def teardown
@@ -156,7 +157,14 @@ def try_write(chunk)
     normal_conf = config_element('match', '**', {}, [
         config_element('server', '', {'name' => 'test', 'host' => 'unexisting.yaaaaaaaaaaaaaay.host.example.com'})
       ])
-    assert_raise SocketError do
+
+    if Socket.const_defined?(:ResolutionError) # as of Ruby 3.3
+      error_class = Socket::ResolutionError
+    else
+      error_class = SocketError
+    end
+
+    assert_raise error_class do
       create_driver(normal_conf)
     end
 
@@ -165,7 +173,7 @@ def try_write(chunk)
       ])
     @d = d = create_driver(conf)
     expected_log = "failed to resolve node name when configured"
-    expected_detail = 'server="test" error_class=SocketError'
+    expected_detail = "server=\"test\" error_class=#{error_class.name}"
     logs = d.logs
     assert{ logs.any?{|log| log.include?(expected_log) && log.include?(expected_detail) } }
   end
@@ -603,7 +611,6 @@ def try_write(chunk)
 
     @d = d = create_driver(config + %[
       require_ack_response true
-      ack_response_timeout 1s
       <buffer tag>
         flush_mode immediate
         retry_type periodic
@@ -651,7 +658,6 @@ def try_write(chunk)
 
     @d = d = create_driver(config + %[
       require_ack_response true
-      ack_response_timeout 10s
       <buffer tag>
         flush_mode immediate
         retry_type periodic
@@ -1241,27 +1247,22 @@ def plugin_id_for_test?
     target_input_driver = create_target_input_driver(conf: target_config)
     output_conf = config
     d = create_driver(output_conf)
-    d.instance_start
 
-    begin
-      chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil))
-      mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port,
-                                               linger_timeout: anything,
-                                               send_timeout: anything,
-                                               recv_timeout: anything,
-                                               connect_timeout: anything
-                                              ) { |sock| mock(sock).close.once; sock }.twice
+    chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil))
+    mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port,
+                                             linger_timeout: anything,
+                                             send_timeout: anything,
+                                             recv_timeout: anything,
+                                             connect_timeout: anything
+                                            ) { |sock| mock(sock).close.once; sock }.twice
 
-      target_input_driver.run(timeout: 15) do
-        d.run(shutdown: false) do
-          node = d.instance.nodes.first
-          2.times do
-            node.send_data('test', chunk) rescue nil
-          end
+    target_input_driver.run(timeout: 15) do
+      d.run do
+        node = d.instance.nodes.first
+        2.times do
+          node.send_data('test', chunk) rescue nil
         end
       end
-    ensure
-      d.instance_shutdown
     end
   end
 
@@ -1275,7 +1276,6 @@ def plugin_id_for_test?
       port #{@target_port}
     </server>
     ])
-    d.instance_start
     assert_nothing_raised { d.run }
   end
 
@@ -1287,33 +1287,28 @@ def plugin_id_for_test?
         keepalive_timeout 2
       ]
       d = create_driver(output_conf)
-      d.instance_start
 
-      begin
-        chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil))
-        mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port,
-                                                 linger_timeout: anything,
-                                                 send_timeout: anything,
-                                                 recv_timeout: anything,
-                                                 connect_timeout: anything
-                                                ) { |sock| mock(sock).close.once; sock }.once
+      chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil))
+      mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port,
+                                               linger_timeout: anything,
+                                               send_timeout: anything,
+                                               recv_timeout: anything,
+                                               connect_timeout: anything
+                                              ) { |sock| mock(sock).close.once; sock }.once
 
-        target_input_driver.run(timeout: 15) do
-          d.run(shutdown: false) do
-            node = d.instance.nodes.first
-            2.times do
-              node.send_data('test', chunk) rescue nil
-            end
+      target_input_driver.run(timeout: 15) do
+        d.run do
+          node = d.instance.nodes.first
+          2.times do
+            node.send_data('test', chunk) rescue nil
           end
         end
-      ensure
-        d.instance_shutdown
       end
     end
 
     test 'create timer of purging obsolete sockets' do
       output_conf = config + %[keepalive true]
-      d = create_driver(output_conf)
+      @d = d = create_driver(output_conf)
 
       mock(d.instance).timer_execute(:out_forward_heartbeat_request, 1).once
       mock(d.instance).timer_execute(:out_forward_keep_alived_socket_watcher, 5).once
@@ -1329,7 +1324,6 @@ def plugin_id_for_test?
           keepalive_timeout 2
         ]
         d = create_driver(output_conf)
-        d.instance_start
 
         chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Fluent::Plugin::Buffer::Metadata.new(nil, nil, nil))
         mock.proxy(d.instance).socket_create_tcp(TARGET_HOST, @target_port,
diff --git a/test/plugin/test_out_stream.rb b/test/plugin/test_out_stream.rb
index 05e9011aab..f19328f5be 100644
--- a/test/plugin/test_out_stream.rb
+++ b/test/plugin/test_out_stream.rb
@@ -54,7 +54,7 @@ class TcpOutputTest < Test::Unit::TestCase
 
   def setup
     super
-    @port = unused_port
+    @port = unused_port(protocol: :tcp)
   end
 
   def teardown
diff --git a/test/plugin/test_parser_json.rb b/test/plugin/test_parser_json.rb
index 19c45402d1..e60784628e 100644
--- a/test/plugin/test_parser_json.rb
+++ b/test/plugin/test_parser_json.rb
@@ -8,6 +8,37 @@ def setup
     @parser = Fluent::Test::Driver::Parser.new(Fluent::Plugin::JSONParser)
   end
 
+  sub_test_case "configure_json_parser" do
+    data("oj", [:oj, [Oj.method(:load), Oj::ParseError]])
+    data("json", [:json, [JSON.method(:load), JSON::ParserError]])
+    data("yajl", [:yajl, [Yajl.method(:load), Yajl::ParseError]])
+    def test_return_each_loader((input, expected_return))
+      result = @parser.instance.configure_json_parser(input)
+      assert_equal expected_return, result
+    end
+
+    def test_raise_exception_for_unknown_input
+      assert_raise RuntimeError do
+        @parser.instance.configure_json_parser(:unknown)
+      end
+    end
+
+    def test_fall_back_oj_to_yajl_if_oj_not_available
+      stub(Fluent::OjOptions).available? { false }
+
+      result = @parser.instance.configure_json_parser(:oj)
+
+      assert_equal [Yajl.method(:load), Yajl::ParseError], result
+      logs = @parser.logs.collect do |log|
+        log.gsub(/\A\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} [-+]\d{4} /, "")
+      end
+      assert_equal(
+        ["[info]: Oj is not installed, and failing back to Yajl for json parser\n"],
+        logs
+      )
+    end
+  end
+
   data('oj' => 'oj', 'yajl' => 'yajl')
   def test_parse(data)
     @parser.configure('json_parser' => data)
diff --git a/test/plugin_helper/test_child_process.rb b/test/plugin_helper/test_child_process.rb
index 5d84a0fd06..b498b7a0eb 100644
--- a/test/plugin_helper/test_child_process.rb
+++ b/test/plugin_helper/test_child_process.rb
@@ -515,6 +515,9 @@ def configure(conf)
   end
 
   test 'can scrub characters without exceptions' do
+    if Gem::Version.create(RUBY_VERSION) >= Gem::Version.create('3.3.0')
+      pend "Behaviour of IO#set_encoding is changed as of Ruby 3.3 (#4058)"
+    end
     m = Mutex.new
     str = nil
     Timeout.timeout(TEST_DEADLOCK_TIMEOUT) do
@@ -529,19 +532,25 @@ def configure(conf)
       sleep TEST_WAIT_INTERVAL_FOR_BLOCK_RUNNING until m.locked? || ran
       m.lock
       assert_equal Encoding.find('utf-8'), str.encoding
-      expected = "\xEF\xBF\xBD\xEF\xBF\xBD\x00\xEF\xBF\xBD\xEF\xBF\xBD".force_encoding("utf-8")
+      replacement = "\uFFFD" # U+FFFD (REPLACEMENT CHARACTER)
+      nul = "\x00" # U+0000 (NUL)
+      expected = replacement * 2 + nul + replacement * 2
       assert_equal expected, str
       @d.stop; @d.shutdown; @d.close; @d.terminate
     end
   end
 
   test 'can scrub characters without exceptions and replace specified chars' do
+    if Gem::Version.create(RUBY_VERSION) >= Gem::Version.create('3.3.0')
+      pend "Behaviour of IO#set_encoding is changed as of Ruby 3.3 (#4058)"
+    end
     m = Mutex.new
     str = nil
+    replacement = "?"
     Timeout.timeout(TEST_DEADLOCK_TIMEOUT) do
       ran = false
       args = ['-e', 'STDOUT.set_encoding("ascii-8bit"); STDOUT.write "\xFF\xFF\x00\xF0\xF0"']
-      @d.child_process_execute(:t13b, "ruby", arguments: args, mode: [:read], scrub: true, replace_string: '?') do |io|
+      @d.child_process_execute(:t13b, "ruby", arguments: args, mode: [:read], scrub: true, replace_string: replacement) do |io|
         m.lock
         ran = true
         str = io.read
@@ -550,7 +559,8 @@ def configure(conf)
       sleep TEST_WAIT_INTERVAL_FOR_BLOCK_RUNNING until m.locked? || ran
       m.lock
       assert_equal Encoding.find('utf-8'), str.encoding
-      expected = "??\x00??".force_encoding("utf-8")
+      nul = "\x00" # U+0000 (NUL)
+      expected = replacement * 2 + nul + replacement * 2
       assert_equal expected, str
       @d.stop; @d.shutdown; @d.close; @d.terminate
     end
diff --git a/test/plugin_helper/test_http_server_helper.rb b/test/plugin_helper/test_http_server_helper.rb
index 4fd044bb0e..815f3b18fd 100644
--- a/test/plugin_helper/test_http_server_helper.rb
+++ b/test/plugin_helper/test_http_server_helper.rb
@@ -14,7 +14,7 @@ class HttpHelperTest < Test::Unit::TestCase
   CERT_CA_DIR = File.expand_path(File.dirname(__FILE__) + '/data/cert/with_ca')
 
   def setup
-    @port = unused_port
+    @port = unused_port(protocol: :tcp)
   end
 
   def teardown
diff --git a/test/plugin_helper/test_server.rb b/test/plugin_helper/test_server.rb
index 347938eca3..63beb0bd70 100644
--- a/test/plugin_helper/test_server.rb
+++ b/test/plugin_helper/test_server.rb
@@ -15,7 +15,7 @@ class Dummy < Fluent::Plugin::TestBase
   TMP_DIR = File.expand_path(File.dirname(__FILE__) + "/../tmp/plugin_helper_server")
 
   setup do
-    @port = unused_port
+    @port = unused_port(protocol: :tcp)
     if Fluent.windows?
       @socket_manager_server = ServerEngine::SocketManager::Server.open
       @socket_manager_path = @socket_manager_server.path
@@ -233,11 +233,12 @@ class Dummy < Fluent::Plugin::TestBase
       # 'server_create_connection tcp' => [:server_create_connection, :unix],
     )
     test 'raise error if udp options specified for tcp/tls/unix' do |(m, proto)|
+      port = unused_port(protocol: proto)
       assert_raise ArgumentError do
-        @d.__send__(m, :myserver, @port, proto: proto, max_bytes: 128){|x| x }
+        @d.__send__(m, :myserver, port, proto: proto, max_bytes: 128){|x| x }
       end
       assert_raise ArgumentError do
-        @d.__send__(m, :myserver, @port, proto: proto, flags: 1){|x| x }
+        @d.__send__(m, :myserver, port, proto: proto, flags: 1){|x| x }
       end
     end
 
@@ -245,8 +246,9 @@ class Dummy < Fluent::Plugin::TestBase
       'server_create udp' => [:server_create, :udp],
     )
     test 'raise error if tcp/tls options specified for udp' do |(m, proto)|
+      port = unused_port(protocol: proto)
       assert_raise(ArgumentError.new("BUG: linger_timeout is available for tcp/tls")) do
-        @d.__send__(m, :myserver, @port, proto: proto, linger_timeout: 1, max_bytes: 128){|x| x }
+        @d.__send__(m, :myserver, port, proto: proto, linger_timeout: 1, max_bytes: 128){|x| x }
       end
     end
 
@@ -254,8 +256,9 @@ class Dummy < Fluent::Plugin::TestBase
       'server_create udp' => [:server_create, :udp],
     )
     test 'raise error if tcp/tls/unix backlog options specified for udp' do |(m, proto)|
+      port = unused_port(protocol: proto)
       assert_raise(ArgumentError.new("BUG: backlog is available for tcp/tls")) do
-        @d.__send__(m, :myserver, @port, proto: proto, backlog: 500){|x| x }
+        @d.__send__(m, :myserver, port, proto: proto, backlog: 500){|x| x }
       end
     end
 
@@ -263,8 +266,9 @@ class Dummy < Fluent::Plugin::TestBase
       'server_create udp' => [:server_create, :udp],
     )
     test 'raise error if tcp/tls send_keepalive_packet option is specified for udp' do |(m, proto)|
+      port = unused_port(protocol: proto)
       assert_raise(ArgumentError.new("BUG: send_keepalive_packet is available for tcp/tls")) do
-        @d.__send__(m, :myserver, @port, proto: proto, send_keepalive_packet: true){|x| x }
+        @d.__send__(m, :myserver, port, proto: proto, send_keepalive_packet: true){|x| x }
       end
     end
 
@@ -276,8 +280,9 @@ class Dummy < Fluent::Plugin::TestBase
       # 'server_create_connection unix' => [:server_create_connection, :unix, {}],
     )
     test 'raise error if tls options specified for tcp/udp/unix' do |(m, proto, kwargs)|
+      port = unused_port(protocol: proto)
       assert_raise(ArgumentError.new("BUG: tls_options is available only for tls")) do
-        @d.__send__(m, :myserver, @port, proto: proto, tls_options: {}, **kwargs){|x| x }
+        @d.__send__(m, :myserver, port, proto: proto, tls_options: {}, **kwargs){|x| x }
       end
     end
 
@@ -289,7 +294,8 @@ class Dummy < Fluent::Plugin::TestBase
       'server_create_connection tls' => [:server_create_connection, :tls, {tls_options: {insecure: true}}],
     )
     test 'can bind specified IPv4 address' do |(m, proto, kwargs)|
-      @d.__send__(m, :myserver, @port, proto: proto, bind: "127.0.0.1", **kwargs){|x| x }
+      port = unused_port(protocol: proto)
+      @d.__send__(m, :myserver, port, proto: proto, bind: "127.0.0.1", **kwargs){|x| x }
       assert_equal "127.0.0.1", @d._servers.first.bind
       assert_equal "127.0.0.1", @d._servers.first.server.instance_eval{ instance_variable_defined?(:@listen_socket) ? @listen_socket : @_io }.addr[3]
     end
@@ -303,7 +309,8 @@ class Dummy < Fluent::Plugin::TestBase
     )
     test 'can bind specified IPv6 address' do |(m, proto, kwargs)| # if available
       omit "IPv6 unavailable here" unless ipv6_enabled?
-      @d.__send__(m, :myserver, @port, proto: proto, bind: "::1", **kwargs){|x| x }
+      port = unused_port(protocol: proto)
+      @d.__send__(m, :myserver, port, proto: proto, bind: "::1", **kwargs){|x| x }
       assert_equal "::1", @d._servers.first.bind
       assert_equal "::1", @d._servers.first.server.instance_eval{ instance_variable_defined?(:@listen_socket) ? @listen_socket : @_io }.addr[3]
     end
@@ -320,10 +327,11 @@ class Dummy < Fluent::Plugin::TestBase
     test 'can create 2 or more servers which share same bind address and port if shared option is true' do |(m, proto, kwargs)|
       begin
         d2 = Dummy.new; d2.start; d2.after_start
+        port = unused_port(protocol: proto)
 
         assert_nothing_raised do
-          @d.__send__(m, :myserver, @port, proto: proto, **kwargs){|x| x }
-          d2.__send__(m, :myserver, @port, proto: proto, **kwargs){|x| x }
+          @d.__send__(m, :myserver, port, proto: proto, **kwargs){|x| x }
+          d2.__send__(m, :myserver, port, proto: proto, **kwargs){|x| x }
         end
       ensure
         d2.stop; d2.before_shutdown; d2.shutdown; d2.after_shutdown; d2.close; d2.terminate
@@ -344,12 +352,13 @@ class Dummy < Fluent::Plugin::TestBase
     test 'cannot create 2 or more servers using same bind address and port if shared option is false' do |(m, proto, kwargs)|
       begin
         d2 = Dummy.new; d2.start; d2.after_start
+        port = unused_port(protocol: proto)
 
         assert_nothing_raised do
-          @d.__send__(m, :myserver, @port, proto: proto, shared: false, **kwargs){|x| x }
+          @d.__send__(m, :myserver, port, proto: proto, shared: false, **kwargs){|x| x }
         end
         assert_raise(Errno::EADDRINUSE, Errno::EACCES) do
-          d2.__send__(m, :myserver, @port, proto: proto, **kwargs){|x| x }
+          d2.__send__(m, :myserver, port, proto: proto, **kwargs){|x| x }
         end
       ensure
         d2.stop; d2.before_shutdown; d2.shutdown; d2.after_shutdown; d2.close; d2.terminate
@@ -365,16 +374,18 @@ class Dummy < Fluent::Plugin::TestBase
       # 'unix' => [:unix, {}],
     )
     test 'raise error if block argument is not specified or too many' do |(proto, kwargs)|
+      port = unused_port(protocol: proto)
       assert_raise(ArgumentError.new("BUG: block must have 1 or 2 arguments")) do
-        @d.server_create(:myserver, @port, proto: proto, **kwargs){ 1 }
+        @d.server_create(:myserver, port, proto: proto, **kwargs){ 1 }
       end
       assert_raise(ArgumentError.new("BUG: block must have 1 or 2 arguments")) do
-        @d.server_create(:myserver, @port, proto: proto, **kwargs){|sock, conn, what_is_this| 1 }
+        @d.server_create(:myserver, port, proto: proto, **kwargs){|sock, conn, what_is_this| 1 }
       end
     end
 
     test 'creates udp server if specified in proto' do
-      @d.server_create(:myserver, @port, proto: :udp, max_bytes: 512){|x| x }
+      port = unused_port(protocol: :udp)
+      @d.server_create(:myserver, port, proto: :udp, max_bytes: 512){|x| x }
 
       created_server_info = @d._servers.first
       assert_equal :udp, created_server_info.proto
@@ -587,7 +598,8 @@ class Dummy < Fluent::Plugin::TestBase
   sub_test_case '#server_create_udp' do
     test 'can accept all keyword arguments valid for udp server' do
       assert_nothing_raised do
-        @d.server_create_udp(:s, @port, bind: '127.0.0.1', shared: false, resolve_name: true, max_bytes: 100, flags: 1) do |data, conn|
+        port = unused_port(protocol: :udp)
+        @d.server_create_udp(:s, port, bind: '127.0.0.1', shared: false, resolve_name: true, max_bytes: 100, flags: 1) do |data, conn|
           # ...
         end
       end
@@ -595,14 +607,15 @@ class Dummy < Fluent::Plugin::TestBase
 
     test 'creates a udp server just to read data' do
       received = ""
-      @d.server_create_udp(:s, @port, max_bytes: 128) do |data|
+      port = unused_port(protocol: :udp)
+      @d.server_create_udp(:s, port, max_bytes: 128) do |data|
         received << data
       end
       bind_port = unused_port(protocol: :udp, bind: "127.0.0.1")
       3.times do
         sock = UDPSocket.new(Socket::AF_INET)
         sock.bind("127.0.0.1", bind_port)
-        sock.connect("127.0.0.1", @port)
+        sock.connect("127.0.0.1", port)
         sock.puts "yay"
         sock.puts "foo"
         sock.close
@@ -614,16 +627,17 @@ class Dummy < Fluent::Plugin::TestBase
     test 'creates a udp server to read and write data' do
       received = ""
       responses = []
-      @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock|
+      port = unused_port(protocol: :udp)
+      @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock|
         received << data
         sock.write "ack\n"
       end
-      bind_port = unused_port
+      bind_port = unused_port(protocol: :udp)
       3.times do
         begin
           sock = UDPSocket.new(Socket::AF_INET)
           sock.bind("127.0.0.1", bind_port)
-          sock.connect("127.0.0.1", @port)
+          sock.connect("127.0.0.1", port)
           th = Thread.new do
             while true
               begin
@@ -654,11 +668,12 @@ class Dummy < Fluent::Plugin::TestBase
 
       received = ""
       responses = []
-      @d.server_create_udp(:s, @port, bind: "::1", max_bytes: 128) do |data, sock|
+      port = unused_port(protocol: :udp)
+      @d.server_create_udp(:s, port, bind: "::1", max_bytes: 128) do |data, sock|
         received << data
         sock.write "ack\n"
       end
-      bind_port = unused_port
+      bind_port = unused_port(protocol: :udp)
       3.times do
         begin
           sock = UDPSocket.new(Socket::AF_INET6)
@@ -667,7 +682,7 @@ class Dummy < Fluent::Plugin::TestBase
             responses << sock.recv(16)
             true
           end
-          sock.connect("::1", @port)
+          sock.connect("::1", port)
           sock.write "yay\nfoo\n"
           th.join(5)
         ensure
@@ -682,13 +697,14 @@ class Dummy < Fluent::Plugin::TestBase
     test 'does not resolve name of client address in default' do
       received = ""
       sources = []
-      @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock|
+      port = unused_port(protocol: :udp)
+      @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock|
         received << data
         sources << sock.remote_host
       end
       3.times do
         sock = UDPSocket.new(Socket::AF_INET)
-        sock.connect("127.0.0.1", @port)
+        sock.connect("127.0.0.1", port)
         sock.puts "yay"
         sock.close
       end
@@ -702,13 +718,14 @@ class Dummy < Fluent::Plugin::TestBase
 
       received = ""
       sources = []
-      @d.server_create_udp(:s, @port, resolve_name: true, max_bytes: 128) do |data, sock|
+      port = unused_port(protocol: :udp)
+      @d.server_create_udp(:s, port, resolve_name: true, max_bytes: 128) do |data, sock|
         received << data
         sources << sock.remote_host
       end
       3.times do
         sock = UDPSocket.new(Socket::AF_INET)
-        sock.connect("127.0.0.1", @port)
+        sock.connect("127.0.0.1", port)
         sock.puts "yay"
         sock.close
       end
@@ -720,7 +737,8 @@ class Dummy < Fluent::Plugin::TestBase
     test 'raises error if plugin registers data callback for connection object from #server_create' do
       received = ""
       errors = []
-      @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock|
+      port = unused_port(protocol: :udp)
+      @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock|
         received << data
         begin
           sock.data{|d| received << d.upcase }
@@ -729,7 +747,7 @@ class Dummy < Fluent::Plugin::TestBase
         end
       end
       sock = UDPSocket.new(Socket::AF_INET)
-      sock.connect("127.0.0.1", @port)
+      sock.connect("127.0.0.1", port)
       sock.write "foo\n"
       sock.close
 
@@ -742,7 +760,8 @@ class Dummy < Fluent::Plugin::TestBase
     test 'raise error if plugin registers write_complete callback for udp' do
       received = ""
       errors = []
-      @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock|
+      port = unused_port(protocol: :udp)
+      @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock|
         received << data
         begin
           sock.on(:write_complete){|conn| "" }
@@ -751,7 +770,7 @@ class Dummy < Fluent::Plugin::TestBase
         end
       end
       sock = UDPSocket.new(Socket::AF_INET)
-      sock.connect("127.0.0.1", @port)
+      sock.connect("127.0.0.1", port)
       sock.write "foo\n"
       sock.close
 
@@ -764,7 +783,8 @@ class Dummy < Fluent::Plugin::TestBase
     test 'raises error if plugin registers close callback for udp' do
       received = ""
       errors = []
-      @d.server_create_udp(:s, @port, max_bytes: 128) do |data, sock|
+      port = unused_port(protocol: :udp)
+      @d.server_create_udp(:s, port, max_bytes: 128) do |data, sock|
         received << data
         begin
           sock.on(:close){|d| "" }
@@ -773,7 +793,7 @@ class Dummy < Fluent::Plugin::TestBase
         end
       end
       sock = UDPSocket.new(Socket::AF_INET)
-      sock.connect("127.0.0.1", @port)
+      sock.connect("127.0.0.1", port)
       sock.write "foo\n"
       sock.close
 
@@ -786,11 +806,12 @@ class Dummy < Fluent::Plugin::TestBase
     test 'can bind IPv4 / IPv6 together' do
       omit "IPv6 unavailable here" unless ipv6_enabled?
 
+      port = unused_port(protocol: :udp)
       assert_nothing_raised do
-        @d.server_create_udp(:s_ipv4_udp, @port, bind: '0.0.0.0', shared: false, max_bytes: 128) do |data, sock|
+        @d.server_create_udp(:s_ipv4_udp, port, bind: '0.0.0.0', shared: false, max_bytes: 128) do |data, sock|
           # ...
         end
-        @d.server_create_udp(:s_ipv6_udp, @port, bind: '::', shared: false, max_bytes: 128) do |data, sock|
+        @d.server_create_udp(:s_ipv6_udp, port, bind: '::', shared: false, max_bytes: 128) do |data, sock|
           # ...
         end
       end
@@ -803,11 +824,12 @@ class Dummy < Fluent::Plugin::TestBase
         max_bytes, records, expected = data.values
 
         actual_records = []
-        @d.server_create_udp(:myserver, @port, max_bytes: max_bytes) do |data, sock|
+        port = unused_port(protocol: :udp)
+        @d.server_create_udp(:myserver, port, max_bytes: max_bytes) do |data, sock|
           actual_records << data
         end
 
-        open_client(:udp, "127.0.0.1", @port) do |sock|
+        open_client(:udp, "127.0.0.1", port) do |sock|
           records.each do |record|
             sock.send(record, 0)
           end
@@ -823,11 +845,12 @@ class Dummy < Fluent::Plugin::TestBase
         max_bytes, records, expected = data.values
 
         actual_records = []
-        @d.server_create_udp(:myserver, @port, max_bytes: max_bytes) do |data|
+        port = unused_port(protocol: :udp)
+        @d.server_create_udp(:myserver, port, max_bytes: max_bytes) do |data|
           actual_records << data
         end
 
-        open_client(:udp, "127.0.0.1", @port) do |sock|
+        open_client(:udp, "127.0.0.1", port) do |sock|
           records.each do |record|
             sock.send(record, 0)
           end
diff --git a/test/plugin_helper/test_socket.rb b/test/plugin_helper/test_socket.rb
index cb2ff32de0..624169f08b 100644
--- a/test/plugin_helper/test_socket.rb
+++ b/test/plugin_helper/test_socket.rb
@@ -11,7 +11,7 @@ class SocketHelperTest < Test::Unit::TestCase
   CERT_CHAINS_DIR = File.expand_path(File.dirname(__FILE__) + '/data/cert/cert_chains')
 
   def setup
-    @port = unused_port
+    @port = unused_port(protocol: :tcp)
   end
 
   def teardown
diff --git a/test/test_config.rb b/test/test_config.rb
index 24f63830d4..7533e6e16b 100644
--- a/test/test_config.rb
+++ b/test/test_config.rb
@@ -167,6 +167,7 @@ def test_included
             tag: tag.dummy
         - source:
             $type: tcp
+            $log_level: info
             tag: tag.tcp
             parse:
               $arg:
@@ -176,6 +177,7 @@ def test_included
         - match:
             $tag: tag.*
             $type: stdout
+            $log_level: debug
             buffer:
               $type: memory
               flush_interval: 1s
@@ -208,10 +210,12 @@ def test_included
           'tag.dummy',
           'tcp',
           'tag.tcp',
+          'info',
           'none',
           'why.parse.section.doesnot.have.arg,huh',
           'stdout',
           'tag.*',
+          'debug',
           'null',
           '**',
           '@FLUENT_LOG',
@@ -224,10 +228,12 @@ def test_included
           dummy_source_conf['tag'],
           tcp_source_conf['@type'],
           tcp_source_conf['tag'],
+          tcp_source_conf['@log_level'],
           parse_tcp_conf['@type'],
           parse_tcp_conf.arg,
           match_conf['@type'],
           match_conf.arg,
+          match_conf['@log_level'],
           fluent_log_conf['@type'],
           fluent_log_conf.arg,
           label_conf.arg,
diff --git a/test/test_event_router.rb b/test/test_event_router.rb
index 3601ddf10f..fda89db230 100644
--- a/test/test_event_router.rb
+++ b/test/test_event_router.rb
@@ -175,7 +175,7 @@ def event_router
       test "don't call default collector when tag matched" do
         event_router.add_rule('test', output)
         assert_rr do
-          dont_allow(default_collector).emit_events('test', is_a(OneEventStream))
+          mock(default_collector).emit_events('test', is_a(OneEventStream)).never
           event_router.emit('test', Engine.now, 'k' => 'v')
         end
         # check emit handler doesn't catch rr error
@@ -201,7 +201,7 @@ def filter_stream(_tag, es); end
         event_router.add_rule('test', filter)
 
         assert_rr do
-          dont_allow(filter).filter_stream('test', is_a(OneEventStream)) { events }
+          mock(filter).filter_stream('test', is_a(OneEventStream)).never
           event_router.emit('foo', Engine.now, 'k' => 'v')
         end
       end