diff --git a/logstash-core/lib/logstash/pipeline.rb b/logstash-core/lib/logstash/pipeline.rb
index e23b40742..1d0ec7f25 100644
--- a/logstash-core/lib/logstash/pipeline.rb
+++ b/logstash-core/lib/logstash/pipeline.rb
@@ -477,7 +477,10 @@ module LogStash; class Pipeline < BasePipeline
         filter_batch(batch)
       end
       flush_filters_to_batch(batch, :final => false) if signal.flush?
-      output_batch(batch) if batch.size > 0
+      if batch.size > 0
+        output_batch(batch)
+        @filter_queue_client.close_batch(batch)
+      end
       # keep break at end of loop, after the read_batch operation, some pipeline specs rely on this "final read_batch" before shutdown.
       break if (shutdown_requested && !draining_queue?)
     end