fix: rethrow exception on failure

akka · Sep 14, 2023 · 272e013 · 272e013
1 parent 9cbe433
commit 272e013
Showing 1 changed file with 4 additions and 3 deletions.
diff --git a/core/src/main/scala/akka/persistence/jdbc/journal/JdbcAsyncWriteJournal.scala b/core/src/main/scala/akka/persistence/jdbc/journal/JdbcAsyncWriteJournal.scala
@@ -118,9 +118,10 @@ class JdbcAsyncWriteJournal(config: Config) extends AsyncWriteJournal {
     journalDao
       .messagesWithBatch(persistenceId, fromSequenceNr, toSequenceNr, journalConfig.daoConfig.replayBatchSize, None)
       .take(max)
-      .collect { case Success(reprAndOrdNr) => reprAndOrdNr }
-      .runForeach { case (repr, _) =>
-        recoveryCallback(repr)
+      .runForeach {
+        case Success((repr, _)) =>
+          recoveryCallback(repr)
+        case Failure(ex) => throw ex
       }
       .map(_ => ())