chore: Avoid materialization FailedSource

Author: He-Pin

akka-stream/src/main/scala/akka/stream/impl/FailedSource.scala

@@ -12,7 +12,7 @@ import akka.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
 /**
  * INTERNAL API
  */
-@InternalApi private[akka] final class FailedSource[T](failure: Throwable) extends GraphStage[SourceShape[T]] {
+@InternalApi private[akka] final class FailedSource[T](val failure: Throwable) extends GraphStage[SourceShape[T]] {
   val out = Outlet[T]("FailedSource.out")
   override val shape = SourceShape(out)
 

akka-stream/src/main/scala/akka/stream/impl/TraversalBuilder.scala

@@ -374,7 +374,8 @@ import akka.util.OptionVal
    */
   def getValuePresentedSource[A >: Null](graph: Graph[SourceShape[A], _]): OptionVal[Graph[SourceShape[A], _]] = {
     def isValuePresentedSource(graph: Graph[SourceShape[_ <: A], _]): Boolean = graph match {
-      case _: SingleSource[_] | _: FutureSource[_] | _: IterableSource[_] | _: JavaStreamSource[_, _] | EmptySource =>
+      case _: SingleSource[_] | _: FutureSource[_] | _: IterableSource[_] | _: JavaStreamSource[_, _] | EmptySource |
+          _: FailedSource[_] =>
         true
       case _ => false
     }

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

@@ -13,6 +13,7 @@ import akka.stream._
 import akka.stream.impl.{
   ActorSubscriberMessage,
   EmptySource,
+  FailedSource,
   JavaStreamSource,
   SubscriptionTimeoutException,
   TraversalBuilder,
@@ -32,7 +33,7 @@ import scala.annotation.{ nowarn, tailrec }
 import scala.collection.immutable
 import scala.concurrent.{ ExecutionContext, Future }
 import scala.concurrent.duration.FiniteDuration
-import scala.util.Try
+import scala.util.{ Failure, Try }
 import scala.util.control.NonFatal
 
 /**
@@ -402,8 +403,9 @@ private[akka] final class FlattenConcat[T, M](parallelism: Int)
               case iterable: IterableSource[T] @unchecked => addSourceElements(iterable.elements.iterator)
               case javaStream: JavaStreamSource[T, _] @unchecked =>
                 addSourceElements(javaStream.open().iterator.asScala)
-              case EmptySource => //Empty source is discarded
-              case _           => attachAndMaterializeSource(source)
+              case failed: FailedSource[T] @unchecked => addCompletedFutureElem(Failure(failed.failure))
+              case EmptySource                        => //Empty source is discarded
+              case _                                  => attachAndMaterializeSource(source)
             }
           case _ => attachAndMaterializeSource(source)
         }