=rem #17213: Remove wrong assertion from EndpointManager

(cherry picked from commit 5b6085664ba2d69cd3f5662c1f4b21c417e4afb1)

akka-remote/src/main/scala/akka/remote/Remoting.scala

@@ -624,6 +624,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
       pendingReadHandoffs.valuesIterator foreach (_.disassociate(AssociationHandle.Shutdown))
 
       // Ignore all other writes
+      normalShutdown = true
       context.become(flushing)
   }
 
@@ -769,8 +770,9 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
                              handleOption: Option[AkkaProtocolHandle],
                              writing: Boolean,
                              refuseUid: Option[Int]): ActorRef = {
-    assert(transportMapping contains localAddress)
-    assert(writing || refuseUid.isEmpty)
+    require(transportMapping contains localAddress, "Transport mapping is not defined for the address")
+    // refuseUid is ignored for read-only endpoints since the UID of the remote system is already known and has passed
+    // quarantine checks
 
     if (writing) context.watch(context.actorOf(RARP(extendedSystem).configureDispatcher(ReliableDeliverySupervisor.props(
       handleOption,
@@ -795,8 +797,21 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
       "endpointWriter-" + AddressUrlEncoder(remoteAddress) + "-" + endpointId.next()))
   }
 
+  private var normalShutdown = false
+
   override def postStop(): Unit = {
     pruneTimerCancellable.foreach { _.cancel() }
+    pendingReadHandoffs.valuesIterator foreach (_.disassociate(AssociationHandle.Shutdown))
+
+    if (!normalShutdown) {
+      // Remaining running endpoints are children, so they will clean up themselves.
+      // We still need to clean up any remaining transports because handles might be in mailboxes, and for example
+      // Netty is not part of the actor hierarchy, so its handles will not be cleaned up if no actor is taking
+      // responsibility of them (because they are sitting in a mailbox).
+      log.error("Remoting system has been terminated abrubtly. Attempting to shut down transports")
+      // The result of this shutdown is async, should we try to Await for a short duration?
+      transportMapping.values map (_.shutdown())
+    }
   }
 
 }