+str #18486 Make sure that Source.subscriber's Subscriber throws if subscribed more than once

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

@@ -379,11 +379,9 @@ private[stream] final class VirtualProcessor[T] extends Processor[T, T] {
               case Completed  ⇒ tryOnComplete(s)
               case Failed(ex) ⇒ tryOnError(s, ex)
               case Allowed    ⇒ // all good
-
             }
           } catch {
-            case ex @ canNotSubscribeTheSameSubscriberMultipleTimesException ⇒ throw ex
-            case NonFatal(ex) ⇒ sub.cancel()
+            case NonFatal(ex) ⇒ if (isCancelled) throw ex else sub.cancel()
           }
       }
   }