Adaptive Backpressure Based on Resource Utilization

[igfasouza]

As a potential future enhancement, it may be worth exploring adaptive backpressure mechanisms driven by runtime resource utilization.

Currently, backpressure is typically configured using fixed thresholds. An alternative approach could be to dynamically adjust consumption rates according to the application's operational state, taking into account metrics such as:

• Heap memory utilization
• CPU usage
• Number of in-flight messages

With this model, consumers could automatically reduce throughput during periods of high memory pressure or elevated CPU utilization, and gradually increase throughput again as resources become available.

While this approach introduces additional complexity and may not align with Kpipe's current design goals, it could provide a more resilient and self-regulating system under varying workloads.

This is intended as an exploratory idea for future discussion, and it would be interesting to understand whether similar concepts have already been evaluated or considered within the project's roadmap.

It could also be an interesting experiment on resource-constrained hardware such as a Raspberry Pi, where CPU and memory limitations are more easily observable. Such an environment might provide a practical way to evaluate whether adaptive backpressure can improve stability and throughput under changing workloads.

[eschizoid]

Hey @igfasouza,

Adaptive backpressure isn't a no. It's just that I've seen the failure modes up close and they're worse than the static-watermark behavior we have today.

Heap-based throttling is the obvious one to want, and the obvious one to get wrong. MemoryMXBean readings spike around GC, so if you throttle on the spike you're cutting allocation pressure exactly when the collector wanted some headroom, sometimes that makes the pause worse, and on G1 or ZGC it's mostly redundant because they're already self-regulating allocation.

CPU is worse. OperatingSystemMXBean.getProcessCpuLoad() is whole-container/cgroup on Linux, so on shared infra a noisy neighbor burning a core will throttle your consumer for reasons that have nothing to do with the consumer. I've watched this footgun fire on EC2.

The good news: the proposal already fits. BackpressureController is a record over a pluggable Strategy, and the existing lagStrategy() / inFlightStrategy(...) are just static factories on it. An adaptiveStrategy(...) is a third sibling, no changes to the controller, the consumer, or the dispatcher:

KPipe.json(topic, props)
    .withBackpressure(new BackpressureController(HIGH, LOW, adaptiveStrategy(...)))
    .toCustom(...)
    .start();

Different sites are going to want different signals anyway (G1 occupancy, RSS, cgroup memory.current, queue-depth derivatives, EWMA of in-flight latency, whatever). A pluggable strategy lets them pick instead of me picking for everyone.

The Pi angle from #146 is actually where this argument flips. Small heap, one dominant process, no noisy neighbors — the noise problems I just listed mostly go away. If someone prototypes adaptiveStrategy(...) against the existing seam on a Pi and shares numbers (pause/resume churn, GC headroom under load, throughput vs the static-watermark baseline), that's the evidence that'd move it from "third-party strategy" into something I'd want in core.

Happy to leave this open for that. If anyone wants to start, copy the shape of lagStrategy() and inFlightStrategy(...) — the Strategy interface is tiny.