CODEWITHSUNDEEP
javascalaakkaakka-stream
Jerald Baker
Jerald Baker

Reputation: 1369

How to recover from Exception thrown in Akka Streams Sink?

How can I recover from an exception thrown in the Sink of Akka Streams?

Simple Example:

    Source<Integer, NotUsed> integerSource = Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9));

    integerSource.runWith(Sink.foreach(x -> {
      if (x == 4) {
        throw new Exception("Error Occurred");
      }
      System.out.println("Sink: " + x);
    }), system);

Output:

Sink: 1
Sink: 2
Sink: 3

How can I handle the exception and move on to the next element from the source? (aka 5,6,7,8,9)

Upvotes: 2

Views: 1000

Answers (1)

Jeffrey Chung
Jeffrey Chung

Reputation: 19527

By default, the supervision strategy stops the stream when an exception is thrown. To change the supervision strategy to drop an exception-causing message and proceed to the next message, use the "resume" strategy. For example:

final Function<Throwable, Supervision.Directive> decider =
  exc -> {
    return Supervision.resume();
  };

final Sink<Integer, CompletionStage<Done>> printSink =
  Sink.foreach(x -> {
    if (x == 4) {
      throw new Exception("Error Occurred");
    }
    System.out.println("Sink: " + x);
  });

final RunnableGraph<CompletionStage<Done>> runnableGraph =
  integerSource.toMat(printSink, Keep.right());

final RunnableGraph<CompletionStage<Done>> withResumingSupervision =
  runnableGraph.withAttributes(ActorAttributes.withSupervisionStrategy(decider));

final CompletionStage<Done> result = withResumingSupervision.run(system);

You could also define different supervision strategies for different kinds of exceptions:

final Function<Throwable, Supervision.Directive> decider =
  exc -> {
    if (exc instanceof MySpecificException) return Supervision.resume();
    else return Supervision.stop();
  };

Upvotes: 4

Related Questions