Call BoundaryEvent cancellation hook for telemetry when ActorGraphInt…

…erpreter stops (#30259)

* Call BoundaryEvent cancellation hook for telemetry when ActorGraphInterpreter stops
* Add missing annotation to ActorCell.sendMessage

akka-actor/src/main/scala/akka/actor/ActorCell.scala

@@ -322,6 +322,7 @@ private[akka] trait Cell {
    * schedule the actor to run, depending on which type of cell it is.
    * Is only allowed to throw Fatal Throwables.
    */
+  @InternalStableApi
   final def sendMessage(message: Any, sender: ActorRef): Unit =
     sendMessage(Envelope(message, sender, system))
 

akka-stream/src/main/scala/akka/stream/impl/fusing/ActorGraphInterpreter.scala

@@ -829,12 +829,23 @@ import akka.util.OptionVal
         newShells.map(shell => shell.toSnapshot.asInstanceOf[UninitializedInterpreter]))
   }
 
-  override def postStop(): Unit =
+  override def postStop(): Unit = {
+    if (shortCircuitBuffer ne null) {
+      while (!shortCircuitBuffer.isEmpty) {
+        shortCircuitBuffer.poll() match {
+          case b: BoundaryEvent =>
+            // signal to telemetry that this event won't be processed
+            b.cancel()
+          case _ => // ignore
+        }
+      }
+    }
     // avoid creating exception in happy case since it uses self.toString which is somewhat slow
     if (activeInterpreters.nonEmpty || newShells.nonEmpty) {
       val ex = AbruptTerminationException(self)
       activeInterpreters.foreach(_.tryAbort(ex))
       activeInterpreters = Set.empty[GraphInterpreterShell]
       newShells.foreach(s => if (tryInit(s)) s.tryAbort(ex))
     }
+  }
 }