Catch thrown errors in parEvalMap, parEvalMapUnordered

core/shared/src/main/scala/fs2/Stream.scala

@@ -2123,7 +2123,7 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
                 Deferred[F2, Unit].flatMap { pushed =>
                   val init = initFork(channel, pushed.complete(()).void)
                   poll(init).onCancel(releaseAndCheckCompletion).flatMap { send =>
-                    val action = f(el).attempt.flatMap(send) *> pushed.get
+                    val action = F.catchNonFatal(f(el)).flatten.attempt.flatMap(send) *> pushed.get
                     F.start(stop.get.race(action) *> releaseAndCheckCompletion)
                   }
                 }

core/shared/src/test/scala/fs2/ParEvalMapSuite.scala

@@ -268,4 +268,26 @@ class ParEvalMapSuite extends Fs2Suite {
       (1 to iterations).toList.as(action).sequence_
     }
   }
+
+  group("issue-2825, Stream shouldn't hang after exceptions thrown in") {
+    test("parEvalMapUnordered") {
+      Stream
+        .eval(IO.unit)
+        .parEvalMapUnordered(Int.MaxValue)(_ => throw new RuntimeException)
+        .compile
+        .drain
+        .attempt
+        .timeout(2.seconds)
+    }
+
+    test("parEvalMap") {
+      Stream
+        .eval(IO.unit)
+        .parEvalMap(Int.MaxValue)(_ => throw new RuntimeException)
+        .compile
+        .drain
+        .attempt
+        .timeout(2.seconds)
+    }
+  }
 }