Backpressure not propagated when output pipelines used

**Logstash information**:
docker: `docker.elastic.co/logstash/logstash:8.12.1`

**JVM** (e.g. `java -version`):
bundled

**OS version** (`uname -a` if on a Unix-like system):
Ubuntu 24.04

**Description of the problem including expected versus actual behaviour**:

I have found what seems to be an issue with back-pressure propagation in lumberjack/beats plugins.

I have written and attached [a simple poc](https://github.com/elastic/logstash/files/14204285/output-pipelines-poc.tar.gz) with docker composed of a data generator and two data receivers. The generator creates an event every 5 seconds and sends it to one of the receivers via the lumberjack output (with both receivers configured in `hosts:[]`). Each receiver will write any event it receives to disk and print it on stdout.

When the receivers have a single output stanza with _multiple outputs_ configured, as soon as one output is unable to process a message successfully, the entire pipeline blocks, which applies back-pressure to the input. This back-pressure propagates across the network to the generator (via a simple disconnect, I expect), which chooses a new destination from `hosts: []`. No data is lost and fail over is instantaneous. This is a GoodThing™.

However, when the receivers are configured with the same outputs as _pipelines_ rather than inline in the output stanza, this doesn't happen. The pipeline stalls but doesn't disconnect upstream, so failover never happens. This is a BadThing™.

The relevant difference can be boiled down to this:

```
.
.
.
output {
stdout { }
file {
id => "raw"
path => "/tmp/raw/%{+YYYY.MM.dd-HH}.json.gz"
flush_interval => 10
codec => json_lines
gzip => true
}
}
```
vs this:
```
.
.
.
output {
pipeline { id => p_stdout send_to => "p_stdout" }
pipeline { id => p_raw send_to => "p_raw" }
}
```

**Steps to reproduce**:

I have attached [a docker-compose stack](https://github.com/elastic/logstash/files/14204285/output-pipelines-poc.tar.gz) that demonstrates the issue:
```
$ tar -zxf output-pipelines-poc.tar.gz
$ cd output-pipelines-poc
$ docker compose up
```
Edit docker-compose.yml to switch between the two scenarios.

When it starts, you will see the generator start sending data to one receiver. You can then break the filesystem output on that receiver by running `docker exec output-pipelines-poc-recv1-1 chmod 444 /tmp/raw/`. What happens next depends on how the output is configured:

Both outputs directly in the `output` stanza:
```
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T04:10:07.007946636Z,
output-pipelines-poc-recv1-1 | "sequence" => 3
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T04:10:12.012313741Z,
output-pipelines-poc-recv1-1 | "sequence" => 4
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T04:10:17.013979664Z,
output-pipelines-poc-recv1-1 | "sequence" => 5
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T04:10:22.014107894Z,
output-pipelines-poc-recv1-1 | "sequence" => 6
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {"level":"INFO","loggerName":"logstash.outputs.file","timeMillis":1707365422036,"thread":"[main]>worker0","logEvent":{"message":"Opening file","path":"/tmp/raw/2024.02/2024.02.08-04.json.gz"}}
output-pipelines-poc-recv1-1 | {"level":"ERROR","loggerName":"logstash.javapipeline","timeMillis":1707365422041,"thread":"[main]>worker0","logEvent":{"message":"Pipeline worker error, the pipeline will be stopped","pipeline_id":"main","error":"(EACCES) Permission denied - /tmp/raw/2024.02/2024.02.08-04.json.gz","exception":"Java::OrgJrubyExceptions::SystemCallError","backtrace":["org.jruby.RubyIO.sysopen(org/jruby/RubyIO.java:1260)","org.jruby.RubyFile.initialize(org/jruby/RubyFile.java:364)","org.jruby.RubyClass.new(org/jruby/RubyClass.java:931)","org.jruby.RubyIO.new(org/jruby/RubyIO.java:869)","RUBY.open(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-file-4.3.0/lib/logstash/outputs/file.rb:276)","RUBY.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-file-4.3.0/lib/logstash/outputs/file.rb:119)","org.jruby.RubyHash.each(org/jruby/RubyHash.java:1601)","RUBY.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-file-4.3.0/lib/logstash/outputs/file.rb:118)","org.jruby.ext.thread.Mutex.synchronize(org/jruby/ext/thread/Mutex.java:171)","RUBY.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-file-4.3.0/lib/logstash/outputs/file.rb:117)","RUBY.multi_receive(/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb:102)","org.logstash.config.ir.compiler.AbstractOutputDelegatorExt.multi_receive(org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java:121)","RUBY.start_workers(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:304)"],"thread":"#"}}
output-pipelines-poc-generator-1 | {"level":"ERROR","loggerName":"logstash.outputs.lumberjack","timeMillis":1707365427023,"thread":"[main]>worker0","logEvent":{"message":"Client write error, trying connect","e":{"stackTrace":[{"class":"org.logstash.log.LoggerExt","method":"error","file":"org/logstash/log/LoggerExt.java","line":127},{"class":"RUBY","method":"flush","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-lumberjack-3.1.9/lib/logstash/outputs/lumberjack.rb","line":66},{"class":"RUBY","method":"buffer_flush","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/stud-0.0.23/lib/stud/buffer.rb","line":219},{"class":"org.jruby.RubyHash","method":"each","file":"org/jruby/RubyHash.java","line":1601},{"class":"RUBY","method":"buffer_flush","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/stud-0.0.23/lib/stud/buffer.rb","line":216},{"class":"RUBY","method":"buffer_receive","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/stud-0.0.23/lib/stud/buffer.rb","line":159},{"class":"RUBY","method":"register","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-lumberjack-3.1.9/lib/logstash/outputs/lumberjack.rb","line":52},{"class":"RUBY","method":"encode","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-codec-json-3.1.1/lib/logstash/codecs/json.rb","line":69},{"class":"RUBY","method":"encode","file":"/usr/share/logstash/logstash-core/lib/logstash/codecs/delegator.rb","line":48},{"class":"org.logstash.instrument.metrics.AbstractSimpleMetricExt","method":"time","file":"org/logstash/instrument/metrics/AbstractSimpleMetricExt.java","line":74},{"class":"org.logstash.instrument.metrics.AbstractNamespacedMetricExt","method":"time","file":"org/logstash/instrument/metrics/AbstractNamespacedMetricExt.java","line":68},{"class":"RUBY","method":"encode","file":"/usr/share/logstash/logstash-core/lib/logstash/codecs/delegator.rb","line":47},{"class":"RUBY","method":"receive","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-lumberjack-3.1.9/lib/logstash/outputs/lumberjack.rb","line":58},{"class":"RUBY","method":"multi_receive","file":"/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb","line":104},{"class":"org.jruby.RubyArray","method":"each","file":"org/jruby/RubyArray.java","line":1989},{"class":"RUBY","method":"multi_receive","file":"/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb","line":104},{"class":"org.logstash.config.ir.compiler.AbstractOutputDelegatorExt","method":"multi_receive","file":"org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java","line":121},{"class":"RUBY","method":"start_workers","file":"/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb","line":304}],"exception":"Connection reset by peer","message":"(IOError) Connection reset by peer","localizedMessage":"(IOError) Connection reset by peer"},"backtrace":["org/jruby/ext/openssl/SSLSocket.java:965:in `syswrite'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/jls-lumberjack-0.0.26/lib/lumberjack/client.rb:107:in `send_window_size'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/jls-lumberjack-0.0.26/lib/lumberjack/client.rb:127:in `write_sync'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/jls-lumberjack-0.0.26/lib/lumberjack/client.rb:42:in `write'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-lumberjack-3.1.9/lib/logstash/outputs/lumberjack.rb:64:in `flush'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/stud-0.0.23/lib/stud/buffer.rb:219:in `block in buffer_flush'","org/jruby/RubyHash.java:1601:in `each'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/stud-0.0.23/lib/stud/buffer.rb:216:in `buffer_flush'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/stud-0.0.23/lib/stud/buffer.rb:159:in `buffer_receive'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-lumberjack-3.1.9/lib/logstash/outputs/lumberjack.rb:52:in `block in register'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-codec-json-3.1.1/lib/logstash/codecs/json.rb:69:in `encode'","/usr/share/logstash/logstash-core/lib/logstash/codecs/delegator.rb:48:in `block in encode'","org/logstash/instrument/metrics/AbstractSimpleMetricExt.java:74:in `time'","org/logstash/instrument/metrics/AbstractNamespacedMetricExt.java:68:in `time'","/usr/share/logstash/logstash-core/lib/logstash/codecs/delegator.rb:47:in `encode'","/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-lumberjack-3.1.9/lib/logstash/outputs/lumberjack.rb:58:in `receive'","/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb:104:in `block in multi_receive'","org/jruby/RubyArray.java:1989:in `each'","/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb:104:in `multi_receive'","org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java:121:in `multi_receive'","/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:304:in `block in start_workers'"]}}
output-pipelines-poc-recv2-1 | {"level":"INFO","loggerName":"logstash.codecs.json","timeMillis":1707365427590,"thread":"defaultEventExecutorGroup-4-1","logEvent":{"message":"ECS compatibility is enabled but `target` option was not specified. This may cause fields to be set at the top-level of the event where they are likely to clash with the Elastic Common Schema. It is recommended to set the `target` option to avoid potential schema conflicts (if your data is ECS compliant or non-conflicting, feel free to ignore this message)"}}
output-pipelines-poc-recv2-1 | {
output-pipelines-poc-recv2-1 | "@timestamp" => 2024-02-08T04:10:27.017584396Z,
output-pipelines-poc-recv2-1 | "sequence" => 7
output-pipelines-poc-recv2-1 | }
output-pipelines-poc-recv2-1 | {"level":"INFO","loggerName":"logstash.outputs.file","timeMillis":1707365427797,"thread":"[main]>worker0","logEvent":{"message":"Opening file","path":"/tmp/raw/2024.02/2024.02.08-04.json.gz"}}
output-pipelines-poc-recv2-1 | {"level":"INFO","loggerName":"logstash.outputs.file","timeMillis":1707365427800,"thread":"[main]>worker0","logEvent":{"message":"Creating directory","directory":"/tmp/raw/2024.02"}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"logstash.javapipeline","timeMillis":1707365428517,"thread":"[main]-pipeline-manager","logEvent":{"message":"Waiting for input plugin to close","pipeline_id":"main","thread":"#"}}
output-pipelines-poc-recv2-1 | {
output-pipelines-poc-recv2-1 | "@timestamp" => 2024-02-08T04:10:32.018297001Z,
output-pipelines-poc-recv2-1 | "sequence" => 8
output-pipelines-poc-recv2-1 | }
output-pipelines-poc-recv1-1 | {"level":"INFO","loggerName":"logstash.javapipeline","timeMillis":1707365436857,"thread":"[main]-pipeline-manager","logEvent":{"message":"Pipeline terminated","pipeline.id":"main"}}
output-pipelines-poc-recv1-1 | {"level":"INFO","loggerName":"logstash.pipelinesregistry","timeMillis":1707365436978,"thread":"Converge PipelineAction::Delete","logEvent":{"message":"Removed pipeline from registry successfully","pipeline_id":"main"}}
output-pipelines-poc-recv1-1 | {"level":"INFO","loggerName":"logstash.runner","timeMillis":1707365436989,"thread":"LogStash::Runner","logEvent":{"message":"Logstash shut down."}}
output-pipelines-poc-recv2-1 | {
output-pipelines-poc-recv2-1 | "@timestamp" => 2024-02-08T04:10:37.018142972Z,
output-pipelines-poc-recv2-1 | "sequence" => 9
output-pipelines-poc-recv2-1 | }
output-pipelines-poc-recv1-1 exited with code 0
output-pipelines-poc-recv2-1 | {
output-pipelines-poc-recv2-1 | "@timestamp" => 2024-02-08T04:10:42.020047169Z,
output-pipelines-poc-recv2-1 | "sequence" => 10
output-pipelines-poc-recv2-1 | }
output-pipelines-poc-recv2-1 | {
output-pipelines-poc-recv2-1 | "@timestamp" => 2024-02-08T04:10:47.020893223Z,
output-pipelines-poc-recv2-1 | "sequence" => 11
output-pipelines-poc-recv2-1 | }
```

Both outputs via output pipelines:
```
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "sequence" => 6,
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T03:53:08.093500216Z
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "sequence" => 7,
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T03:53:13.094506620Z
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "sequence" => 8,
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T03:53:18.094557528Z
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "sequence" => 9,
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T03:53:23.094548073Z
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {"level":"INFO","loggerName":"logstash.outputs.file","timeMillis":1707364408150,"thread":"[raw]>worker0","logEvent":{"message":"Opening file","path":"/tmp/raw/2024.02/2024.02.08-03.json.gz"}}
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "sequence" => 10,
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T03:53:28.094767419Z
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {"level":"ERROR","loggerName":"logstash.javapipeline","timeMillis":1707364408153,"thread":"[raw]>worker0","logEvent":{"message":"Pipeline worker error, the pipeline will be stopped","pipeline_id":"raw","error":"(EACCES) Permission denied - /tmp/raw/2024.02/2024.02.08-03.json.gz","exception":"Java::OrgJrubyExceptions::SystemCallError","backtrace":["org.jruby.RubyIO.sysopen(org/jruby/RubyIO.java:1260)","org.jruby.RubyFile.initialize(org/jruby/RubyFile.java:364)","org.jruby.RubyClass.new(org/jruby/RubyClass.java:931)","org.jruby.RubyIO.new(org/jruby/RubyIO.java:869)","RUBY.open(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-file-4.3.0/lib/logstash/outputs/file.rb:276)","RUBY.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-file-4.3.0/lib/logstash/outputs/file.rb:119)","org.jruby.RubyHash.each(org/jruby/RubyHash.java:1601)","RUBY.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-file-4.3.0/lib/logstash/outputs/file.rb:118)","org.jruby.ext.thread.Mutex.synchronize(org/jruby/ext/thread/Mutex.java:171)","RUBY.multi_receive_encoded(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-output-file-4.3.0/lib/logstash/outputs/file.rb:117)","RUBY.multi_receive(/usr/share/logstash/logstash-core/lib/logstash/outputs/base.rb:102)","org.logstash.config.ir.compiler.AbstractOutputDelegatorExt.multi_receive(org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java:121)","RUBY.start_workers(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:304)"],"thread":"#"}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"logstash.javapipeline","timeMillis":1707364408157,"thread":"[raw]-pipeline-manager","logEvent":{"message":"Waiting for input plugin to close","pipeline_id":"raw","thread":"#"}}
output-pipelines-poc-recv1-1 | {"level":"INFO","loggerName":"logstash.javapipeline","timeMillis":1707364410440,"thread":"[raw]-pipeline-manager","logEvent":{"message":"Pipeline terminated","pipeline.id":"raw"}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364413153,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
output-pipelines-poc-recv1-1 | {
output-pipelines-poc-recv1-1 | "sequence" => 11,
output-pipelines-poc-recv1-1 | "@timestamp" => 2024-02-08T03:53:33.101254311Z
output-pipelines-poc-recv1-1 | }
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364414155,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364415156,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364416157,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364417159,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364418160,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364419161,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364420167,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
output-pipelines-poc-recv1-1 | {"level":"WARN","loggerName":"org.logstash.plugins.pipeline.PipelineBus","timeMillis":1707364421169,"thread":"[recv]>worker0","logEvent":{"message":"Attempted to send event to 'p_raw' but that address was unavailable. Maybe the destination pipeline is down or stopping? Will Retry."}}
```