test: make mapWithResource cleanup tests deterministic

Motivation:
The abrupt materializer termination tests used Source.single, which can complete across mapWithResource's async boundary and close the resource before shutdown.

Modification:
Keep the upstream open with Source.maybe and assert that the resource has not been closed before shutting down the materializer.

Result:
The tests now distinguish abrupt termination cleanup from normal upstream completion.

Tests:
- FlowMapWithResourceSpec on Scala 2.13 and Scala 3
- sbt +headerCheckAll
- sbt checkCodeStyle
- sbt validatePullRequest (3392 passed, 0 failed)

References:
Refs #3310

Author: He-Pin

stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala

@@ -416,7 +416,7 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
       val promise = Promise[Done]()
       val created = Promise[Done]()
       val matVal = Source
-        .single(1)
+        .maybe[Int]
         .mapWithResource(() => {
           created.trySuccess(Done)
           newBufferedReader()
@@ -429,6 +429,7 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
         .mapConcat(identity)
         .runWith(Sink.never)(mat)
       Await.result(created.future, 3.seconds.dilated) shouldBe Done
+      promise.isCompleted shouldBe false
       mat.shutdown()
       matVal.failed.futureValue shouldBe an[AbruptTerminationException]
       Await.result(promise.future, 3.seconds.dilated) shouldBe Done
@@ -531,10 +532,11 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
         }
       }
       val matVal = Source
-        .single(1)
+        .maybe[Int]
         .mapWithResource(create, (_: AutoCloseable, count) => count)
         .runWith(Sink.never)(mat)
       Await.result(created.future, 3.seconds.dilated) shouldBe Done
+      closedCounter.get shouldBe 0
       mat.shutdown()
       matVal.failed.futureValue shouldBe an[AbruptTerminationException]
       Await.result(promise.future, 3.seconds.dilated) shouldBe Done