Emit buffered logs normally on timeout instead of throwing error

Previously, when a timeout occurred while waiting for the next multiline log, the plugin raised a `TimeoutError` via `emit_error_event`.
This caused the remaining buffered logs—most notably the last line of log outputs—to be dropped or incorrectly routed.

While the plugin provides a `timeout_label` option to route logs on timeout, users often use it without realizing they must explicitly configure a
`<label>` block to catch and handle the rerouted logs.
Without this awareness, it simply results in silent data loss.

This commit changes the behavior to log an info message and emit the buffered record normally using `router.emit` when no `timeout_label` is set,
ensuring no logs are lost by default.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

lib/fluent/plugin/filter_concat.rb

@@ -460,19 +460,19 @@ def flush_remaining_buffer
           @key => lines.join(@separator)
         }
         tag, time, record = elements.first
-        message = "Flush remaining buffer: #{stream_identity}"
-        handle_timeout_error(tag, time, record.merge(new_record), message)
-        log.info(message)
+        handle_timeout_error(tag, time, record.merge(new_record))
+        log.info("Flush remaining buffer: #{stream_identity}")
       end
       @buffer.clear
     end
 
-    def handle_timeout_error(tag, time, record, message)
+    def handle_timeout_error(tag, time, record)
       if @timeout_label
         event_router = event_emitter_router(@timeout_label)
         event_router.emit(tag, time, record)
       else
-        router.emit_error_event(tag, time, record, TimeoutError.new(message))
+        log.info "Timeout occurred while waiting for the next multiline log. Emitting the buffered log normally. tag: #{tag}"
+        router.emit(tag, time, record)
       end
     end
 

test/plugin/test_filter_concat.rb

@@ -224,7 +224,7 @@ def filter_with_time(conf, messages, wait: nil)
       ]
       filtered = filter(CONFIG + "flush_interval 2s", messages, wait: 3) do |d|
         errored = { "container_id" => "1", "message" => "message 1\nmessage 2" }
-        mock(d.instance.router).emit_error_event("test", anything, errored, anything)
+        mock(d.instance.router).emit("test", anything, errored)
       end
       assert_equal([], filtered)
     end
@@ -307,8 +307,8 @@ def filter_with_time(conf, messages, wait: nil)
         errored1 = { "container_id" => "1", "message" => "start" }
         errored2 = { "container_id" => "2", "message" => "start" }
         router = d.instance.router
-        mock(router).emit_error_event("test", anything, errored1, anything)
-        mock(router).emit_error_event("test", anything, errored2, anything)
+        mock(router).emit("test", anything, errored1)
+        mock(router).emit("test", anything, errored2)
       end
       assert_equal(expected, filtered)
     end
@@ -456,7 +456,7 @@ def filter_with_time(conf, messages, wait: nil)
       ]
       filtered = filter(config, messages, wait: 3) do |d|
         errored = { "container_id" => "1", "message" => "start\n  message 1\n  message 2" }
-        mock(d.instance.router).emit_error_event("test", anything, errored, anything)
+        mock(d.instance.router).emit("test", anything, errored)
       end
       assert_equal([], filtered)
     end
@@ -976,7 +976,7 @@ def filter_with_time(conf, messages, wait: nil)
       ]
       filtered = filter_with_time(config, messages, wait: 3) do |d|
         errored = { "container_id" => "1", "message" => "start\n  message 3\n  message 4" }
-        mock(d.instance.router).emit_error_event("test", @time, errored, anything)
+        mock(d.instance.router).emit("test", @time, errored)
       end
       expected = [
         [@time, { "container_id" => "1", "message" => "start\n  message 1\n  message 2" }]
@@ -1000,7 +1000,7 @@ def filter_with_time(conf, messages, wait: nil)
       filtered = filter_with_time(config, messages, wait: 3) do |d|
         mock(d.instance).flush_timeout_buffer.at_most(0)
         errored = { "container_id" => "1", "message" => "start" }
-        mock(d.instance.router).emit_error_event("test", @time, errored, anything)
+        mock(d.instance.router).emit("test", @time, errored)
       end
       expected = [
         [@time, { "container_id" => "1", "message" => "start\n  message 1\n  message 2" }]
@@ -1060,6 +1060,7 @@ def filter_with_time(conf, messages, wait: nil)
         "[error]: failed to flush timeout buffer error_class=StandardError error=\"timeout\"",
         "[error]: failed to flush timeout buffer error_class=StandardError error=\"timeout\"",
         "[error]: failed to flush timeout buffer error_class=StandardError error=\"timeout\"",
+        "[info]: Timeout occurred while waiting for the next multiline log. Emitting the buffered log normally. tag: test",
         "[info]: Flush remaining buffer: test:default"
       ]
       log_messages = logs.map do |line|