Update scala-library to 2.13.8 (#31111)

* Update scala-library to 2.13.8
* Deal with some new deprecation errors
* Remove test for deprecated feature
Co-authored-by: Arnout Engelen <arnout@bzzt.net>

akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala

@@ -19,7 +19,6 @@ import org.scalatest.Assertions._
 
 import akka.actor._
 import akka.dispatch._
-import akka.dispatch.sysmsg.SystemMessageList
 import akka.event.Logging.Error
 import akka.pattern.ask
 import akka.testkit._
@@ -398,32 +397,7 @@ abstract class ActorModelSpec(config: String) extends AkkaSpec(config) with Defa
               }(dispatcher)
           }).start()
           boss ! "run"
-          try {
           assertCountDown(cachedMessage.latch, waitTime, "Counting down from " + num)
-          } catch {
-            case e: Throwable =>
-              dispatcher match {
-                case dispatcher: BalancingDispatcher =>
-                  val team = dispatcher.team
-                  val mq = dispatcher.messageQueue
-
-                  System.err.println(
-                    "Teammates left: " + team.size + " stopLatch: " + stopLatch.getCount + " inhab:" + dispatcher.inhabitants)
-
-                  import akka.util.ccompat.JavaConverters._
-                  team.asScala.toList.sortBy(_.self.path).foreach { (cell: ActorCell) =>
-                    System.err.println(
-                      " - " + cell.self.path + " " + cell.isTerminated + " " + cell.mailbox.currentStatus + " "
-                      + cell.mailbox.numberOfMessages + " " + cell.mailbox.systemDrain(SystemMessageList.LNil).size)
-                  }
-
-                  System.err.println("Mailbox: " + mq.numberOfMessages + " " + mq.hasMessages)
-                  Iterator.continually(mq.dequeue()).takeWhile(_ ne null).foreach(System.err.println)
-                case _ =>
-              }
-
-              throw e
-          }
           assertCountDown(stopLatch, waitTime, "Expected all children to stop")
         } finally {
           keepAliveLatch.countDown()

akka-actor/src/main/scala/akka/io/SimpleDnsManager.scala

@@ -5,13 +5,13 @@
 package akka.io
 
 import java.util.concurrent.TimeUnit
-
 import scala.concurrent.duration.Duration
-
 import akka.actor.{ Actor, ActorLogging, Deploy, Props }
 import akka.dispatch.{ RequiresMessageQueue, UnboundedMessageQueueSemantics }
 import akka.routing.FromConfig
 
+import scala.annotation.nowarn
+
 final class SimpleDnsManager(val ext: DnsExt)
     extends Actor
     with RequiresMessageQueue[UnboundedMessageQueueSemantics]
@@ -38,12 +38,15 @@ final class SimpleDnsManager(val ext: DnsExt)
     system.scheduler.scheduleWithFixedDelay(interval, interval, self, SimpleDnsManager.CacheCleanup)
   }
 
-  // the inet resolver supports the old and new DNS APIs
+  @nowarn("cat=deprecation")
-  override def receive: Receive = {
+  val oldApis: Receive = {
     case r @ Dns.Resolve(name) =>
       log.debug("(deprecated) Resolution request for {} from {}", name, sender())
       resolver.forward(r)
+  }
 
+  // the inet resolver supports the old and new DNS APIs
+  override def receive: Receive = oldApis.orElse {
     case m: dns.DnsProtocol.Resolve =>
       log.debug("Resolution request for {} from {}", m.name, sender())
       resolver.forward(m)

akka-actor/src/main/scala/akka/io/TcpConnection.scala

@@ -421,6 +421,7 @@ private[io] abstract class TcpConnection(val tcp: TcpExt, val channel: SocketCha
   override def postRestart(reason: Throwable): Unit =
     throw new IllegalStateException("Restarting not supported for connection actors.")
 
+  @nowarn("cat=deprecation")
   def PendingWrite(commander: ActorRef, write: WriteCommand): PendingWrite = {
     @tailrec def create(head: WriteCommand, tail: WriteCommand): PendingWrite =
       head match {

akka-actor/src/main/scala/akka/io/UdpManager.scala

@@ -52,9 +52,9 @@ private[io] class UdpManager(udp: UdpExt)
       val commander = sender() // cache because we create a function that will run asynchly
       (registry => Props(classOf[UdpListener], udp, registry, commander, b))
 
-    case SimpleSender(options) =>
+    case s: SimpleSender =>
       val commander = sender() // cache because we create a function that will run asynchly
-      (registry => Props(classOf[UdpSender], udp, registry, commander, options))
+      (registry => Props(classOf[UdpSender], udp, registry, commander, s.options))
   }
 
 }

akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala

@@ -546,6 +546,8 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
     context.become(initialized)
   }
 
+  // Still supports classic remoting as well
+  @nowarn("msg=Classic remoting is deprecated, use Artery")
   def initialized: Actor.Receive =
     ({
       case msg: GossipEnvelope => receiveGossip(msg)

akka-remote-tests/src/multi-jvm/scala/akka/remote/classic/RemoteRestartedQuarantinedSpec.scala

@@ -1,155 +0,0 @@
-/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
- */
-
-package akka.remote.classic
-
-import scala.annotation.nowarn
-import scala.concurrent.Await
-import scala.concurrent.duration._
-import scala.language.postfixOps
-
-import com.typesafe.config.ConfigFactory
-
-import akka.actor.{ ActorIdentity, Identify, _ }
-import akka.remote.{ AddressUidExtension, RARP, RemotingMultiNodeSpec, ThisActorSystemQuarantinedEvent }
-import akka.remote.testconductor.RoleName
-import akka.remote.testkit.MultiNodeConfig
-import akka.testkit._
-
-object RemoteRestartedQuarantinedSpec extends MultiNodeConfig {
-  val first = role("first")
-  val second = role("second")
-
-  commonConfig(
-    debugConfig(on = false).withFallback(ConfigFactory.parseString(
-      """
-      akka.remote.artery.enabled = off
-      # Keep it long, we don't want reconnects
-      akka.remote.classic.retry-gate-closed-for  = 1 s
-
-      # Important, otherwise it is very racy to get a non-writing endpoint: the only way to do it if the two nodes
-      # associate to each other at the same time. Setting this will ensure that the right scenario happens.
-      akka.remote.classic.use-passive-connections = off
-
-      # TODO should not be needed, but see TODO at the end of the test
-      akka.remote.classic.transport-failure-detector.heartbeat-interval = 1 s
-      akka.remote.classic.transport-failure-detector.acceptable-heartbeat-pause = 10 s
-      # test is using Java serialization and not priority to rewrite
-      akka.actor.allow-java-serialization = on
-      akka.actor.warn-about-java-serializer-usage = off
-      """)))
-
-  testTransport(on = true)
-
-  class Subject extends Actor {
-    def receive = {
-      case "shutdown" => context.system.terminate()
-      case "identify" => sender() ! (AddressUidExtension(context.system).longAddressUid -> self)
-    }
-  }
-
-}
-
-class RemoteRestartedQuarantinedSpecMultiJvmNode1 extends RemoteRestartedQuarantinedSpec
-class RemoteRestartedQuarantinedSpecMultiJvmNode2 extends RemoteRestartedQuarantinedSpec
-
-abstract class RemoteRestartedQuarantinedSpec extends RemotingMultiNodeSpec(RemoteRestartedQuarantinedSpec) {
-
-  import RemoteRestartedQuarantinedSpec._
-
-  override def initialParticipants = 2
-
-  def identifyWithUid(role: RoleName, actorName: String): (Long, ActorRef) = {
-    system.actorSelection(node(role) / "user" / actorName) ! "identify"
-    expectMsgType[(Long, ActorRef)]
-  }
-
-  "A restarted quarantined system" must {
-
-    "should not crash the other system (#17213)" taggedAs LongRunningTest in {
-
-      system.actorOf(Props[Subject](), "subject")
-      enterBarrier("subject-started")
-
-      runOn(first) {
-        val secondAddress = node(second).address
-
-        val (uid, _) = identifyWithUid(second, "subject")
-
-        RARP(system).provider.transport.quarantine(node(second).address, Some(uid), "test")
-
-        enterBarrier("quarantined")
-        enterBarrier("still-quarantined")
-
-        testConductor.shutdown(second).await
-
-        within(30.seconds) {
-          awaitAssert {
-            system.actorSelection(RootActorPath(secondAddress) / "user" / "subject") ! Identify("subject")
-            expectMsgType[ActorIdentity](1.second).ref.get
-          }
-        }
-
-        system.actorSelection(RootActorPath(secondAddress) / "user" / "subject") ! "shutdown"
-      }
-
-      runOn(second) {
-        val address = system.asInstanceOf[ExtendedActorSystem].provider.getDefaultAddress
-        val firstAddress = node(first).address
-        @nowarn
-        val thisActorSystemQuarantinedEventCls = classOf[ThisActorSystemQuarantinedEvent]
-        system.eventStream.subscribe(testActor, thisActorSystemQuarantinedEventCls)
-
-        val (_, ref) = identifyWithUid(first, "subject")
-
-        enterBarrier("quarantined")
-
-        // Check that quarantine is intact
-        within(10.seconds) {
-          awaitAssert {
-            EventFilter.warning(pattern = "The remote system has quarantined this system", occurrences = 1).intercept {
-              ref ! "boo!"
-            }
-          }
-        }
-
-        expectMsgPF(10 seconds) {
-          case ThisActorSystemQuarantinedEvent(_, _) =>
-        }
-
-        enterBarrier("still-quarantined")
-
-        Await.result(system.whenTerminated, 10.seconds)
-
-        val freshSystem = ActorSystem(
-          system.name,
-          ConfigFactory.parseString(s"""
-                    akka.remote.retry-gate-closed-for = 0.5 s
-                    akka.remote.classic.netty.tcp {
-                      hostname = ${address.host.get}
-                      port = ${address.port.get}
-                    }
-                    """).withFallback(system.settings.config))
-
-        // retry because it's possible to loose the initial message here, see issue #17314
-        val probe = TestProbe()(freshSystem)
-        probe.awaitAssert(
-          {
-            freshSystem
-              .actorSelection(RootActorPath(firstAddress) / "user" / "subject")
-              .tell(Identify("subject"), probe.ref)
-            probe.expectMsgType[ActorIdentity](1.second).ref should not be None
-          },
-          30.seconds)
-
-        // Now the other system will be able to pass, too
-        freshSystem.actorOf(Props[Subject](), "subject")
-
-        Await.ready(freshSystem.whenTerminated, 10.seconds)
-      }
-
-    }
-
-  }
-}

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

@@ -108,6 +108,8 @@ private[akka] object RemoteActorRefProvider {
       extends DeadLetterActorRef(_provider, _path, _eventStream) {
     import EndpointManager.Send
 
+    // Still supports classic remoting as well
+    @nowarn("msg=Classic remoting is deprecated, use Artery")
     override def !(message: Any)(implicit sender: ActorRef): Unit = message match {
       case Send(m, senderOption, recipient, seqOpt) =>
         // else ignore: it is a reliably delivered message that might be retried later, and it has not yet deserved

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

@@ -83,16 +83,10 @@ import akka.util.ccompat._
   override def create(context: MaterializationContext): (AnyRef, Publisher[In]) = {
 
     val proc = new VirtualPublisher[In]
-    context.materializer match {
-      case am: ActorMaterializer =>
     val StreamSubscriptionTimeout(timeout, mode) =
       context.effectiveAttributes.mandatoryAttribute[StreamSubscriptionTimeout]
     if (mode != StreamSubscriptionTimeoutTerminationMode.noop) {
-          am.scheduleOnce(timeout, new Runnable {
+      context.materializer.scheduleOnce(timeout, () => proc.onSubscriptionTimeout(context.materializer, mode))
-            def run(): Unit = proc.onSubscriptionTimeout(am, mode)
-          })
-        }
-      case _ => // not possible to setup timeout
     }
     (proc, proc)
   }

project/Dependencies.scala

@@ -28,7 +28,7 @@ object Dependencies {
   val jacksonVersion = "2.11.4"
 
   val scala212Version = "2.12.15"
-  val scala213Version = "2.13.7"
+  val scala213Version = "2.13.8"
   // To get the fix for https://github.com/lampepfl/dotty/issues/13106
   // and restored static forwarders
   val scala3Version = "3.1.1-RC2"