diff --git a/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala index 147716ea62..ae6f74f5fc 100644 --- a/akka-actor-tests/src/test/scala/akka/actor/dispatch/ActorModelSpec.scala +++ b/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(cachedMessage.latch, waitTime, "Counting down from " + num) assertCountDown(stopLatch, waitTime, "Expected all children to stop") } finally { keepAliveLatch.countDown() diff --git a/akka-actor/src/main/scala/akka/io/SimpleDnsManager.scala b/akka-actor/src/main/scala/akka/io/SimpleDnsManager.scala index a6781ff471..1e5279c64c 100644 --- a/akka-actor/src/main/scala/akka/io/SimpleDnsManager.scala +++ b/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 - override def receive: Receive = { + @nowarn("cat=deprecation") + 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) diff --git a/akka-actor/src/main/scala/akka/io/TcpConnection.scala b/akka-actor/src/main/scala/akka/io/TcpConnection.scala index 53d235888b..0585f6752d 100644 --- a/akka-actor/src/main/scala/akka/io/TcpConnection.scala +++ b/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 { diff --git a/akka-actor/src/main/scala/akka/io/UdpManager.scala b/akka-actor/src/main/scala/akka/io/UdpManager.scala index bc38c091b5..462f7621e5 100644 --- a/akka-actor/src/main/scala/akka/io/UdpManager.scala +++ b/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)) } } diff --git a/akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala b/akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala index 6a1f9d3352..2d4f0e00a8 100644 --- a/akka-cluster/src/main/scala/akka/cluster/ClusterDaemon.scala +++ b/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) diff --git a/akka-docs/src/test/java/jdocs/dispatcher/MyUnboundedMailbox.java b/akka-docs/src/test/java/jdocs/dispatcher/MyUnboundedMailbox.java index 1ef506bf70..5a150c6e4a 100644 --- a/akka-docs/src/test/java/jdocs/dispatcher/MyUnboundedMailbox.java +++ b/akka-docs/src/test/java/jdocs/dispatcher/MyUnboundedMailbox.java @@ -41,7 +41,7 @@ public class MyUnboundedMailbox } public void cleanUp(ActorRef owner, MessageQueue deadLetters) { - while(!queue.isEmpty()) { + while (!queue.isEmpty()) { deadLetters.enqueue(owner, dequeue()); } } diff --git a/akka-remote-tests/src/multi-jvm/scala/akka/remote/classic/RemoteRestartedQuarantinedSpec.scala b/akka-remote-tests/src/multi-jvm/scala/akka/remote/classic/RemoteRestartedQuarantinedSpec.scala deleted file mode 100644 index e3d464725b..0000000000 --- a/akka-remote-tests/src/multi-jvm/scala/akka/remote/classic/RemoteRestartedQuarantinedSpec.scala +++ /dev/null @@ -1,155 +0,0 @@ -/* - * Copyright (C) 2009-2022 Lightbend Inc. - */ - -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) - } - - } - - } -} diff --git a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala index 3f60d96ed0..c29037ea3b 100644 --- a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala +++ b/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 diff --git a/akka-stream/src/main/scala/akka/stream/impl/Sinks.scala b/akka-stream/src/main/scala/akka/stream/impl/Sinks.scala index 39bf4373fc..b0262a7130 100644 --- a/akka-stream/src/main/scala/akka/stream/impl/Sinks.scala +++ b/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 { - def run(): Unit = proc.onSubscriptionTimeout(am, mode) - }) - } - case _ => // not possible to setup timeout + val StreamSubscriptionTimeout(timeout, mode) = + context.effectiveAttributes.mandatoryAttribute[StreamSubscriptionTimeout] + if (mode != StreamSubscriptionTimeoutTerminationMode.noop) { + context.materializer.scheduleOnce(timeout, () => proc.onSubscriptionTimeout(context.materializer, mode)) } (proc, proc) } diff --git a/project/Dependencies.scala b/project/Dependencies.scala index 3c2bf79d58..ee07483164 100644 --- a/project/Dependencies.scala +++ b/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"