feat(stream): replace fanout publisher runtime with GraphStage bridge by He-Pin · Pull Request #2874 · apache/pekko

He-Pin · 2026-04-18T23:12:16Z

Summary

This is the second #2860 slice.

It replaces the actor-backed runtime behind Sink.asPublisher(fanout = true) with FanoutPublisherBridgeStage, removes the legacy FanoutProcessor.scala, replaces the old implementation-bound spec with FanoutPublisherBehaviorSpec, and adds the matching 2.0.x MiMa exclusions.

The runtime stays on a dedicated internal fanout bridge and preserves the existing terminal-signal contract already exercised by FlowSpec.

Motivation

remove the remaining legacy actor-backed fanout runtime from the fanout publisher path
keep Sink.asPublisher(fanout = true) behavior-compatible while moving the runtime to GraphStage-based stream infrastructure
replace implementation-bound actor tests with black-box behavior coverage and TCK-backed validation

Modification

rewire FanoutPublisherSink.create in stream/src/main/scala/org/apache/pekko/stream/impl/Sinks.scala
add stream/src/main/scala/org/apache/pekko/stream/impl/FanoutPublisherBridgeStage.scala
delete stream/src/main/scala/org/apache/pekko/stream/impl/FanoutProcessor.scala
replace stream-tests/src/test/scala/org/apache/pekko/stream/impl/FanoutProcessorSpec.scala with stream-tests/src/test/scala/org/apache/pekko/stream/impl/FanoutPublisherBehaviorSpec.scala
add stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-fanout-processor.excludes
add a follow-up scalafmt-only commit on FanoutPublisherBridgeStage.scala to satisfy the GitHub Code is formatted check

Result

the fanout publisher path no longer depends on FanoutProcessorImpl
timeout, late-subscriber, downstream-failure, and multi-subscriber delivery behavior remain covered
the slice is independently reviewable and leaves the TLS migration as the next focused step
the PR branch now includes the CI formatting-only follow-up needed by the GitHub scalafmt diff-ref workflow

Validation

sbt scalafmtAll
sbt "stream/test:compile"
sbt "stream/mimaReportBinaryIssues"
sbt "stream-tests/testOnly org.apache.pekko.stream.impl.FanoutPublisherBehaviorSpec org.apache.pekko.stream.impl.TimeoutsSpec org.apache.pekko.stream.scaladsl.FlowSpec org.apache.pekko.stream.scaladsl.SinkSpec"
sbt "stream-tests/testOnly org.apache.pekko.stream.javadsl.SinkTest"
sbt "stream-tests-tck/testOnly org.apache.pekko.stream.tck.FanoutPublisherTest"
sbt "scalafmtOnly stream/src/main/scala/org/apache/pekko/stream/impl/FanoutPublisherBridgeStage.scala"

Upstream / references

Issue: Rewritten more to graphstage #2860
Related prior exploration: POC: chore: use Hub #1705 and Remove FanoutProcessorImpl and friends akka/akka-core#21261

Motivation: Sink.asPublisher(fanout = true) still depended on the legacy actor-backed FanoutProcessorImpl runtime, which kept issue apache#2860 blocked on old processor infrastructure and implementation-bound tests. Modification: Route FanoutPublisherSink through a new FanoutPublisherBridgeStage, delete the legacy FanoutProcessor implementation, replace the old actor-bound spec with FanoutPublisherBehaviorSpec, and add the matching 2.0.x MiMa excludes for the removed binary-visible classes. Result: The fanout publisher path now runs on a dedicated GraphStage bridge with the existing terminal-signal contract preserved, broader behavior coverage added, and compile/MiMa/TCK validation passing. References: apache#2860 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Motivation: GitHub's scalafmt diff-ref job flagged FanoutPublisherBridgeStage.scala on PR apache#2874. Modification: Applied scalafmt formatting to FanoutPublisherBridgeStage.scala without changing behavior. Result: The fanout bridge file now matches the repository's CI formatting expectations. References: apache#2860 apache#2874 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot

Pull request overview

This PR migrates the runtime behind Sink.asPublisher(fanout = true) from the legacy actor-based fanout processor to a GraphStage-based bridge (FanoutPublisherBridgeStage), while keeping the existing fanout publisher behavior contract intact and updating tests and MiMa filters accordingly.

Changes:

Rewires FanoutPublisherSink materialization to use a GraphStage bridge instead of FanoutProcessorImpl/ActorPublisher.
Introduces FanoutPublisherBridgeStage (GraphStage + RS Publisher) implementing the fanout subscription/buffering/timeout behavior.
Replaces the implementation-bound actor spec with a black-box behavior spec and adds MiMa exclusions for removed internal classes.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
stream/src/main/scala/org/apache/pekko/stream/impl/Sinks.scala	Switches `FanoutPublisherSink.create` to materialize a `Source.asSubscriber` → `FanoutPublisherBridgeStage` bridge.
stream/src/main/scala/org/apache/pekko/stream/impl/FanoutPublisherBridgeStage.scala	Adds the new GraphStage-based fanout publisher bridge implementation.
stream/src/main/scala/org/apache/pekko/stream/impl/FanoutProcessor.scala	Removes the legacy actor-backed fanout processor implementation.
stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-fanout-processor.excludes	Adds MiMa filters for the removed internal fanout processor classes.
stream-tests/src/test/scala/org/apache/pekko/stream/impl/FanoutPublisherBehaviorSpec.scala	Adds black-box behavior tests for `Sink.asPublisher(fanout = true)` (including timeout + multi-subscriber behavior).
stream-tests/src/test/scala/org/apache/pekko/stream/impl/FanoutProcessorSpec.scala	Removes the actor-implementation-specific spec.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

pjfanning

lgtm

He-Pin · 2026-04-20T03:05:35Z

+  override def subscribe(subscriber: Subscriber[_ >: T]): Unit = {
+    requireNonNullSubscriber(subscriber)
+
+    @tailrec def doSubscribe(): Unit = {


Long-term safety risk: `FanoutPublisherBridgePublisher` holds a reference to `registerPendingSubscribers: AsyncCallback[Unit]`, which internally holds a reference to the `GraphStageLogic`. If the `Publisher` is retained by user code after the stage terminates, this creates a reference chain that prevents GC of the entire stage logic and all its state (buffer, subscriptions, callbacks). Consider using a weak reference pattern or clearing the callback reference in `shutdown()` to break this chain for long-running applications.

would this be possible to add?

I think there is no need to fix this.

Motivation: FanoutPublisherBridgeStage.postStop always completed the exposed publisher with ActorPublisher.NormalShutdownReason, even when the stage was stopped abruptly without an upstream or subscriber terminal signal. That made active and late subscribers observe a normal shutdown instead of the GraphStage abrupt termination failure. Modification: Track whether the bridge is stopping from an intentional terminal path. Only synthesize AbruptStageTerminationException from postStop when no terminal path was already signalled, and add a bridge-level regression test for active and late subscribers. Result: Controlled completion, failure, timeout, and last-subscriber cancellation keep their existing reasons, while abrupt bridge shutdown is now reported as AbruptStageTerminationException. References: apache#2874

Motivation: FanoutPublisherBridgeStage.postStop always completed the exposed publisher with ActorPublisher.NormalShutdownReason, even when the stage was stopped abruptly without an upstream or subscriber terminal signal. That made active and late subscribers observe a normal shutdown instead of the GraphStage abrupt termination failure. Modification: Track whether the bridge is stopping from an intentional terminal path. Only synthesize AbruptStageTerminationException from postStop when no terminal path was already signalled, and add a bridge-level regression test for active and late subscribers. Result: Controlled completion, failure, timeout, and last-subscriber cancellation keep their existing reasons, while abrupt bridge shutdown is now reported as AbruptStageTerminationException. References: #2874

He-Pin mentioned this pull request Apr 18, 2026

Rewritten more to graphstage #2860

Open

He-Pin marked this pull request as ready for review April 19, 2026 09:23

He-Pin requested a review from Copilot April 19, 2026 09:32

Copilot started reviewing on behalf of He-Pin April 19, 2026 09:32 View session

Copilot AI reviewed Apr 19, 2026

View reviewed changes

He-Pin mentioned this pull request Apr 19, 2026

feat(stream): add GraphStage-backed TLS path behind feature switch (#2860) #2877

Closed

pjfanning approved these changes Apr 19, 2026

View reviewed changes

He-Pin added the t:stream Pekko Streams label Apr 20, 2026

He-Pin added this to the 2.0.0-M2 milestone Apr 20, 2026

He-Pin commented Apr 20, 2026

View reviewed changes

He-Pin merged commit f5c37ba into apache:main Apr 20, 2026
13 checks passed

He-Pin deleted the issue-2860-fanout-graphstage-cleanup branch April 20, 2026 12:41

He-Pin mentioned this pull request Apr 25, 2026

fix(stream): preserve abrupt fanout publisher shutdown signal #2910

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(stream): replace fanout publisher runtime with GraphStage bridge#2874

feat(stream): replace fanout publisher runtime with GraphStage bridge#2874
He-Pin merged 2 commits intoapache:mainfrom
He-Pin:issue-2860-fanout-graphstage-cleanup

He-Pin commented Apr 18, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

pjfanning left a comment

Uh oh!

He-Pin Apr 20, 2026

Uh oh!

pjfanning Apr 20, 2026

Uh oh!

He-Pin Apr 20, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

He-Pin commented Apr 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Motivation

Modification

Result

Validation

Upstream / references

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

pjfanning left a comment

Choose a reason for hiding this comment

Uh oh!

He-Pin Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

pjfanning Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

He-Pin Apr 20, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

He-Pin commented Apr 18, 2026 •

edited

Loading