chore: Drop JDK 8 (#32127)

* remove java 8 home and other complicated things from the build
* move src from java-jdk9-only, scala-jdk-9, java-jdk-9
* compiler warnings
* migration guide
* fail fast if not JDK 11 or higher
* not anonymous actor class, #32128

.github/workflows/nightly-builds.yml

@@ -85,11 +85,9 @@ jobs:
       matrix:
         include:
           # leaving out some combinations (note that `build-test-prValidations.yml` doesn't run all test
-          - { scalaVersion: "2.13", jdkVersion: "1.8.0",  jvmName: "temurin:1.8.0",  extraOpts: '' }
-          # { scalaVersion: "2.13", jdkVersion: "1.11.0", jvmName: "temurin:1.11.0", extraOpts: '' }
+          - { scalaVersion: "2.13", jdkVersion: "1.11.0", jvmName: "temurin:1.11.0", extraOpts: '' }
           - { scalaVersion: "2.13", jdkVersion: "1.17.0", jvmName: "temurin:1.17.0", extraOpts: '' }
 
-          # { scalaVersion: "3.3",  jdkVersion: "1.8.0",  jvmName: "temurin:1.8.0",  extraOpts: '' }
           # { scalaVersion: "3.3",  jdkVersion: "1.11.0", jvmName: "temurin:1.11.0", extraOpts: '' }
           - { scalaVersion: "3.3",  jdkVersion: "1.17.0", jvmName: "temurin:1.17.0", extraOpts: '' }
     steps:

akka-actor-tests/src/test/scala/akka/actor/FSMActorSpec.scala

@@ -178,18 +178,28 @@ class FSMActorSpec extends AkkaSpec(Map("akka.actor.debug.fsm" -> true)) with Im
 
     "run onTermination upon ActorRef.stop()" in {
       val started = TestLatch(1)
-      /*
-       * This lazy val trick is beyond evil: KIDS, DON'T TRY THIS AT HOME!
-       * It is necessary here because of the path-dependent type fsm.StopEvent.
-       */
-      lazy val fsm = new Actor with FSM[Int, Null] {
-        override def preStart() = { started.countDown() }
+
+      // can't be anonymous class due to https://github.com/akka/akka/issues/32128
+      class FsmActor extends Actor with FSM[Int, Null] {
+        override def preStart() = {
+          started.countDown()
+        }
+
         startWith(1, null)
-        when(1) { FSM.NullFunction }
+        when(1) {
+          FSM.NullFunction
+        }
         onTermination {
           case x => testActor ! x
         }
       }
+
+      /*
+       * This lazy val trick is beyond evil: KIDS, DON'T TRY THIS AT HOME!
+       * It is necessary here because of the path-dependent type fsm.StopEvent.
+       */
+      lazy val fsm = new FsmActor
+
       val ref = system.actorOf(Props(fsm))
       Await.ready(started, timeout.duration)
       system.stop(ref)
@@ -214,8 +224,8 @@ class FSMActorSpec extends AkkaSpec(Map("akka.actor.debug.fsm" -> true)) with Im
     "cancel all timers when terminated" in {
       val timerNames = List("timer-1", "timer-2", "timer-3")
 
-      // Lazy so fsmref can refer to checkTimersActive
-      lazy val fsmref = TestFSMRef(new Actor with FSM[String, Null] {
+      // can't be anonymous class due to https://github.com/akka/akka/issues/32128
+      class FsmActor extends Actor with FSM[String, Null] {
         startWith("not-started", null)
         when("not-started") {
           case Event("start", _) => goto("started").replying("starting")
@@ -233,7 +243,10 @@ class FSMActorSpec extends AkkaSpec(Map("akka.actor.debug.fsm" -> true)) with Im
             testActor ! "stopped"
           }
         }
-      })
+      }
+
+      // Lazy so fsmref can refer to checkTimersActive
+      lazy val fsmref = TestFSMRef(new FsmActor)
 
       def checkTimersActive(active: Boolean): Unit = {
         for (timer <- timerNames) fsmref.isTimerActive(timer) should ===(active)

akka-actor-tests/src/test/scala/akka/actor/SupervisorSpec.scala

@@ -219,27 +219,46 @@ class SupervisorSpec
 
     "restart properly when same instance is returned" in {
       val restarts = 3 //max number of restarts
-      lazy val childInstance = new Actor {
+
+      // can't be anonymous class due to https://github.com/akka/akka/issues/32128
+      class ChildActor extends Actor {
         var preRestarts = 0
         var postRestarts = 0
         var preStarts = 0
         var postStops = 0
+
         override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
-          preRestarts += 1; testActor ! ("preRestart" + preRestarts)
+          preRestarts += 1;
+          testActor ! ("preRestart" + preRestarts)
         }
+
         override def postRestart(reason: Throwable): Unit = {
-          postRestarts += 1; testActor ! ("postRestart" + postRestarts)
+          postRestarts += 1;
+          testActor ! ("postRestart" + postRestarts)
+        }
+
+        override def preStart(): Unit = {
+          preStarts += 1; testActor ! ("preStart" + preStarts)
+        }
+
+        override def postStop(): Unit = {
+          postStops += 1; testActor ! ("postStop" + postStops)
         }
-        override def preStart(): Unit = { preStarts += 1; testActor ! ("preStart" + preStarts) }
-        override def postStop(): Unit = { postStops += 1; testActor ! ("postStop" + postStops) }
+
         def receive = {
-          case "crash" => { testActor ! "crashed"; throw new RuntimeException("Expected") }
-          case "ping"  => sender() ! "pong"
+          case "crash" => {
+            testActor ! "crashed"; throw new RuntimeException("Expected")
+          }
+          case "ping" => sender() ! "pong"
         }
       }
+
+      lazy val childInstance = new ChildActor
+
       val master = system.actorOf(Props(new Actor {
         override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = restarts)(List(classOf[Exception]))
         val child = context.actorOf(Props(childInstance))
+
         def receive = {
           case msg => child.forward(msg)
         }

akka-actor-typed/src/main/scala-jdk-9/akka/actor/typed/internal/jfr/Events.scala → akka-actor-typed/src/main/scala/akka/actor/typed/internal/jfr/Events.scala

@@ -4,6 +4,8 @@
 
 package akka.actor.typed.internal.jfr
 
+import scala.annotation.nowarn
+
 import jdk.jfr.Category
 import jdk.jfr.Enabled
 import jdk.jfr.Event
@@ -34,7 +36,10 @@ final class DeliveryProducerStarted(val producerId: String, val actorPath: Strin
 @Enabled(false) // hi frequency event
 @StackTrace(false)
 @Category(Array("Akka", "Delivery", "ProducerController")) @Label("Delivery ProducerController sent RequestNext")
-final class DeliveryProducerRequestNext(val producerId: String, val currentSeqNr: Long, val confirmedSeqNr: Long)
+final class DeliveryProducerRequestNext(
+    val producerId: String,
+    val currentSeqNr: Long,
+    @nowarn("msg=never used") val confirmedSeqNr: Long)
     extends Event
 
 /** INTERNAL API */
@@ -86,7 +91,10 @@ final class DeliveryProducerReceived(val producerId: String, val currentSeqNr: L
 @Enabled(true)
 @StackTrace(false)
 @Category(Array("Akka", "Delivery", "ProducerController")) @Label("Delivery ProducerController received demand request")
-final class DeliveryProducerReceivedRequest(val producerId: String, val requestedSeqNr: Long, confirmedSeqNr: Long)
+final class DeliveryProducerReceivedRequest(
+    val producerId: String,
+    val requestedSeqNr: Long,
+    @nowarn("msg=never used") confirmedSeqNr: Long)
     extends Event
 
 /** INTERNAL API */

akka-actor-typed/src/main/scala-jdk-9/akka/actor/typed/internal/jfr/JFRActorFlightRecorder.scala → akka-actor-typed/src/main/scala/akka/actor/typed/internal/jfr/JFRActorFlightRecorder.scala

@@ -5,7 +5,6 @@
 package akka.actor.typed.internal.jfr
 
 import akka.actor.ActorPath
-import akka.actor.typed.ActorSystem
 import akka.actor.typed.internal.ActorFlightRecorder
 import akka.actor.typed.internal.DeliveryFlightRecorder
 import akka.annotation.InternalApi

akka-cluster-sharding/src/main/scala-jdk-9/akka/cluster/sharding/internal/jfr/Events.scala → akka-cluster-sharding/src/main/scala/akka/cluster/sharding/internal/jfr/Events.scala

@@ -3,7 +3,7 @@
  */
 
 package akka.cluster.sharding.internal.jfr
-import jdk.jfr.{ Category, Enabled, Event, Label, StackTrace, Timespan }
+import jdk.jfr.{ Category, Event, Label, StackTrace, Timespan }
 
 import akka.annotation.InternalApi
 

akka-docs/src/main/paradox/project/migration-guide-2.8.x-2.9.x.md

@@ -12,18 +12,26 @@ Rolling updates of Akka Cluster from Akka 2.8.x to 2.9.x is fully supported.
 
 A few deprecated features has been removed in Akka 2.9.x, see sections below.
 
-## Cluster Client
+## Support for Java 8 removed
+
+The published artifacts are targeting Java 11, and later. Supported Java versions are 11 and 17.
+
+## Support for Scala 2.12 removed
+
+The published artifacts are targeting Scala 2.13 and Scala 3.3.
+
+## Deprecated Cluster Client removed
 
 Cluster client has been deprecated since Akka 2.6.0 (2019-11-06). Details on how to migrate can be found in the Akka 
 2.6 docs here: https://doc.akka.io/docs/akka/2.6/cluster-client.html#migration-to-akka-grpc
 
 
-## Typed Actor
+## Deprecated Typed Actor removed
 
 The old "Typed Actor" API (`akka.actor.TypedActor`) has been deprecated since Akka 2.6.0 (2019-11-06) and has been dropped.
 No detailed migration guide exists, the recommendation is to move to the new Akka Typed APIs. 
 
-## Akka SSLConfig
+## Deprecated Akka SSLConfig removed
 
 The Akka SSLConfig convenience and methods accepting it has been deprecated since Akka 2.6.0 and has been dropped. 
 Usage should be replaced with directly creating a `javax.net.ssl.SSLEngine` using the JDK APIs.

akka-docs/src/main/paradox/stream/operators/Source/asSubscriber.md

@@ -7,7 +7,7 @@ Integration with Reactive Streams, materializes into a @javadoc[Subscriber](java
 ## Signature
 
 Scala
-:   @@snip[JavaFlowSupport.scala](/akka-stream/src/main/scala-jdk-9/akka/stream/scaladsl/JavaFlowSupport.scala) { #asSubscriber }
+:   @@snip[JavaFlowSupport.scala](/akka-stream/src/main/scala/akka/stream/scaladsl/JavaFlowSupport.scala) { #asSubscriber }
 
 Java
 :   @@snip[JavaFlowSupport.java](/akka-docs/src/test/java-jdk9-only/jdocs/stream/operators/source/AsSubscriber.java) { #api }

akka-docs/src/main/paradox/stream/operators/Source/fromPublisher.md

@@ -7,7 +7,7 @@ Integration with Reactive Streams, subscribes to a @javadoc[Publisher](java.util
 ## Signature
 
 Scala
-:   @@snip[JavaFlowSupport.scala](/akka-stream/src/main/scala-jdk-9/akka/stream/scaladsl/JavaFlowSupport.scala) { #fromPublisher }
+:   @@snip[JavaFlowSupport.scala](/akka-stream/src/main/scala/akka/stream/scaladsl/JavaFlowSupport.scala) { #fromPublisher }
 
 Java
 :   @@snip[JavaFlowSupport.java](/akka-docs/src/test/java-jdk9-only/jdocs/stream/operators/source/FromPublisher.java) { #api }

akka-remote/src/main/scala-jdk-9/akka/remote/artery/jfr/JFRRemotingFlightRecorder.scala → akka-remote/src/main/scala/akka/remote/artery/jfr/JFRRemotingFlightRecorder.scala

@@ -7,7 +7,6 @@ package akka.remote.artery.jfr
 import java.net.InetSocketAddress
 
 import akka.actor.Address
-import akka.actor.ExtendedActorSystem
 import akka.annotation.InternalApi
 import akka.remote.UniqueAddress
 import akka.remote.artery.RemotingFlightRecorder

akka-stream-tests/src/test/java/akka/stream/javadsl/JavaFlowSupportCompileTest.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (C) 2017-2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.stream.javadsl;
+
+import akka.NotUsed;
+import akka.japi.Pair;
+import java.util.concurrent.Flow;
+import org.junit.Test;
+import org.scalatestplus.junit.JUnitSuite;
+
+public class JavaFlowSupportCompileTest extends JUnitSuite {
+  @Test
+  public void shouldCompile() throws Exception {
+    final Flow.Processor<String, String> processor =
+        new Flow.Processor<String, String>() {
+          @Override
+          public void subscribe(Flow.Subscriber<? super String> subscriber) {}
+
+          @Override
+          public void onSubscribe(Flow.Subscription subscription) {}
+
+          @Override
+          public void onNext(String item) {}
+
+          @Override
+          public void onError(Throwable throwable) {}
+
+          @Override
+          public void onComplete() {}
+        };
+
+    final Source<String, Flow.Subscriber<String>> stringSubscriberSource =
+        JavaFlowSupport.Source.asSubscriber();
+    final Source<String, NotUsed> stringNotUsedSource =
+        JavaFlowSupport.Source.fromPublisher(processor);
+
+    final akka.stream.javadsl.Flow<String, String, NotUsed> stringStringNotUsedFlow =
+        JavaFlowSupport.Flow.fromProcessor(() -> processor);
+    final akka.stream.javadsl.Flow<String, String, NotUsed> stringStringNotUsedFlow1 =
+        JavaFlowSupport.Flow.fromProcessorMat(() -> Pair.apply(processor, NotUsed.getInstance()));
+
+    final Sink<String, Flow.Publisher<String>> stringPublisherSink =
+        JavaFlowSupport.Sink.asPublisher(AsPublisher.WITH_FANOUT);
+    final Sink<String, NotUsed> stringNotUsedSink = JavaFlowSupport.Sink.fromSubscriber(processor);
+  }
+}

akka-stream-tests/src/test/scala-jdk9-only/akka/stream/scaladsl/FlowPublisherSinkSpec.scala → akka-stream-tests/src/test/scala/akka/stream/scaladsl/FlowPublisherSinkSpec.scala

@@ -7,26 +7,24 @@ package akka.stream.scaladsl
 import scala.concurrent.Await
 import scala.concurrent.duration._
 
-import akka.stream.ClosedShape
 import akka.stream.testkit.StreamSpec
-import akka.stream.testkit.Utils._
-import akka.stream.testkit.scaladsl.StreamTestKit._
 
 class FlowPublisherSinkSpec extends StreamSpec {
 
   "A FlowPublisherSink" must {
 
     "work with SubscriberSource" in {
-      val (sub, pub) = JavaFlowSupport.Source.asSubscriber[Int].toMat(JavaFlowSupport.Sink.asPublisher(false))(Keep.both).run()
+      val (sub, pub) =
+        JavaFlowSupport.Source.asSubscriber[Int].toMat(JavaFlowSupport.Sink.asPublisher(false))(Keep.both).run()
       Source(1 to 100).to(JavaFlowSupport.Sink.fromSubscriber(sub)).run()
-      Await.result(JavaFlowSupport.Source.fromPublisher(pub).limit(1000).runWith(Sink.seq), 3.seconds) should ===(1 to 100)
+      Await.result(JavaFlowSupport.Source.fromPublisher(pub).limit(1000).runWith(Sink.seq), 3.seconds) should ===(
+        1 to 100)
     }
 
     "be able to use Publisher in materialized value transformation" in {
-      val f = Source(1 to 3).runWith(
-        JavaFlowSupport.Sink.asPublisher[Int](false).mapMaterializedValue { p =>
-          JavaFlowSupport.Source.fromPublisher(p).runFold(0)(_ + _)
-        })
+      val f = Source(1 to 3).runWith(JavaFlowSupport.Sink.asPublisher[Int](false).mapMaterializedValue { p =>
+        JavaFlowSupport.Source.fromPublisher(p).runFold(0)(_ + _)
+      })
 
       Await.result(f, 3.seconds) should be(6)
     }