Merge pull request #8 from CatTail/chore/error-message

return detailed error message than BackpressureTimeout
aserrallerios · Sep 4, 2018 · cbc8ae4 · cbc8ae4
2 parents e5ad806 + d2cde96
commit cbc8ae4
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 8 deletions.
diff --git a/src/main/scala/aserralle/akka/stream/kcl/Errors.scala b/src/main/scala/aserralle/akka/stream/kcl/Errors.scala
@@ -4,14 +4,12 @@
 
 package aserralle.akka.stream.kcl
 
-import scala.util.control.NoStackTrace
-
 object Errors {
+  sealed class KinesisWorkerSourceError(err: Throwable) extends Throwable(err)
 
-  sealed trait KinesisWorkerSourceError extends NoStackTrace
   case class WorkerUnexpectedShutdown(cause: Throwable)
-      extends KinesisWorkerSourceError
-
-  case object BackpressureTimeout extends KinesisWorkerSourceError
+      extends KinesisWorkerSourceError(cause)
 
+  case class BackpressureTimeout(cause: Throwable)
+      extends KinesisWorkerSourceError(cause)
 }
diff --git a/src/main/scala/aserralle/akka/stream/kcl/scaladsl/KinesisWorkerSource.scala b/src/main/scala/aserralle/akka/stream/kcl/scaladsl/KinesisWorkerSource.scala
@@ -53,7 +53,7 @@ object KinesisWorkerSource {
                     (Exception.nonFatalCatch either Await.result(
                       queue.offer(record),
                       settings.backpressureTimeout) left)
-                      .foreach(_ => queue.fail(BackpressureTimeout))
+                      .foreach(err => queue.fail(BackpressureTimeout(err)))
                     semaphore.release()
                   },
                   settings.terminateStreamGracePeriod

diff --git a/src/test/scala/aserralle/akka/stream/kcl/KinesisWorkerSourceSourceSpec.scala b/src/test/scala/aserralle/akka/stream/kcl/KinesisWorkerSourceSourceSpec.scala
@@ -202,7 +202,9 @@ class KinesisWorkerSourceSourceSpec
 
       Await.ready(watch, 5.seconds)
       val Failure(exception) = watch.value.get
-      assert(exception == BackpressureTimeout)
+      assert(
+        exception.getCause.getMessage
+          .contains("Futures timed out after [100 milliseconds]"))
 
       killSwitch.shutdown()
     }