in_tail: fix NameError in flush_buffer with emit_unmatched_lines

**Which issue(s) this PR fixes**:
Fixes #

**What this PR does / why we need it**:
TailInput#flush_buffer referenced an undefined local variable
`tail_watcher` instead of its argument `tw`, so it raised
`NameError: undefined local variable or method 'tail_watcher'` when an
incomplete buffered line is flushed and the parser returns no record
while `emit_unmatched_lines` is enabled. This can happen with parser
plugins which don't implement the `unmatched_lines` feature, because
in_tail only emits the buffered line as `unmatched_line` in that case.

Add a regression test which flushes a buffered line with such a parser
and checks that `unmatched_line` and `path_key` are emitted.

**Docs Changes**:
None

**Release Note**:
in_tail: fix NameError raised when flushing buffered lines with
emit_unmatched_lines true

Co-Authored-By: Qwen3.8-Flash-Next
Signed-off-by: Takuro Ashie <ashie@clear-code.com>

Author: ashie

lib/fluent/plugin/in_tail.rb

@@ -706,7 +706,7 @@ def flush_buffer(tw, buf)
         else
           if @emit_unmatched_lines
             record = { 'unmatched_line' => buf }
-            record[@path_key] ||= tail_watcher.path unless @path_key.nil?
+            record[@path_key] ||= tw.path unless @path_key.nil?
             tag = if @tag_prefix || @tag_suffix
                     @tag_prefix + tw.tag + @tag_suffix
                   else

test/plugin/test_in_tail.rb

@@ -110,6 +110,15 @@ def common_follow_inode_config
                     })
     ])
 
+  # A parser which never matches, like parser plugins which don't implement
+  # the `unmatched_lines` feature. Such parsers yield (nil, nil) to the caller.
+  class NeverMatchingParser
+    def parse(text, &block)
+      block.call(nil, nil)
+      nil
+    end
+  end
+
   EX_ROTATE_WAIT = 0
   EX_FOLLOW_INODES = false
 
@@ -1310,6 +1319,32 @@ def test_multiline_with_emit_unmatched_lines_true(data)
       assert_equal({"message1" => "test8"}, events[4][2])
     end
 
+    data(flat: MULTILINE_CONFIG,
+         parse: PARSE_MULTILINE_CONFIG)
+    def test_multiline_flush_buffer_with_emit_unmatched_lines_and_path_key(data)
+      config = data + config_element("", "", { "emit_unmatched_lines" => true, "path_key" => "path" })
+      Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "wb") { |f| }
+
+      d = create_driver(config)
+      d.run(expect_emits: 1) do
+        plugin = d.instance
+        # `unmatched_lines` is a feature of each parser plugin, so a parser
+        # which doesn't implement it yields (nil, nil) even if
+        # `emit_unmatched_lines` is enabled. in_tail must emit the buffered
+        # line as `unmatched_line` in that case.
+        plugin.instance_variable_set(:@parser, NeverMatchingParser.new)
+        tw = Fluent::Plugin::TailInput::TailWatcher.new(
+          create_target_info("#{@tmp_dir}/tail.txt"), nil, $log, true, false, nil, nil, nil, nil
+        )
+        plugin.flush_buffer(tw, "incomplete line\n")
+      end
+
+      events = d.events
+      assert_equal(1, events.length)
+      assert_equal("t1", events[0][0])
+      assert_equal({ "unmatched_line" => "incomplete line", "path" => "#{@tmp_dir}/tail.txt" }, events[0][2])
+    end
+
     data(
       flat: MULTILINE_CONFIG_WITH_NEWLINE,
       parse: PARSE_MULTILINE_CONFIG_WITH_NEWLINE)