Fix pipeline shutdown ordering

Fixes #10872

logstash-core/lib/logstash/plugins/builtin/pipeline/input.rb

@@ -48,9 +48,10 @@ module ::LogStash; module Plugins; module Builtin; module Pipeline; class Input
   end
 
   def stop
-    # We stop receiving events before we unlisten to prevent races
-    @running.set(false) if @running # If register wasn't yet called, no @running!
     pipeline_bus.unlisten(self, address)
+    # We stop receiving events _after_ we unlisten to pick up any events sent by upstream outputs that
+    # have not yet stopped
+    @running.set(false) if @running # If register wasn't yet called, no @running!
   end
 
   def isRunning

logstash-core/spec/logstash/plugins/builtin/pipeline_input_output_spec.rb

@@ -101,6 +101,27 @@ describe ::LogStash::Plugins::Builtin::Pipeline do
         output.do_close
       end
     end
+
+    it "stopped input should process events until upstream outputs stop" do
+      start_input
+      output.register
+      pipeline_bus.setBlockOnUnlisten(true)
+
+      output.multi_receive([event])
+      expect(queue.pop(true).to_hash_with_metadata).to match(event.to_hash_with_metadata)
+
+      @thread = Thread.new do
+        begin
+          input.do_stop
+        end
+      end
+
+      sleep 1
+      output.multi_receive([event])
+      expect(queue.pop(true).to_hash_with_metadata).to match(event.to_hash_with_metadata)
+
+      output.do_close
+    end
   end
 
   describe "one output to multiple inputs" do

logstash-core/src/main/java/org/logstash/plugins/pipeline/PipelineBus.java

@@ -10,10 +10,12 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Stream;
 
 /**
- * This class is essentially the communication bus / central state for the `pipeline` inputs/outputs to talk to each
+ * This class is the communication bus for the `pipeline` inputs and outputs to talk to each other.
- * other. This class is threadsafe.
+ *
+ * This class is threadsafe.
  */
 public class PipelineBus {
+
     final ConcurrentHashMap<String, AddressState> addressStates = new ConcurrentHashMap<>();
     final ConcurrentHashMap<PipelineOutput, ConcurrentHashMap<String, AddressState>> outputsToAddressStates = new ConcurrentHashMap<>();
     volatile boolean blockOnUnlisten = false;
@@ -159,9 +161,10 @@ public class PipelineBus {
     }
 
     /**
-     * Stop listening on the given address with the given listener
+     * Stop listening on the given address with the given listener. Blocks until upstream outputs have
+     * stopped.
      *
-     * @param input   Input that should stop listening
+     * @param input Input that should stop listening
      * @param address Address on which to stop listening
      * @throws InterruptedException if interrupted while attempting to stop listening
      */