@@ -816,7 +816,7 @@ class TailInput
       none regexp json csv tsv ltsv msgpack apache apache2 apache_error nginx syslog
     ].freeze
     WORKER_QUEUE_BYTES = 64 * 1024 * 1024
-    WorkerBatch = Struct.new(:watcher, :lines, :fallback_lines)
+    WorkerBatch = Struct.new(:watcher, :lines)
 
     private
 
@@ -866,7 +866,7 @@ def setup_worker_pool
         thread_create: ->(title, &block) { Thread.new { Thread.current.name = title.to_s if Thread.current.respond_to?(:name=); block.call } },
         notify: @worker_completion_watcher.method(:signal)
       ) do |index, batch|
-        @worker_line_feeders[index].parse(batch.lines, batch.watcher)
+        @worker_line_feeders[index].parse(batch.lines.map(&:dup), batch.watcher)
       end
       @worker_pool.start
     rescue
@@ -892,12 +892,8 @@ def build_worker_line_feeder(parser)
     def async_receive_lines(lines, watcher)
       return @line_feeder.feed_lines(lines, watcher) unless @worker_pool.statistics[:state] == :running
 
-      worker_lines = lines.map(&:dup)
-      fallback_lines = lines.map(&:dup)
       bytes = lines.sum(&:bytesize) * 3
-      return @line_feeder.feed_lines(lines, watcher) if bytes > WORKER_QUEUE_BYTES
-
-      batch = WorkerBatch.new(watcher, worker_lines, fallback_lines)
+      batch = WorkerBatch.new(watcher, lines)
       if @worker_pool.submit(watcher, batch, bytes: bytes)
         update_worker_metrics
         @worker_waiting_watchers.delete(watcher)
@@ -914,7 +910,7 @@ def process_worker_completions
         if result.error
           @worker_metrics[:parse_errors].inc
           log.warn 'worker parsing failed; retrying synchronously', path: batch.watcher.path, error: result.error
-          emitted = @line_feeder.feed_lines(batch.fallback_lines, batch.watcher)
+          emitted = @line_feeder.feed_lines(batch.lines, batch.watcher)
         else
           emitted = @line_feeder.emit(result.value, batch.watcher)
         end
 
@@ -41,6 +41,8 @@ def initialize(num_threads:, task_limit:, byte_limit:, thread_create:, notify:,
         @results = Queue.new
         @active = {}
         @bytes = 0
+        @regular_bytes = 0
+        @oversized_task = nil
         @threads = []
         @state = :new
         @failure = nil
@@ -75,11 +77,22 @@ def submit(key, payload, bytes:)
         @mutex.synchronize do
           return false unless @state == :running
           return false if @active.key?(key) || @active.size >= @task_limit
-          return false if @bytes + bytes > @byte_limit
+          # A single over-limit batch may use an extra reservation alongside the
+          # bounded regular budget so it does not force parsing on the event loop.
+          if bytes > @byte_limit
+            return false if @oversized_task
+          else
+            return false if @regular_bytes + bytes > @byte_limit
+          end
 
           task = Task.new(key, bytes, payload).freeze
           @active[key] = task
           @bytes += bytes
+          if bytes > @byte_limit
+            @oversized_task = task
+          else
+            @regular_bytes += bytes
+          end
           @jobs << task
           true
         end
@@ -100,6 +113,11 @@ def acknowledge(result)
 
           @active.delete(task.key)
           @bytes -= task.bytes
+          if task.bytes > @byte_limit
+            @oversized_task = nil
+          else
+            @regular_bytes -= task.bytes
+          end
         end
       end
 
 
@@ -93,7 +93,31 @@ def singleton_override.build_line_feeder; end
     pool.acknowledge(first)
     pool.acknowledge(second)
     assert_equal 0, pool.statistics[:bytes]
-    assert_false pool.submit(:too_large, 'oversized', bytes: 11)
+    assert_true pool.submit(:too_large, 'oversized', bytes: 11)
+    @release << true
+    pool.acknowledge(result_from(pool))
+  end
+
+  test 'one oversized batch uses a separate reservation beside regular capacity' do
+    started = Queue.new
+    pool = create_pool(task_limit: 3, byte_limit: 10, num_threads: 2) do |_, payload|
+      started << payload
+      @release.pop
+      payload
+    end
+    assert_true pool.submit(:large, :oversized, bytes: 11)
+    assert_equal :oversized, Timeout.timeout(5) { started.pop }
+    assert_true pool.submit(:small, :regular, bytes: 10)
+    assert_equal :regular, Timeout.timeout(5) { started.pop }
+    assert_false pool.submit(:another_large, :oversized, bytes: 12)
+    assert_false pool.submit(:another_small, :regular, bytes: 1)
+    assert_equal({ tasks: 2, bytes: 21, state: :running }, pool.statistics)
+
+    2.times { @release << true }
+    results = 2.times.map { result_from(pool) }
+    assert_equal %i[oversized regular], results.map(&:value).sort
+    results.each { |result| pool.acknowledge(result) }
+    assert_equal({ tasks: 0, bytes: 0, state: :running }, pool.statistics)
   end
 
   test 'a failed task produces an error result and the worker handles the next file' do
 
@@ -1031,6 +1031,52 @@ def test_worker_queue_saturation_drains_detached_waiting_file
       assert_empty driver.instance.instance_variable_get(:@worker_deferred_unwatch)
     end
 
+    def test_oversized_worker_batch_does_not_block_other_files
+      tail_input = Fluent::Plugin::TailInput
+      original_byte_limit = tail_input.const_get(:WORKER_QUEUE_BYTES)
+      tail_input.send(:remove_const, :WORKER_QUEUE_BYTES)
+      tail_input.const_set(:WORKER_QUEUE_BYTES, 6)
+      large_path = "#{@tmp_dir}/tail-large.txt"
+      short_path = "#{@tmp_dir}/tail-short.txt"
+      [large_path, short_path].each { |path| Fluent::FileWrapper.open(path, 'wb') {} }
+      config = config_element('ROOT', '', {
+        'path' => "#{@tmp_dir}/tail-*.txt",
+        'pos_file' => "#{@tmp_dir}/tail.pos",
+        'tag' => 't1',
+        'read_from_head' => true,
+        'format' => '/(?<message>.*)/',
+        'num_threads' => 2,
+        'read_lines_limit' => 1,
+        'enable_stat_watcher' => false,
+      })
+      driver = create_driver(config, false)
+      large_started = Queue.new
+      continue_large = Queue.new
+      released = false
+
+      driver.run(expect_records: 2, timeout: 10) do
+        pool = driver.instance.instance_variable_get(:@worker_pool)
+        wrap_worker_pool_process(pool) do |batch|
+          if batch.lines.any? { |line| line.include?('oversized-record') }
+            large_started << true
+            continue_large.pop
+          end
+        end
+        Fluent::FileWrapper.open(large_path, 'ab') { |file| file.puts('oversized-record') }
+        Fluent::FileWrapper.open(short_path, 'ab') { |file| file.puts('x') }
+        waiting(5) { sleep 0.01 until large_started.size == 1 }
+        waiting(5) { sleep 0.01 until driver.events.any? { |event| event[2]['message'] == 'x' } }
+        continue_large << true
+        released = true
+      end
+
+      assert_equal %w[oversized-record x], driver.events.map { |event| event[2]['message'] }.sort
+    ensure
+      continue_large << true if continue_large && !released
+      tail_input.send(:remove_const, :WORKER_QUEUE_BYTES) if tail_input.const_defined?(:WORKER_QUEUE_BYTES, false)
+      tail_input.const_set(:WORKER_QUEUE_BYTES, original_byte_limit) if original_byte_limit
+    end
+
     def test_worker_queue_saturation_drains_rotated_file
       paths = %w[tail-a.txt tail-b.txt tail-c.txt tail-d.txt tail-e.txt tail-f.txt].map { |name| "#{@tmp_dir}/#{name}" }
       paths.each { |path| Fluent::FileWrapper.open(path, 'wb') {} }