fix the build on main (#31251)

* Fix compilation on scala 2.12

* Add mergePrioritizedN to docs

* Format Java code

akka-docs/src/main/paradox/stream/operators/index.md

@@ -269,6 +269,7 @@ the inputs in different ways.
 |Source/Flow|<a name="mergelatest"></a>@ref[mergeLatest](Source-or-Flow/mergeLatest.md)|Merge multiple sources.|
 |Source/Flow|<a name="mergepreferred"></a>@ref[mergePreferred](Source-or-Flow/mergePreferred.md)|Merge multiple sources.|
 |Source/Flow|<a name="mergeprioritized"></a>@ref[mergePrioritized](Source-or-Flow/mergePrioritized.md)|Merge multiple sources.|
+|Source|<a name="mergeprioritizedn"></a>@ref[mergePrioritizedN](Source/mergePrioritizedN.md)|Merge multiple sources with priorities.|
 |Source/Flow|<a name="mergesorted"></a>@ref[mergeSorted](Source-or-Flow/mergeSorted.md)|Merge multiple sources.|
 |Source/Flow|<a name="orelse"></a>@ref[orElse](Source-or-Flow/orElse.md)|If the primary source completes without emitting any elements, the elements from the secondary source are emitted.|
 |Source/Flow|<a name="prepend"></a>@ref[prepend](Source-or-Flow/prepend.md)|Prepends the given source to the flow, consuming it until completion before the original source is consumed.|
@@ -499,6 +500,7 @@ For more background see the @ref[Error Handling in Streams](../stream-error.md)
 * [mergeLatest](Source-or-Flow/mergeLatest.md)
 * [mergePreferred](Source-or-Flow/mergePreferred.md)
 * [mergePrioritized](Source-or-Flow/mergePrioritized.md)
+* [mergePrioritizedN](Source/mergePrioritizedN.md)
 * [MergeSequence](MergeSequence.md)
 * [mergeSorted](Source-or-Flow/mergeSorted.md)
 * [monitor](Source-or-Flow/monitor.md)

akka-docs/src/test/java/jdocs/stream/javadsl/cookbook/RecipeSplitter.java

@@ -19,62 +19,62 @@ import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
 
 public class RecipeSplitter {
-    private static ActorSystem system;
+  private static ActorSystem system;
 
-    @Test
+  @Test
-    public void simpleSplit() throws ExecutionException, InterruptedException {
+  public void simpleSplit() throws ExecutionException, InterruptedException {
-        // #Simple-Split
+    // #Simple-Split
-        // Sample Source
+    // Sample Source
-        Source<String, NotUsed> source = Source.from(Arrays.asList("1-2-3", "2-3", "3-4"));
+    Source<String, NotUsed> source = Source.from(Arrays.asList("1-2-3", "2-3", "3-4"));
 
-        CompletionStage<List<Integer>> ret =
+    CompletionStage<List<Integer>> ret =
-                source
+        source
-                        .map(s -> Arrays.asList(s.split("-")))
+            .map(s -> Arrays.asList(s.split("-")))
-                        .mapConcat(f -> f)
+            .mapConcat(f -> f)
-                        // Sub-streams logic
+            // Sub-streams logic
-                        .map(s -> Integer.valueOf(s))
+            .map(s -> Integer.valueOf(s))
-                        .runWith(Sink.seq(), system);
+            .runWith(Sink.seq(), system);
 
-        // Verify results
+    // Verify results
-        List<Integer> list = ret.toCompletableFuture().get();
+    List<Integer> list = ret.toCompletableFuture().get();
-        assert list.equals(Arrays.asList(1, 2, 3, 2, 3, 3, 4));
+    assert list.equals(Arrays.asList(1, 2, 3, 2, 3, 3, 4));
-        // #Simple-Split
+    // #Simple-Split
-    }
+  }
 
-    @Test
+  @Test
-    public void splitAggregate() throws ExecutionException, InterruptedException {
+  public void splitAggregate() throws ExecutionException, InterruptedException {
-        // #Aggregate-Split
+    // #Aggregate-Split
-        // Sample Source
+    // Sample Source
-        Source<String, NotUsed> source = Source.from(Arrays.asList("1-2-3", "2-3", "3-4"));
+    Source<String, NotUsed> source = Source.from(Arrays.asList("1-2-3", "2-3", "3-4"));
 
-        CompletionStage<List<Integer>> ret =
+    CompletionStage<List<Integer>> ret =
-                source
+        source
-                        .map(s -> Arrays.asList(s.split("-")))
+            .map(s -> Arrays.asList(s.split("-")))
-                        // split all messages into sub-streams
+            // split all messages into sub-streams
-                        .splitWhen(a -> true)
+            .splitWhen(a -> true)
-                        // now split each collection
+            // now split each collection
-                        .mapConcat(f -> f)
+            .mapConcat(f -> f)
-                        // Sub-streams logic
+            // Sub-streams logic
-                        .map(s -> Integer.valueOf(s))
+            .map(s -> Integer.valueOf(s))
-                        // aggregate each sub-stream
+            // aggregate each sub-stream
-                        .reduce((a, b) -> a + b)
+            .reduce((a, b) -> a + b)
-                        // and merge back the result into the original stream
+            // and merge back the result into the original stream
-                        .mergeSubstreams()
+            .mergeSubstreams()
-                        .runWith(Sink.seq(), system);
+            .runWith(Sink.seq(), system);
 
-        // Verify results
+    // Verify results
-        List<Integer> list = ret.toCompletableFuture().get();
+    List<Integer> list = ret.toCompletableFuture().get();
-        assert list.equals(Arrays.asList(6, 5, 7));
+    assert list.equals(Arrays.asList(6, 5, 7));
-        // #Aggregate-Split
+    // #Aggregate-Split
-    }
+  }
 
-    @BeforeClass
+  @BeforeClass
-    public static void setup() throws Exception {
+  public static void setup() throws Exception {
-        system = ActorSystem.create();
+    system = ActorSystem.create();
-    }
+  }
 
-    @AfterClass
+  @AfterClass
-    public static void teardown() throws Exception {
+  public static void teardown() throws Exception {
-        TestKit.shutdownActorSystem(system);
+    TestKit.shutdownActorSystem(system);
-    }
+  }
 }

akka-docs/src/test/java/jdocs/stream/operators/SourceOrFlow.java

@@ -208,10 +208,8 @@ class SourceOrFlow {
     Source<Integer, NotUsed> sourceA = Source.from(Arrays.asList(1, 2, 3, 4));
     Source<Integer, NotUsed> sourceB = Source.from(Arrays.asList(10, 20, 30, 40));
     Source<Integer, NotUsed> sourceC = Source.from(Arrays.asList(100, 200, 300, 400));
-    List<Pair<Source<Integer, ?>,Integer>> sourcesAndPriorities = Arrays.asList(
+    List<Pair<Source<Integer, ?>, Integer>> sourcesAndPriorities =
-        new Pair<>(sourceA, 9900),
+        Arrays.asList(new Pair<>(sourceA, 9900), new Pair<>(sourceB, 99), new Pair<>(sourceC, 1));
-        new Pair<>(sourceB, 99),
-        new Pair<>(sourceC, 1));
     Source.mergePrioritizedN(sourcesAndPriorities, false).runForeach(System.out::println, system);
     // prints e.g. 1, 100, 2, 3, 4, 10, 20, 30, 40, 200, 300, 400  since both sources have their
     // first element ready and

akka-persistence-testkit/src/main/scala/akka/persistence/testkit/internal/EventSourcedBehaviorTestKitImpl.scala

@@ -255,7 +255,7 @@ import akka.stream.scaladsl.Sink
         case _         => throw new IllegalArgumentException("Cannot initialize from state when snapshots are not used.")
       }
     }
-    persistenceTestKit.persistForRecovery(persistenceId.id, events)
+    persistenceTestKit.persistForRecovery(persistenceId.id, collection.immutable.Seq.empty ++ events)
 
     restart()
   }