test_fluentd: fix flaky zero_downtime_restart from per-chunk log parsing (#5450)

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

**What this PR does / why we need it**:
The read loop in `run_fluentd` split each `eager_read` chunk on its own
(`logs = buf.split("\n")`) and yielded that to the caller for counting.
A single read can end in the middle of a line -- this happens when the
source-only buffer is replayed and a whole chunk of records is flushed
to `stdout` in one burst while the reader catches the pipe mid-write,
which is much more likely on a loaded CI runner.
When a record's line straddles two reads, neither half matches the
record regexp, so the caller drops it from its count while the full line
is still present in the accumulated output.

Carry the incomplete trailing line across reads and yield only complete
lines, the same way `assert_log_matches` already reads against the
accumulated buffer.

This PR will fix following error:
```
4) Failure: test: should restart with zero downtime (no data loss)(TestFluentdCommand::zero_downtime_restart)
/Users/runner/work/fluentd/fluentd/test/command/test_fluentd.rb:1574:in `block (2 levels) in <class:TestFluentdCommand>'
     1571:         end
     1572:       end
     1573:
  => 1574:       assert_equal(
     1575:         [(0..499).to_a, (0..499).to_a, (0..499).to_a],
     1576:         [
     1577:           records_by_type["udp"].sort,
<[[0,
  1,
  2,
  3,
  4,
  5,
...
```
Ref.
https://github.com/fluent/fluentd/actions/runs/29796336422/job/88528337017#step:6:5375

**Docs Changes**:
N/A

**Release Note**:
N/A

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: Watson1978

test/command/test_fluentd.rb

@@ -1413,6 +1413,13 @@ def run_fluentd(config)
       stdio_buf = ""
       execute_command(cmdline) do |pid, stdout|
         begin
+          # Keep an incomplete trailing line across reads. A single read can end
+          # in the middle of a line (e.g. when the buffered records are flushed
+          # in a burst and the reader catches the pipe mid-write), so splitting
+          # each read chunk independently would drop the straddling record from
+          # the caller's counting. Carry the partial line over and only yield
+          # complete lines, exactly once each.
+          line_buf = +""
           waiting(60) do
             while true
               readables, _, _ = IO.select([stdout], nil, nil, 1)
@@ -1421,11 +1428,16 @@ def run_fluentd(config)
 
               buf = eager_read(readables.first)
               stdio_buf << buf
-              logs = buf.split("\n")
+              line_buf << buf
+              # Complete lines to `logs`, trailing remainder to `line_buf`.
+              # The `-1` is required: without it split drops the trailing empty
+              # field, so a buffer ending in "\n" (the usual case) would carry
+              # over the last complete line and merge it with the next read.
+              *logs, line_buf = line_buf.split("\n", -1)
 
               yield logs
 
-              break if buf.include? "finish test"
+              break if logs.any? { |log| log.include?("finish test") }
             end
           end
         ensure