Use a simple timer thread for flushes instead of a task

logstash-core/lib/logstash/pipeline.rb

@@ -14,7 +14,6 @@
 require "logstash/shutdown_controller"
 require "logstash/util/wrapped_synchronous_queue"
 require "logstash/pipeline_reporter"
-require "concurrent/timer_task"
 require "logstash/output_delegator"
 
 module LogStash; class Pipeline
@@ -69,6 +68,7 @@ def initialize(config_str, settings = {})
     @settings = DEFAULT_SETTINGS.clone
     # @ready requires thread safety since it is typically polled from outside the pipeline thread
     @ready = Concurrent::AtomicBoolean.new(false)
+    @running = Concurrent::AtomicBoolean.new(false)
     @flushing = Concurrent::AtomicReference.new(false)
     settings.each {|setting, value| configure(setting, value) }
 
@@ -128,7 +128,11 @@ def run
 
     # Block until all inputs have stopped
     # Generally this happens if SIGINT is sent and `shutdown` is called from an external thread
+
+    @running.make_true
     wait_inputs
+    @running.make_false
+
     @logger.info("Input plugins stopped! Will shutdown filter/output workers.")
 
     shutdown_flusher
@@ -419,15 +423,16 @@ def update(time, result, exception)
   end
 
   def start_flusher
-    @flusher_task = Concurrent::TimerTask.new { flush }
-    @flusher_task.execution_interval = @settings[:flush_interval]
-    @flusher_task.timeout_interval = @settings[:flush_timeout_interval]
-    @flusher_task.add_observer(FlusherObserver.new(@logger, @flushing))
-    @flusher_task.execute
+    @flusher_thread = Thread.new do
+      while Stud.stoppable_sleep(5, 0.1) { @running.false? }
+        flush
+        break if @running.false?
+      end
+    end
   end
 
   def shutdown_flusher
-    @flusher_task.shutdown
+    @flusher_thread.join
   end
 
   def flush

logstash-core/spec/logstash/pipeline_spec.rb

@@ -345,7 +345,10 @@ class TestPipeline < LogStash::Pipeline
       Thread.new { pipeline.run }
       sleep 0.1 while !pipeline.ready?
       # give us a bit of time to flush the events
-      wait(5).for { output.events.first["message"].split("\n").count }.to eq(number_of_events)
+      wait(5).for do
+        next unless output && output.events && output.events.first
+        output.events.first["message"].split("\n").count
+      end.to eq(number_of_events)
       pipeline.shutdown
     end
   end