stream: deprecate Source.queue overloads that materialize SourceQueueWithComplete#3095
stream: deprecate Source.queue overloads that materialize SourceQueueWithComplete#3095He-Pin wants to merge 1 commit into
Conversation
He-Pin
force-pushed
the
deprecate-source-queue-overloads
branch
from
ebc5b14 to
44e3d67
Compare
|
doc build broken |
| @deprecated( | ||
| "Prefer Source.queue(bufferSize) which materializes a BoundedSourceQueue with synchronous feedback " + | ||
| "(dropping the newest element when the buffer is full). For backpressure, use Source.actorRefWithBackpressure " + | ||
| "or MergeHub.source instead. See the Pekko Streams documentation for the Source.queue migration guide.", |
There was a problem hiding this comment.
I'm not 100% sure but will go with the consensus.
Could we consider not deprecating and just being more persuasive in the docs?
I've already run into issues with the methods that we removed already in 2.0.0 and it being far from obvious what the replacement is. For me, if it's not broken, why remove it?
Having alternatives is great.
There was a problem hiding this comment.
it will cause OOM and FGC at $Work, the real bug, trust me ,I was using it and later migrate to the BoundedSourceQueue one by @jrudolph , the now it works very well, my system is the Taobao Live.
There was a problem hiding this comment.
Since these messages all end up in the mailbox, if the system is under heavy load, the backpressure mechanism fails; you then see heap memory steadily rising, followed by Full GCs, and eventually the system crashes. I think it’s better to just remove it—how should I put it? The feature works fine under low load, but it’s riddled with bugs under high load. akkadotnet/akka.net#8248 @Aaronontheweb FYI
There was a problem hiding this comment.
What I mean is, since this has indeed caused problems in my experience, we could deprecate it first and then remove it after a few years once hardly anyone is using it.
There was a problem hiding this comment.
there are other overflow stategies - should we consider doing something about OverflowStrategy .backpressure? We've already deprecated and removed OverflowStrategy.dropNew.
There was a problem hiding this comment.
It’s futile, because messages must first enter the mailbox before policies can actually be executed; if the actor never gets scheduled, none of those policies take effect. Consequently, an Out-of-Memory (OOM) error occurs as the mailbox grows indefinitely. In my view, this design is fundamentally flawed—it poses serious production safety risks under high load and, quite simply, is prone to causing production incidents.
There was a problem hiding this comment.
Yeah these days we have Source.Channel which uses a .NET System.Threading.Channel, which has bounded write / multi or single reader / writer semantics built in. I modified the queue here to accept a cancellation token mostly to bring it in line with .NET idioms for async programming.
There was a problem hiding this comment.
Wow, nice! Thanks for the comment. I maintain a Taobao livestreaming system; initially, I used RxJava, but I frequently ran into OOM (Out of Memory) and FGC (Full Garbage Collection) issues during broadcasts by top streamers. Later, I switched to Akka Streams' Source.queue, but I still encountered FGC problems. This had a huge impact on the streams of individual top broadcasters, and the issue wasn't resolved until I finally switched to BoundedSourceQueue. The system is running very smoothly now. thanks @jrudolph
…tion docs Motivation: Paradox @apidoc directives only resolve class/object names. Method references (Source.actorRefWithBackpressure, MergeHub.source$), arbitrary text (single imperative producer), and parenthetical text immediately following @apidoc[...] were all misparsed, causing the docs/paradox build to fail in CI. Modification: - Replace @apidoc[Source.actorRefWithBackpressure] with @ref links to the existing operator page. - Replace @apidoc[MergeHub.source$] with plain code (no operator page). - Replace @apidoc[SourceQueueWithComplete] and @apidoc[BoundedSourceQueue.fail] with plain code (class in context / method reference). - Restructure queue.md:9 to remove parentheses immediately after @apidoc[OverflowStrategy] that Paradox misread as anchor syntax. Result: sbt docs/paradox passes. CI paradoxMarkdownToHtml step succeeds. Tests: - sbt docs/paradox — pass References: Fixes CI failure in PR apache#3095
…WithComplete Motivation: The Source.queue overloads returning SourceQueueWithComplete have an async offer future that can hang indefinitely under backpressure, causing OOM/FGC in production. BoundedSourceQueue is the safer alternative. Modification: - Deprecate the four Source.queue overloads (scaladsl + javadsl) - Update docs with migration table and warnings - Fix @apidoc references in Paradox docs - Suppress deprecation warnings at internal call sites Result: Users are guided toward BoundedSourceQueue and Source.actorRef, reducing the risk of production OOM from hung offer futures. Tests: - sbt "stream / Test / compile" References: Refs apache#3095
He-Pin
force-pushed
the
deprecate-source-queue-overloads
branch
from
afa7323 to
2ceeea1
Compare
…WithComplete Motivation: The Source.queue overloads returning SourceQueueWithComplete have an async offer future that can hang indefinitely under backpressure, causing OOM/FGC in production. BoundedSourceQueue is the safer alternative. Modification: - Deprecate the four Source.queue overloads (scaladsl + javadsl) - Update docs with migration table and warnings - Fix @apidoc references in Paradox docs - Suppress deprecation warnings at internal call sites Result: Users are guided toward BoundedSourceQueue and Source.actorRef, reducing the risk of production OOM from hung offer futures. Tests: - sbt "stream / Test / compile" References: Refs apache#3095
He-Pin
force-pushed
the
deprecate-source-queue-overloads
branch
from
2ceeea1 to
aea36d1
Compare
3 participants
Motivation
The
Source.queue(bufferSize, overflowStrategy)overloads (both Scala and Java DSLs) materialize aSourceQueueWithCompletewhose asynchronousofferfuture can hang indefinitely underOverflowStrategy.backpressurewhen the buffer is full and downstream stalls. This is a long-standing trap that has caused real-world deadlocks (the same class of issue was documented upstream as akka/akka-core#29557, and more recently prompted akka/akka.net#8248 to add a cancellation-aware offer overload).Pekko already provides a safer replacement:
Source.queue[T](bufferSize)materializes aBoundedSourceQueuewith synchronous feedback (drops the newest element when the buffer is full, never hangs).Source.actorRefWithBackpressureandMergeHub.sourcecover the common use cases.The upstream Akka discussion in akka/akka-core#29801 ("Replace old Source.queue") has been open since 2021 without converging on a new backpressure primitive. We should not wait: the drop-new path (
BoundedSourceQueue) is already stable and production-proven, and the backpressure path is covered by existing operators.Modification
API deprecation:
@deprecated(since = "2.0.0")on the fourSource.queueoverloads that accept anOverflowStrategy(two scaladsl + two javadsl).Source.queue[T](bufferSize)overloads that returnBoundedSourceQueueare not deprecated.SourceQueue/SourceQueueWithCompletetraits are not deprecated (they are still referenced by the deprecated factory return types and by user code).Internal suppression:
@nowarn("msg=deprecated")on the deprecated factories (so their delegation chain does not retrigger the warning under-Xfatal-warnings/-Werror) and on the two Pekko-internal call sites that intentionally exercise the deprecated API:InvokeWithFeedbackBenchmarkandQueueSourceSpec.Docs: preserve the existing
Source/queue.mdprose (both theBoundedSourceQueueand theSourceQueuesections, both examples, all existing wording). Add:@@@ warningcallout immediately after the operator summary pointing to the recommended API and the migration table.Signature (SourceQueue)block.OverflowStrategymigration table appended at the bottom.actor-interop.md'sSource.queuesubsection explaining that the referenced snippet has been updated.Source/actorRefWithBackpressure.md("BoundedSourceQueue" instead of "SourceQueue").#source-queuesnippet inIntegrationDocSpec.scalawas fixed to use a directmatchonQueueOfferResult. The previous snippet usedqueue.offer(x).map { case ... }, butBoundedSourceQueue.offerreturnsQueueOfferResultsynchronously —QueueOfferResulthas no.mapmethod, so the snippet did not type-check against the API it was nominally demonstrating. The sibling#source-queue-synchronousScala snippet had a misleadingimplicit val ec = system.dispatcherdeclaration that served no purpose (the synchronousmatchdoes not need anExecutionContext); it was removed. Java#source-queuesnippet was already consistent and is unchanged.Result
Source.queue(...)with anOverflowStrategyget a compile-time deprecation warning pointing them toSource.queue(bufferSize)for drop-new, orSource.actorRefWithBackpressure/MergeHub.sourcefor backpressure.-Xfatal-warnings.#source-queuesnippets type-check againstBoundedSourceQueue.offerreturningQueueOfferResultsynchronously, match the Java counterpart semantically, and no longer declare unused imports.Tests
sbt "stream / compile"— pass (213 Scala + 5 Java sources)sbt "stream-tests / Test / compile"— pass (229 Scala + 32 Java sources)sbt "bench-jmh / compile"— pass (98 Scala + 1 Java sources)sbt "docs / Compile / managedResources"— pass (paradoxStreamOperatorsIndexGeneratorextracts the summary/category link successfully)sbt "docs / Test / compile"— pass (224 Scala + 234 Java sources; updated#source-queuesnippet type-checks againstBoundedSourceQueue.offerreturningQueueOfferResult; subsequent incremental recompile after the unused-ecremoval also passed)References