Merge branch 'main' into update/jackson-datatype-guava-2.16.0

.scala-steward.conf

@@ -20,4 +20,9 @@ updates.ignore = [
   { groupId = "com.lightbend.sbt", artifactId = "sbt-java-formatter" }
 ]
 
+updates.pin = [
+  # Pin logback to v1.3.x because v1.4.x needs JDK11
+  { groupId = "ch.qos.logback", version="1.3." }
+]
+
 updatePullRequests = "always"

actor-tests/src/test/scala/org/apache/pekko/actor/ActorPathSpec.scala

@@ -32,6 +32,21 @@ class ActorPathSpec extends AnyWordSpec with Matchers {
       ActorPath.fromString(remote).toString should ===(remote)
     }
 
+    "support parsing remote akka paths" in {
+      val remote = "akka://my_sys@host:1234/some/ref"
+      ActorPath.fromString(remote).toString should ===(remote)
+    }
+
+    "support parsing AddressFromURIString" in {
+      val remote = "pekko://my_sys@host:1234"
+      AddressFromURIString(remote) should ===(Address("pekko", "my_sys", Some("host"), Some(1234)))
+    }
+
+    "support parsing akka AddressFromURIString" in {
+      val remote = "akka://my_sys@host:1234"
+      AddressFromURIString(remote) should ===(Address("akka", "my_sys", Some("host"), Some(1234)))
+    }
+
     "throw exception upon malformed paths" in {
       intercept[MalformedURLException] { ActorPath.fromString("") }
       intercept[MalformedURLException] { ActorPath.fromString("://hallo") }

docs/src/main/paradox/logging.md

@@ -119,11 +119,12 @@ pekko {
 }
 ```
 
-This config option is very good if you want to know what config settings are loaded by Pekko:
+This config option is useful if you want to know what config settings are loaded by Pekko:
 
 ```ruby
 pekko {
   # Log the complete configuration at INFO level when the actor system is started.
+  # We do not recommend using this logging in production environments as it can include sensitive values.
   # This is useful when you are uncertain of what configuration is used.
   log-config-on-start = on
 }

docs/src/main/paradox/release-notes/index.md

@@ -8,6 +8,11 @@ A minor bug fix release.
 * Do not render env variables when logging configs. This relates to the optional config `pekko.log-config-on-start`. We do not recommend logging configs in production environments. ([PR771](https://github.com/apache/incubator-pekko/pull/771))
 * Allow `pekko-actor-testkit-typed` to work with slf4j-api v2.0.x. ([PR784](https://github.com/apache/incubator-pekko/pull/784))
 
+### Additional Changes
+
+* Deprecate statefulMapConcat ([#601](https://github.com/apache/incubator-pekko/issues/601))
+* Add section on using Scala 3 Union types to eliminate msg adapters ([PR741](https://github.com/apache/incubator-pekko/pull/741))
+
 ## 1.0.1
 A minor bug fix release. The class renaming described below (`#491`) is not expected to affect anyone
 upgrading from version 1.0.0 but it is strongly recommended that Apache Pekko users switch to the 1.0.1

docs/src/main/paradox/typed/logging.md

@@ -337,11 +337,12 @@ pekko {
 }
 ```
 
-This config option is very good if you want to know what config settings are loaded by Pekko:
+This config option is useful if you want to know what config settings are loaded by Pekko:
 
 ```
 pekko {
   # Log the complete configuration at INFO level when the actor system is started.
+  # We do not recommend using this logging in production environments as it can include sensitive values.
   # This is useful when you are uncertain of what configuration is used.
   log-config-on-start = on
 }

project/Dependencies.scala

@@ -31,7 +31,7 @@ object Dependencies {
   val agronaVersion = "1.19.2"
   val nettyVersion = "4.1.101.Final"
   val protobufJavaVersion = "3.20.3"
-  val logbackVersion = "1.3.11"
+  val logbackVersion = "1.3.13"
 
   val jacksonCoreVersion = "2.14.3"
   val jacksonDatabindVersion = jacksonCoreVersion
@@ -71,7 +71,7 @@ object Dependencies {
 
     val sigar = "org.fusesource" % "sigar" % "1.6.4"
 
-    val jctools = "org.jctools" % "jctools-core" % "4.0.1"
+    val jctools = "org.jctools" % "jctools-core" % "4.0.2"
 
     // reactive streams
     val reactiveStreams = "org.reactivestreams" % "reactive-streams" % reactiveStreamsVersion

project/plugins.sbt

@@ -7,7 +7,7 @@
  * This file is part of the Apache Pekko project, which was derived from Akka.
  */
 
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.4")
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.5")
 
 addSbtPlugin("com.lightbend.sbt" % "sbt-java-formatter" % "0.7.0")
 addSbtPlugin("com.lightbend.sbt" % "sbt-bill-of-materials" % "1.0.2")

stream-tests/src/test/java-jdk9-only/org/apache/pekko/stream/javadsl/FlowUnfoldAsyncTest.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.javadsl;
+
+import org.apache.pekko.japi.Pair;
+import org.apache.pekko.stream.StreamTest;
+import org.apache.pekko.testkit.PekkoJUnitActorSystemResource;
+import org.apache.pekko.testkit.PekkoSpec;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+public class FlowUnfoldAsyncTest extends StreamTest {
+    @ClassRule
+    public static PekkoJUnitActorSystemResource actorSystemResource =
+        new PekkoJUnitActorSystemResource("SourceTest", PekkoSpec.testConf());
+
+    public FlowUnfoldAsyncTest() {
+        super(actorSystemResource);
+    }
+
+    @Test
+    public void testFoldAsync() throws Exception {
+        final Integer result = Source.unfoldAsync(
+                0,
+                idx -> {
+                    if (idx >= 10) {
+                        return CompletableFuture.completedStage(Optional.empty());
+                    } else {
+                        return CompletableFuture.completedStage(Optional.of(Pair.create(idx + 1, idx)));
+                    }
+                })
+            .runFold(0, Integer::sum, system)
+            .toCompletableFuture()
+            .get(3, TimeUnit.SECONDS);
+        Assert.assertEquals(45, result.intValue());
+    }
+}

stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala

@@ -21,7 +21,6 @@ import pekko.stream.impl.Stages.DefaultAttributes
 import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
 
 import java.util.Optional
-import java.util.concurrent.CompletableFuture
 import java.util.concurrent.CompletionStage
 import scala.concurrent.Future
 import scala.util.{ Failure, Success, Try }
@@ -124,21 +123,20 @@ import scala.util.{ Failure, Success, Try }
       }
 
       def onPull(): Unit = {
-        f.apply(state) match {
-          case cf: CompletableFuture[Optional[Pair[S, E]] @unchecked] if cf.isDone && !cf.isCompletedExceptionally =>
-            handle(cf.join())
-          case future =>
-            future.handle((r, ex) => {
-              if (ex != null) {
-                asyncHandler(Failure(ex))
-              } else {
-                asyncHandler(Success(r))
-              }
-              null
-            })
+        val future = f.apply(state).toCompletableFuture
+        if (future.isDone && !future.isCompletedExceptionally) {
+          handle(future.getNow(null))
+        } else {
+          future.handle((r, ex) => {
+            if (ex != null) {
+              asyncHandler(Failure(ex))
+            } else {
+              asyncHandler(Success(r))
+            }
+            null
+          })
         }
       }
-
       setHandler(out, this)
     }
 }