SourceWithContextSpec test that fails intermittently

[pjfanning]

• Was part of a 'Nightly Build' but one that was temporarily enabled to run for PRs.
• Failed in a few of the matrix of runs but passed in most runs

[12-17 12:13:48.607] [info] - must Apply a viaFlow with optional elements using unsafeOptionalVia *** FAILED *** (4 seconds, 24 milliseconds)
[12-17 12:13:48.607] [info]   java.lang.AssertionError: assertion failed: expected OnNext((Some(1),1)), found OnNext((None,2))
[12-17 12:13:48.607] [info]   at scala.Predef$.assert(Predef.scala:223)
[12-17 12:13:48.607] [info]   at org.apache.pekko.testkit.TestKitBase.expectMsg_internal(TestKit.scala:473)
[12-17 12:13:48.607] [info]   at org.apache.pekko.testkit.TestKitBase.expectMsg(TestKit.scala:449)
[12-17 12:13:48.607] [info]   at org.apache.pekko.testkit.TestKitBase.expectMsg$(TestKit.scala:449)
[12-17 12:13:48.607] [info]   at org.apache.pekko.testkit.TestKit.expectMsg(TestKit.scala:982)
[12-17 12:13:48.607] [info]   at org.apache.pekko.stream.testkit.TestSubscriber$ManualProbe.$anonfun$expectNextN$1(StreamTestKit.scala:585)
[12-17 12:13:48.607] [info]   at scala.collection.Iterator.foreach(Iterator.scala:943)
[12-17 12:13:48.607] [info]   at scala.collection.Iterator.foreach$(Iterator.scala:943)
[12-17 12:13:48.607] [info]   at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
[12-17 12:13:48.608] [info]   at scala.collection.IterableLike.foreach(IterableLike.scala:74)
[12-17 12:13:48.608] [info]   at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
[12-17 12:13:48.608] [info]   at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
[12-17 12:13:48.608] [info]   at org.apache.pekko.stream.testkit.TestSubscriber$ManualProbe.expectNextN(StreamTestKit.scala:585)
[12-17 12:13:48.608] [info]   at org.apache.pekko.stream.testkit.TestSubscriber$ManualProbe.expectNext(StreamTestKit.scala:554)
[12-17 12:13:48.608] [info]   at org.apache.pekko.stream.scaladsl.SourceWithContextSpec.$anonfun$new$55(SourceWithContextSpec.scala:251)

[He-Pin]

I will look into this weekend

[He-Pin]

I looked into it, the current implementation is using a Merge which will not keep the orgin order.
@mdedetrich cc

So I think the current implementation is broken.