[EXPERIMENT] build with Pekko2

.github/workflows/build-test.yml

@@ -22,11 +22,11 @@ jobs:
           # we don't know what commit the last tag was it's safer to get entire repo so previousStableVersion resolves
           fetch-depth: 0
 
-      - name: Setup Java 11
+      - name: Setup Java 17
         uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad # v6.1.0
         with:
           distribution: temurin
-          java-version: 11
+          java-version: 17
 
       - name: Install sbt
         uses: sbt/setup-sbt@b113b960ba361e87e08ce02a390d51871b7ee459 # v1.5.13

.github/workflows/publish-nightly.yml

@@ -37,11 +37,11 @@ jobs:
           # we don't know what commit the last tag was it's safer to get entire repo so previousStableVersion resolves
           fetch-depth: 0
 
-      - name: Setup Java 11
+      - name: Setup Java 17
         uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad # v6.1.0
         with:
           distribution: temurin
-          java-version: 11
+          java-version: 17
 
       - name: Cache Coursier cache
         uses: coursier/cache-action@ee0650f98295f27a505e4bfd284c9bfaa913a465 # v9.0.1

docs-gen/.sbtopts

@@ -0,0 +1 @@
+-J-Xmx2G

docs-gen/project/build.properties

@@ -1 +1 @@
-sbt.version=1.12.9
+sbt.version=1.13.1

docs-gen/project/plugins.sbt

@@ -1 +1,3 @@
 addSbtPlugin("org.apache.pekko" % "pekko-sbt-paradox" % "1.0.1")
+addSbtPlugin("com.lightbend.paradox" % "sbt-paradox" % "0.11.1")
+addSbtPlugin("com.lightbend.paradox" % "sbt-paradox-theme" % "0.11.1")

pekko-sample-cluster-client-grpc-java/.mvn/jvm.config

@@ -0,0 +1 @@
+--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED

pekko-sample-cluster-client-grpc-java/pom.xml

@@ -10,11 +10,21 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <pekko.version>1.5.0</pekko.version>
-    <pekko.grpc.version>1.2.0</pekko.grpc.version>
+    <pekko.version>2.0.0</pekko.version>
+    <pekko.grpc.version>2.0.0-M2</pekko.grpc.version>
     <scala.binary.version>3</scala.binary.version>
+    <!-- Agrona (used by pekko-remote Artery) needs access to jdk.internal.misc.Unsafe -->
+    <argLine>--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED</argLine>
   </properties>
 
+  <repositories>
+    <repository>
+      <id>apache-staging</id>
+      <name>Apache Staging Repository</name>
+      <url>https://repository.apache.org/content/groups/staging/</url>
+    </repository>
+  </repositories>
+
   <dependencies>
     <dependency>
       <groupId>org.apache.pekko</groupId>
@@ -67,8 +77,7 @@
         <artifactId>maven-compiler-plugin</artifactId>
         <version>3.8.1</version>
         <configuration>
-          <source>1.8</source>
-          <target>1.8</target>
+          <release>17</release>
           <fork>true</fork>
           <compilerArgs>
             <arg>-Xlint</arg>

pekko-sample-cluster-client-grpc-java/src/main/java/sample/cluster/client/grpc/ClusterClient.java

@@ -157,29 +157,26 @@ private static CompletionStage<ActorRef> newSession(
             // never fail from stream element
             elem -> Optional.empty(),
             settings.bufferSize,
-            OverflowStrategy.dropNew()
+            OverflowStrategy.dropTail()
             )
           .via(killSwitch.flow())
           .map(msg -> {
-            if (msg instanceof Send) {
-              Send send = (Send) msg;
+            if (msg instanceof Send send) {
               Payload payload = serialization.serializePayload(send.msg);
               return Req.newBuilder()
                 .setSend(SendReq.newBuilder()
                   .setPath(send.path)
                   .setLocalAffinity(send.localAffinity)
                   .setPayload(payload))
                   .build();
-            } else if (msg instanceof SendToAll) {
-              SendToAll sendToAll = (SendToAll) msg;
+            } else if (msg instanceof SendToAll sendToAll) {
               Payload payload = serialization.serializePayload(sendToAll.msg);
               return Req.newBuilder()
                 .setSendToAll(SendToAllReq.newBuilder()
                   .setPath(sendToAll.path)
                   .setPayload(payload))
                   .build();
-            } else if (msg instanceof Publish) {
-              Publish publish = (Publish) msg;
+            } else if (msg instanceof Publish publish) {
               Payload payload = serialization.serializePayload(publish.msg);
               return Req.newBuilder()
                 .setPublish(PublishReq.newBuilder()

pekko-sample-cluster-client-grpc-java/src/main/java/sample/cluster/client/grpc/ClusterClientReceptionistGrpcImpl.java

@@ -40,7 +40,7 @@ public Source<Rsp, NotUsed> newSession(Source<Req, NotUsed> in) {
         // never fail from stream element
         elem -> Optional.empty(),
         settings.bufferSize,
-        OverflowStrategy.dropNew())
+        OverflowStrategy.dropTail())
       .map( rsp -> {
         Payload payload = serialization.serializePayload(rsp);
         return Rsp.newBuilder().setPayload(payload).build();

pekko-sample-cluster-client-grpc-scala/.jvmopts

@@ -0,0 +1 @@
+--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED

pekko-sample-cluster-client-grpc-scala/build.sbt

@@ -1,6 +1,8 @@
 import com.typesafe.sbt.MultiJvmPlugin.multiJvmSettings
 
-val pekkoVersion = "1.5.0"
+ThisBuild / evictionErrorLevel := Level.Info
+ThisBuild / resolvers += Resolver.ApacheMavenStagingRepo
+val pekkoVersion = "2.0.0"
 
 lazy val `pekko-sample-cluster-client-grpc-scala` = project
   .in(file("."))
@@ -9,7 +11,9 @@ lazy val `pekko-sample-cluster-client-grpc-scala` = project
   .settings(multiJvmSettings: _*)
   .settings(
     organization := "org.apache.pekko",
-    scalaVersion := "3.3.7",
+    // Agrona (used by pekko-remote Artery) needs access to jdk.internal.misc.Unsafe
+    MultiJvm / jvmOptions ++= Seq("--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED"),
+    scalaVersion := "3.3.9",
     Compile / scalacOptions ++= Seq(
       "-deprecation",
       "-feature",

pekko-sample-cluster-client-grpc-scala/project/build.properties

@@ -1 +1 @@
-sbt.version=1.12.9
+sbt.version=1.13.1

pekko-sample-cluster-client-grpc-scala/project/plugins.sbt

@@ -1,3 +1,3 @@
-addSbtPlugin("org.apache.pekko" % "pekko-grpc-sbt-plugin" % "1.2.0")
+addSbtPlugin("org.apache.pekko" % "pekko-grpc-sbt-plugin" % "2.0.0-M2")
 addSbtPlugin("com.github.sbt" % "sbt-javaagent" % "0.2.0")
 addSbtPlugin("com.github.sbt" % "sbt-multi-jvm" % "0.6.0")

pekko-sample-cluster-client-grpc-scala/src/main/scala/sample/cluster/client/grpc/ClusterClient.scala

@@ -65,12 +65,12 @@ object ClusterClient {
         Source
           .actorRef[Any](
             bufferSize = settings.bufferSize,
-            overflowStrategy = OverflowStrategy.dropNew,
+            overflowStrategy = OverflowStrategy.dropTail,
             // never complete from stream element
             completionMatcher = PartialFunction.empty,
             // never fail from stream element
             failureMatcher = PartialFunction.empty)
-          // .actorRef[Any](bufferSize = settings.bufferSize, overflowStrategy = OverflowStrategy.dropNew)
+          // .actorRef[Any](bufferSize = settings.bufferSize, overflowStrategy = OverflowStrategy.dropTail)
           .via(killSwitch.flow)
           .map {
             case send: Send =>

pekko-sample-cluster-client-grpc-scala/src/main/scala/sample/cluster/client/grpc/ClusterClientReceptionist.scala

@@ -17,7 +17,6 @@ import org.apache.pekko.cluster.pubsub.DistributedPubSub
 import org.apache.pekko.cluster.pubsub.DistributedPubSubMediator
 import org.apache.pekko.event.Logging
 import org.apache.pekko.http.scaladsl.Http
-import org.apache.pekko.http.scaladsl.HttpConnectionContext
 import org.apache.pekko.http.scaladsl.model.HttpRequest
 import org.apache.pekko.http.scaladsl.model.HttpResponse
 import org.apache.pekko.stream.Materializer
@@ -100,11 +99,7 @@ final class ClusterClientReceptionist(system: ExtendedActorSystem) extends Exten
       ClusterClientReceptionistServiceHandler(
         new ClusterClientReceptionistGrpcImpl(settings, pubSubMediator, serialization)(materializer, log))
 
-    Http().bindAndHandleAsync(
-      service,
-      interface = settings.hostPort.hostname,
-      settings.hostPort.port,
-      connectionContext = HttpConnectionContext())
+    Http().newServerAt(settings.hostPort.hostname, settings.hostPort.port).bind(service)
   }
 
   server.onComplete { result =>

pekko-sample-cluster-client-grpc-scala/src/main/scala/sample/cluster/client/grpc/ClusterClientReceptionistGrpcImpl.scala

@@ -28,7 +28,7 @@ class ClusterClientReceptionistGrpcImpl(
     Source
       .actorRef[Any](
         bufferSize = settings.bufferSize,
-        overflowStrategy = OverflowStrategy.dropNew,
+        overflowStrategy = OverflowStrategy.dropTail,
         // never complete from stream element
         completionMatcher = PartialFunction.empty,
         // never fail from stream element

pekko-sample-cluster-docker-compose-java/.jvmopts

@@ -0,0 +1 @@
+--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED

pekko-sample-cluster-docker-compose-java/build.sbt

@@ -1,15 +1,16 @@
 organization := "org.apache.pekko"
 
 /* scala versions and options */
-scalaVersion := "3.3.7"
+scalaVersion := "3.3.9"
 
 // These options will be used for *all* versions.
 scalacOptions ++= Seq(
   "-deprecation",
   "-unchecked",
   "-encoding", "UTF-8")
 
-val pekkoVersion = "1.5.0"
+ThisBuild / resolvers += Resolver.ApacheMavenStagingRepo
+val pekkoVersion = "2.0.0"
 val logbackVersion = "1.3.15"
 
 /* dependencies */
@@ -28,5 +29,7 @@ Docker / dockerEntrypoint := Seq("sh", "-c", "bin/clustering $*")
 
 dockerRepository := Some("pekko")
 
-dockerBaseImage := "eclipse-temurin:11"
+dockerBaseImage := "eclipse-temurin:17"
+// Agrona (used by pekko-remote Artery) needs access to jdk.internal.misc.Unsafe
+bashScriptExtraDefines += """addJava "--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED""""
 enablePlugins(JavaAppPackaging)

pekko-sample-cluster-docker-compose-java/project/build.properties

@@ -1 +1 @@
-sbt.version=1.12.9
+sbt.version=1.13.1

pekko-sample-cluster-docker-compose-scala/.jvmopts

@@ -0,0 +1 @@
+--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED

pekko-sample-cluster-docker-compose-scala/build.sbt

@@ -2,15 +2,16 @@ organization := "org.apache.pekko"
 name := "pekko-sample-cluster-docker-compose-scala"
 
 /* scala versions and options */
-scalaVersion := "3.3.7"
+scalaVersion := "3.3.9"
 
 // These options will be used for *all* versions.
 scalacOptions ++= Seq(
   "-deprecation",
   "-unchecked",
   "-encoding", "UTF-8")
 
-val pekkoVersion = "1.5.0"
+ThisBuild / resolvers += Resolver.ApacheMavenStagingRepo
+val pekkoVersion = "2.0.0"
 val logbackVersion = "1.3.15"
 
 /* dependencies */
@@ -24,5 +25,7 @@ libraryDependencies ++= Seq(
 Docker / version := "latest"
 Docker / dockerExposedPorts := Seq(1600)
 dockerRepository := Some("pekko")
-dockerBaseImage := "eclipse-temurin:11"
+dockerBaseImage := "eclipse-temurin:17"
+// Agrona (used by pekko-remote Artery) needs access to jdk.internal.misc.Unsafe
+bashScriptExtraDefines += """addJava "--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED""""
 enablePlugins(JavaAppPackaging)

pekko-sample-cluster-docker-compose-scala/project/build.properties

@@ -1 +1 @@
-sbt.version=1.12.9
+sbt.version=1.13.1

pekko-sample-cluster-java/.jvmopts

@@ -0,0 +1 @@
+--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED

pekko-sample-cluster-java/.mvn/jvm.config

@@ -0,0 +1 @@
+--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED

pekko-sample-cluster-java/build.sbt

@@ -1,14 +1,17 @@
 import com.typesafe.sbt.MultiJvmPlugin.multiJvmSettings
 
-val pekkoVersion = "1.5.0"
+ThisBuild / resolvers += Resolver.ApacheMavenStagingRepo
+val pekkoVersion = "2.0.0"
 val logbackVersion = "1.3.15"
 
 lazy val `pekko-sample-cluster-java` = project
   .in(file("."))
   .settings(multiJvmSettings: _*)
   .settings(
     organization := "org.apache.pekko",
-    scalaVersion := "3.3.7",
+    // Agrona (used by pekko-remote Artery) needs access to jdk.internal.misc.Unsafe
+    MultiJvm / jvmOptions ++= Seq("--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED"),
+    scalaVersion := "3.3.9",
     Compile / scalacOptions ++= Seq("-deprecation", "-feature", "-unchecked"),
     Compile / javacOptions ++= Seq("-parameters", "-Xlint:unchecked", "-Xlint:deprecation"),
     run / javaOptions ++= Seq("-Xms128m", "-Xmx1024m", "-Djava.library.path=./target/native"),

pekko-sample-cluster-java/pom.xml

@@ -11,10 +11,20 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <pekko.version>1.5.0</pekko.version>
+    <pekko.version>2.0.0</pekko.version>
     <scala.binary.version>3</scala.binary.version>
+    <!-- Agrona (used by pekko-remote Artery) needs access to jdk.internal.misc.Unsafe -->
+    <argLine>--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED</argLine>
   </properties>
 
+  <repositories>
+    <repository>
+      <id>apache-staging</id>
+      <name>Apache Staging Repository</name>
+      <url>https://repository.apache.org/content/groups/staging/</url>
+    </repository>
+  </repositories>
+
   <dependencies>
     <dependency>
       <groupId>org.apache.pekko</groupId>
@@ -51,8 +61,7 @@
         <artifactId>maven-compiler-plugin</artifactId>
         <version>3.8.1</version>
         <configuration>
-          <source>1.8</source>
-          <target>1.8</target>
+          <release>17</release>
           <fork>true</fork>
           <compilerArgs>
             <arg>-Xlint</arg>

pekko-sample-cluster-java/project/build.properties

@@ -1 +1 @@
-sbt.version=1.12.9
+sbt.version=1.13.1

pekko-sample-cluster-java/src/main/java/sample/cluster/stats/App.java

@@ -13,10 +13,9 @@
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.List;
 
 public class App {
 
@@ -72,7 +71,7 @@ private static void startup(String role, int port) {
       // Override the configuration of the port
       Map<String, Object> overrides = new HashMap<>();
       overrides.put("pekko.remote.artery.canonical.port", port);
-      overrides.put("pekko.cluster.roles", Collections.singletonList(role));
+      overrides.put("pekko.cluster.roles", List.of(role));
 
       Config config = ConfigFactory.parseMap(overrides)
           .withFallback(ConfigFactory.load("stats"));

pekko-sample-cluster-java/src/main/java/sample/cluster/stats/AppOneMaster.java

@@ -16,10 +16,9 @@
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.List;
 
 public class AppOneMaster {
 
@@ -87,7 +86,7 @@ private static void startup(String role, int port) {
     // Override the configuration of the port
     Map<String, Object> overrides = new HashMap<>();
     overrides.put("pekko.remote.artery.canonical.port", port);
-    overrides.put("pekko.cluster.roles", Collections.singletonList(role));
+    overrides.put("pekko.cluster.roles", List.of(role));
 
     Config config = ConfigFactory.parseMap(overrides)
         .withFallback(ConfigFactory.load("stats"));

pekko-sample-cluster-java/src/main/java/sample/cluster/stats/StatsService.java

@@ -9,7 +9,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import sample.cluster.CborSerializable;
 
-import java.util.Arrays;
 import java.util.List;
 
 public final class StatsService extends AbstractBehavior<StatsService.Command> {
@@ -78,7 +77,7 @@ public Receive<Command> createReceive() {
 
   private Behavior<Command> process(ProcessText command) {
     getContext().getLog().info("Delegating request");
-    List<String> words = Arrays.asList(command.text.split(" "));
+    List<String> words = List.of(command.text.split(" "));
     getContext().spawnAnonymous(StatsAggregator.create(words, workers, command.replyTo));
     return this;
   }

pekko-sample-cluster-java/src/main/java/sample/cluster/transformation/App.java

@@ -7,10 +7,9 @@
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.List;
 
 public class App {
 
@@ -53,7 +52,7 @@ private static void startup(String role, int port) {
     // Override the configuration of the port
     Map<String, Object> overrides = new HashMap<>();
     overrides.put("pekko.remote.artery.canonical.port", port);
-    overrides.put("pekko.cluster.roles", Collections.singletonList(role));
+    overrides.put("pekko.cluster.roles", List.of(role));
 
     Config config = ConfigFactory.parseMap(overrides)
         .withFallback(ConfigFactory.load("transformation"));

pekko-sample-cluster-kubernetes-java/.mvn/jvm.config

@@ -0,0 +1 @@
+--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED