What happens when a function in a stream pipeline throws an exception?

← Streams and Pipelines · Ref: Q1500

An exception that a stage's code does not catch itself breaks the WHOLE pipeline. No further item is pulled, no later stage runs, nothing more reaches the terminal, and the exception flies to the nearest enclosing try - in the function that runs the pipeline, or in whatever called it. If nothing catches it, the program ends with the Exception's own exit code (see Q1432).

This holds for every stage that runs your code: filter, map, call, flatten, sort by a comparator, group by, uniq by, tee into your own sink, your own source iterator, collect into your own type, and '>' into your own StringOutput. In the program below the first pipeline parses "12", throws on "oops", and the catch prints 'pipeline stopped'; "30" and "45" are never parsed.

TO SKIP A BAD ITEM, CATCH INSIDE THE STAGE

Put the try in the function the stage calls. parsePriceOrZero catches the failure, gives 0 for that item, and the second pipeline carries on to the end.

ASYNC

When an async task throws while others run, the FIRST failure observed is the one that propagates, when the pipeline next takes a result. The other tasks are cancelled and waited for first, so none of them is still running when your catch runs.

RESOURCES

A pipeline inside a try with resources behaves like any other statement there: the exception goes to the catch, then finally runs, then the resources are closed in reverse declaration order.

See Q134 for try/catch. See Q1436 for catching versus letting the program exit.

Example

defines module qa.streams.pipelineexceptions

  defines function

    parsePrice()
      -> text as String
      <- rtn as Integer: Integer(text)
      stdout <- Stdout()
      stdout.println(`parsing ${text}`)
      if not rtn?
        throw Exception(`Not a price: ${text}`, 3)

    parsePriceOrZero()
      -> text as String
      <- rtn as Integer: 0
      try
        rtn: parsePrice(text)
      catch
        -> ex as Exception
        stdout <- Stdout()
        stdout.println(`skipped: ${ex.reason()}`)

  defines program

    PipelineExceptionDemo()
      stdout <- Stdout()
      prices <- ["12", "oops", "30", "45"]

      try
        total <- cat prices | map with parsePrice | collect as List of Integer
        stdout.println(`unreachable: ${total}`)
      catch
        -> ex as Exception
        stdout.println(`pipeline stopped: ${ex.reason()}`)

      recovered <- cat prices | map with parsePriceOrZero | collect as List of Integer
      stdout.println(`recovered: ${recovered}`)
Other ways to ask this
  • Does a stream pipeline keep going after one item throws?
  • How do I catch an exception thrown inside a map or filter stage?
  • What happens to async tasks when one of them throws?
  • Can I skip a bad item in a pipeline instead of stopping it?

Coming from another language?

Java streams: an exception from a lambda also stops the stream and propagates, but a checked exception cannot leave a lambda at all. Python generators: an exception from the generator stops iteration. EK9: every stage behaves the same way, including async (first failure wins, the other tasks are stopped first), and the exception is an ordinary EK9 Exception you catch around the pipeline or inside the stage.

Keywords: exception, filter, stage, map, try, async, propagate, pipeline, catch, stream