diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/GraphStageLogicSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/GraphStageLogicSpec.scala
index 28b7b9a524..14e89983eb 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/GraphStageLogicSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/GraphStageLogicSpec.scala
@@ -260,6 +260,8 @@ class GraphStageLogicSpec extends StreamSpec with GraphInterpreterSpecKit with S
 
       // note: a bit dangerous assumptions about connection and logic positions here
       // if anything around creating the logics and connections in the builder changes this may fail
+      val gLogic = interpreter.logics(1)
+      val passThroughLogic = interpreter.logics(2)
       interpreter.complete(interpreter.connections(0))
       interpreter.cancel(interpreter.connections(1), SubscriptionWithCancelException.NoMoreElementsNeeded)
       interpreter.execute(2)
@@ -269,8 +271,8 @@ class GraphStageLogicSpec extends StreamSpec with GraphInterpreterSpecKit with S
 
       interpreter.isCompleted should ===(false)
       interpreter.isSuspended should ===(false)
-      interpreter.isStageCompleted(interpreter.logics(1)) should ===(true)
-      interpreter.isStageCompleted(interpreter.logics(2)) should ===(false)
+      interpreter.isStageCompleted(gLogic) should ===(true)
+      interpreter.isStageCompleted(passThroughLogic) should ===(false)
     }
 
     "not allow push from constructor" in {
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpec.scala
index 10bcb91d40..4dc037f36f 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpec.scala
@@ -315,6 +315,99 @@ class GraphInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
       interpreter.isSuspended should be(false)
     }
 
+    "release references to stage logics when finishing the interpreter" in new TestSetup {
+      val source = new UpstreamProbe[Int]("source")
+      val sink = new DownstreamProbe[Int]("sink")
+      val identityStage = GraphStages.identity[Int]
+
+      builder(identityStage)
+        .connect(source, identityStage.in)
+        .connect(identityStage.out, sink)
+        .init()
+
+      lastEvents() should ===(Set.empty[TestEvent])
+
+      sink.requestOne()
+      lastEvents() should ===(Set(RequestOne(source)))
+
+      source.onNext(1)
+      lastEvents() should ===(Set(OnNext(sink, 1)))
+
+      val logics = interpreter.logics
+      val connections = interpreter.connections
+
+      logics.foreach(logic => logic should not be null)
+      connections.foreach { connection =>
+        connection.inOwner should not be null
+        connection.outOwner should not be null
+        connection.inHandler should not be null
+        connection.outHandler should not be null
+      }
+
+      interpreter.finish()
+
+      logics.foreach(logic => logic should be(null))
+      connections.foreach { connection =>
+        connection.inOwner should be(null)
+        connection.outOwner should be(null)
+        connection.inHandler should be(null)
+        connection.outHandler should be(null)
+      }
+
+      val snapshot = interpreter.toSnapshot
+      snapshot.logics.map(_.label).toSet should ===(Set("<completed>"))
+      snapshot.connections should have size connections.length
+    }
+
+    "release references to completed stage logics while the interpreter keeps running" in new TestSetup {
+      val source1 = new UpstreamProbe[Int]("source1")
+      val source2 = new UpstreamProbe[Int]("source2")
+      val mergeStage = Merge[Int](2)
+      val sink = new DownstreamProbe[Int]("sink")
+
+      builder(mergeStage)
+        .connect(source1, mergeStage.in(0))
+        .connect(source2, mergeStage.in(1))
+        .connect(mergeStage.out, sink)
+        .init()
+
+      lastEvents() should ===(Set.empty[TestEvent])
+
+      sink.requestOne()
+      lastEvents() should ===(Set(RequestOne(source1), RequestOne(source2)))
+
+      val logics = interpreter.logics
+      val connections = interpreter.connections
+
+      logics.foreach(logic => logic should not be null)
+
+      source1.onComplete()
+      lastEvents() should ===(Set.empty[TestEvent])
+
+      logics(0) should be(null)
+      logics(1) should not be null
+      logics(2) should not be null
+      logics(3) should not be null
+
+      connections(0).outOwner should be(null)
+      connections(0).outHandler should be(null)
+      connections(0).inOwner should not be null
+      connections(0).inHandler should not be null
+
+      connections(1).inOwner should not be null
+      connections(1).outOwner should not be null
+      connections(1).inHandler should not be null
+      connections(1).outHandler should not be null
+
+      val snapshot = interpreter.toSnapshot
+      snapshot.connections should have size connections.length
+      snapshot.connections.head.out.label should ===("<completed>")
+      snapshot.connections.head.in.label should not be "<completed>"
+
+      source2.onNext(1)
+      lastEvents() should ===(Set(OnNext(sink, 1), RequestOne(source2)))
+    }
+
   }
 
 }
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
index 7c7db1bca0..6ddf49cb70 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
@@ -244,6 +244,25 @@ import pekko.stream.stage._
   // Marks whether a stage has been finalized (finalizeStage been called) or not
   private val finalizedMark = Array.fill(logics.length)(false)
 
+  private val connectionInStageIds = connectionStageIds(isInput = true)
+  private val connectionOutStageIds = connectionStageIds(isInput = false)
+
+  private def connectionStageIds(isInput: Boolean): Array[Int] = {
+    val stageIds = new Array[Int](connections.length)
+    var i = 0
+    while (i < connections.length) {
+      val connection = connections(i)
+      if (connection ne null) {
+        val owner = if (isInput) connection.inOwner else connection.outOwner
+        stageIds(i) = if (owner ne null) owner.stageId else -1
+      } else {
+        stageIds(i) = -1
+      }
+      i += 1
+    }
+    stageIds
+  }
+
   private var _subFusingMaterializer: Materializer = _
   private lazy val defaultErrorReportingLogLevel = LogLevels.defaultErrorLevel(materializer.system)
 
@@ -342,25 +361,106 @@ import pekko.stream.stage._
     var i = 0
     while (i < logics.length) {
       val logic = logics(i)
-      if (!isStageCompleted(logic) && !isStageFinalized(logic)) {
-        markStageFinalized(logic)
-        finalizeStage(logic)
+      if (logic ne null) {
+        if (!isStageFinalized(logic)) {
+          markStageFinalized(logic)
+          finalizeStage(logic)
+        }
+        logics(i) = null
+      }
+      i += 1
+    }
+    releaseAllConnectionReferences()
+  }
+
+  private def releaseAllConnectionReferences(): Unit = {
+    var i = 0
+    while (i < connections.length) {
+      val connection = connections(i)
+      if (connection ne null) {
+        connection.inHandler = null
+        connection.outHandler = null
+        connection.inOwner = null
+        connection.outOwner = null
+      }
+      i += 1
+    }
+  }
+
+  private def logicName(stageId: Int): String =
+    if (stageId >= 0) {
+      val logic = logics(stageId)
+      if (logic ne null) logic.toString else "<completed>"
+    } else "<completed>"
+
+  private def stageIdFor(stageIds: Array[Int], connection: Connection): Int = {
+    val connectionId = connection.id
+    if (connectionId >= 0 && connectionId < stageIds.length) stageIds(connectionId) else -1
+  }
+
+  private def connectionInStageId(connection: Connection): Int =
+    stageIdFor(connectionInStageIds, connection)
+
+  private def connectionOutStageId(connection: Connection): Int =
+    stageIdFor(connectionOutStageIds, connection)
+
+  private def releaseConnectionReferences(logic: GraphStageLogic): Unit = {
+    val stageId = logic.stageId
+    val portToConn = logic.portToConn
+    var i = 0
+    while (i < portToConn.length) {
+      val connection = portToConn(i)
+      if (connection ne null) {
+        val connectionId = connection.id
+        if (connectionId >= 0 && connectionId < connectionInStageIds.length) {
+          if (connectionInStageIds(connectionId) == stageId) {
+            connection.inHandler = null
+            connection.inOwner = null
+          }
+          if (connectionOutStageIds(connectionId) == stageId) {
+            connection.outHandler = null
+            connection.outOwner = null
+          }
+        }
+      }
+      i += 1
+    }
+  }
+
+  private def finalizeCompletedStage(logic: GraphStageLogic): Unit = {
+    markStageFinalized(logic)
+    runningStages -= 1
+    finalizeStage(logic)
+    releaseConnectionReferences(logic)
+    logics(logic.stageId) = null
+  }
+
+  private def finalizeCompletedStages(): Unit = {
+    var i = 0
+    while (i < logics.length) {
+      val logic = logics(i)
+      if ((logic ne null) && isStageCompleted(logic) && !isStageFinalized(logic)) {
+        finalizeCompletedStage(logic)
       }
       i += 1
     }
   }
 
   // Debug name for a connections input part
-  private def inOwnerName(connection: Connection): String = connection.inOwner.toString
+  private def inOwnerName(connection: Connection): String =
+    if (connection.inOwner ne null) connection.inOwner.toString else logicName(connectionInStageId(connection))
 
   // Debug name for a connections output part
-  private def outOwnerName(connection: Connection): String = connection.outOwner.toString
+  private def outOwnerName(connection: Connection): String =
+    if (connection.outOwner ne null) connection.outOwner.toString else logicName(connectionOutStageId(connection))
 
   // Debug name for a connections input part
-  private def inLogicName(connection: Connection): String = logics(connection.inOwner.stageId).toString
+  private def inLogicName(connection: Connection): String =
+    logicName(connectionInStageId(connection))
 
   // Debug name for a connections output part
-  private def outLogicName(connection: Connection): String = logics(connection.outOwner.stageId).toString
+  private def outLogicName(connection: Connection): String =
+    logicName(connectionOutStageId(connection))
 
   private def shutdownCounters: String =
     shutdownCounter.map(x => if (x >= KeepGoingFlag) s"${x & KeepGoingMask}(KeepGoing)" else x.toString).mkString(",")
@@ -435,7 +535,6 @@ import pekko.stream.stage._
           case NonFatal(e) => reportStageError(e)
         }
         if (pendingFinalization) {
-          pendingFinalization = false
           afterStageHasRun(activeStage)
         }
 
@@ -471,7 +570,6 @@ import pekko.stream.stage._
             case NonFatal(e) => reportStageError(e)
           }
           if (pendingFinalization) {
-            pendingFinalization = false
             afterStageHasRun(activeStage)
           }
         }
@@ -485,7 +583,6 @@ import pekko.stream.stage._
             case NonFatal(e) => reportStageError(e)
           }
           if (pendingFinalization) {
-            pendingFinalization = false
             afterStageHasRun(activeStage)
           }
         }
@@ -559,7 +656,7 @@ import pekko.stream.stage._
           s"$Name CANCEL ${inOwnerName(connection)} -> ${outOwnerName(
               connection)} (${connection.outHandler}) [${outLogicName(connection)}]")
       connection.portState |= OutClosed
-      completeConnection(connection.outOwner.stageId)
+      completeConnection(connectionOutStageId(connection))
       val cause = connection.slot.asInstanceOf[Cancelled].cause
       connection.slot = Empty
       connection.outHandler.onDownstreamFinish(cause)
@@ -574,7 +671,7 @@ import pekko.stream.stage._
                 connection)} (${connection.inHandler}) [${inLogicName(connection)}]")
         connection.portState |= InClosed
         activeStage = connection.inOwner
-        completeConnection(connection.inOwner.stageId)
+        completeConnection(connectionInStageId(connection))
         if ((connection.portState & InFailed) == 0) connection.inHandler.onUpstreamFinish()
         else connection.inHandler.onUpstreamFailure(connection.slot.asInstanceOf[Failed].ex)
       } else {
@@ -630,12 +727,15 @@ import pekko.stream.stage._
     queueTail += 1
   }
 
-  def afterStageHasRun(logic: GraphStageLogic): Unit =
-    if (isStageCompleted(logic) && !isStageFinalized(logic)) {
-      markStageFinalized(logic)
-      runningStages -= 1
-      finalizeStage(logic)
+  def afterStageHasRun(logic: GraphStageLogic): Unit = {
+    if ((logic ne null) && isStageCompleted(logic) && !isStageFinalized(logic)) {
+      finalizeCompletedStage(logic)
+    }
+    if (pendingFinalization) {
+      pendingFinalization = false
+      finalizeCompletedStages()
     }
+  }
 
   // Returns true if the given stage is already completed
   def isStageCompleted(stage: GraphStageLogic): Boolean = (stage ne null) && shutdownCounter(stage.stageId) == 0
@@ -649,11 +749,13 @@ import pekko.stream.stage._
   // Register that a connection in which the given stage participated has been completed and therefore the stage
   // itself might stop, too.
   private def completeConnection(stageId: Int): Unit = {
-    val activeConnections = shutdownCounter(stageId)
-    if (activeConnections > 0) {
-      val next = activeConnections - 1
-      shutdownCounter(stageId) = next
-      if (next == 0) pendingFinalization = true
+    if (stageId >= 0) {
+      val activeConnections = shutdownCounter(stageId)
+      if (activeConnections > 0) {
+        val next = activeConnections - 1
+        shutdownCounter(stageId) = next
+        if (next == 0) pendingFinalization = true
+      }
     }
   }
 
@@ -701,7 +803,7 @@ import pekko.stream.stage._
       enqueue(connection)
     } else if ((currentState & (InClosed | Pushing | Pulling | OutClosed)) == 0) enqueue(connection)
 
-    if ((currentState & OutClosed) == 0) completeConnection(connection.outOwner.stageId)
+    if ((currentState & OutClosed) == 0) completeConnection(connectionOutStageId(connection))
   }
 
   @InternalStableApi
@@ -720,7 +822,7 @@ import pekko.stream.stage._
         enqueue(connection)
       }
     }
-    if ((currentState & OutClosed) == 0) completeConnection(connection.outOwner.stageId)
+    if ((currentState & OutClosed) == 0) completeConnection(connectionOutStageId(connection))
   }
 
   @InternalStableApi
@@ -738,7 +840,7 @@ import pekko.stream.stage._
         enqueue(connection)
       }
     }
-    if ((currentState & InClosed) == 0) completeConnection(connection.inOwner.stageId)
+    if ((currentState & InClosed) == 0) completeConnection(connectionInStageId(connection))
   }
 
   /**
@@ -751,26 +853,29 @@ import pekko.stream.stage._
   def toSnapshot: RunningInterpreter = {
 
     val logicSnapshots = logics.zipWithIndex.map {
-      case (logic, idx) =>
+      case (logic, idx) if logic ne null =>
         LogicSnapshotImpl(idx, logic.toString, logic.attributes)
+      case (_, idx) =>
+        LogicSnapshotImpl(idx, "<completed>", Attributes.none)
     }
-    val logicIndexes = logics.zipWithIndex.map { case (stage, idx) => stage -> idx }.toMap
-    val connectionSnapshots = connections.filter(_ ne null).map { connection =>
-      ConnectionSnapshotImpl(
-        connection.id,
-        logicSnapshots(logicIndexes(connection.inOwner)),
-        logicSnapshots(logicIndexes(connection.outOwner)),
-        connection.portState match {
-          case InReady | Pushing                                           => ConnectionSnapshot.ShouldPull
-          case OutReady | Pulling                                          => ConnectionSnapshot.ShouldPush
-          case x if (x & (InClosed | OutClosed)) == (InClosed | OutClosed) =>
-            // At least one side of the connection is closed: we show it as closed
-            ConnectionSnapshot.Closed
-          case _ =>
-            // This should not be possible: connection alive and both push and pull enqueued but not received
-            throw new IllegalStateException(s"Unexpected connection state for $connection: ${connection.portState}")
-
-        })
+    val connectionSnapshots = connections.zipWithIndex.collect {
+      case (connection, idx)
+          if (connection ne null) && connectionInStageIds(idx) >= 0 && connectionOutStageIds(idx) >= 0 =>
+        ConnectionSnapshotImpl(
+          connection.id,
+          logicSnapshots(connectionInStageIds(idx)),
+          logicSnapshots(connectionOutStageIds(idx)),
+          connection.portState match {
+            case InReady | Pushing                                           => ConnectionSnapshot.ShouldPull
+            case OutReady | Pulling                                          => ConnectionSnapshot.ShouldPush
+            case x if (x & (InClosed | OutClosed)) == (InClosed | OutClosed) =>
+              // At least one side of the connection is closed: we show it as closed
+              ConnectionSnapshot.Closed
+            case _ =>
+              // This should not be possible: connection alive and both push and pull enqueued but not received
+              throw new IllegalStateException(s"Unexpected connection state for $connection: ${connection.portState}")
+
+          })
     }
 
     val stoppedStages: List[LogicSnapshot] = shutdownCounter.zipWithIndex.collect {