He-Pin opened a new issue, #3215:
URL: https://github.com/apache/pekko/issues/3215
### Motivation
`MergeHub` 的中央队列使用无界的 `AbstractNodeQueue`，在动态 producer/consumer
场景下，如果下游停滞，中央队列可以无限增长，导致 OutOfMemoryError。
### 当前代码行为
`Hub.scala` 中 MergeHub 的中央队列：
```scala
// Hub.scala:182-185
queue = new AbstractNodeQueue[Event] {}
```
生产者仅在 `perProducerBufferSize`（默认 16）填满后才被 demand 节流（通过
`InputState.onElement`/`signalDemand`，160-173
行）。但中央队列本身没有上限——下游停滞时，生产者仍然可以继续向中央队列推送，导致队列无限增长。
### 预期行为
中央队列应有上界限制，或在队列增长到阈值时对生产者施加背压，防止 OOM。
### 代码证据
- `stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala:182-185`
— 无界 `AbstractNodeQueue`
- `stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala:160-173`
— 生产者 demand 节流逻辑
### 复现方式
创建 MergeHub，添加多个快速 producer 和一个慢速 consumer（例如 delay 阶段），观察内存增长。
### 影响范围
- 模块：`pekko-stream`
- 影响使用 MergeHub 的动态 fan-in 场景
- 生产环境中慢速下游可导致 OOM crash
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
