diff --git a/lib/fluent/event_router.rb b/lib/fluent/event_router.rb
index <HASH>..<HASH> 100644
--- a/lib/fluent/event_router.rb
+++ b/lib/fluent/event_router.rb
@@ -116,6 +116,8 @@ module Fluent
       if callback = find_callback
         callback.call(es)
       end
+    rescue Pipeline::OutputError => e
+      @emit_error_handler.handle_emits_error(tag, e.processed_es, e.internal_error)
     rescue => e
       @emit_error_handler.handle_emits_error(tag, es, e)
     end
@@ -161,6 +163,17 @@ module Fluent
     private
 
     class Pipeline
+
+      class OutputError < StandardError
+        attr_reader :internal_error
+        attr_reader :processed_es
+
+        def initialize(internal_error, processed_es)
+          @internal_error = internal_error
+          @processed_es = processed_es
+        end
+      end
+
       def initialize
         @filters = []
         @output = nil
@@ -178,7 +191,12 @@ module Fluent
 
       def emit_events(tag, es)
         processed = @optimizer.filter_stream(tag, es)
-        @output.emit_events(tag, processed)
+
+        begin
+          @output.emit_events(tag, processed)
+        rescue => e
+          raise OutputError.new(e, processed)
+        end
       end
 
       class FilterOptimizer
Apply modifications in pipeline to the records being passed to `@Error` label

Fix #<I>

When an error occurs in the output process in a pipeline,
the modifications made by the filter process are applied to the records
that is passsed to `@Error` label.