Error handling

Failing consumer

When a consumer fails to read from Kafka due to connection problems, it throws a WakeupException which is handled internally with retries. Refer to consumer configuration settings for details on wakeup-timeout and max-wakeups if you’re interested in tweaking the retry handling parameters. When the last retry fails, source stage will be failed with an exception.

Failing producer

Retry handling for producers is built-in into Kafka. In case of failure when sending a message, an exception will be thrown, which should fail the stream.

Restarting the stream with a backoff stage

Akka streams provides graph stages to gracefully restart a stream on failure, with a configurable backoff. This can be taken advantage of to restart a failing consumer with an exponential backoff, by wrapping it in a RestartSource:

Scala
RestartSource
  .withBackoff(
    minBackoff = 3.seconds,
    maxBackoff = 30.seconds,
    randomFactor = 0.2
  ) { () =>
    Source.fromFuture {
      val source = Consumer.plainSource(consumerSettings, Subscriptions.topics("topic1"))
      source
        .via(business)
        .watchTermination() {
          case (consumerControl, futureDone) =>
            futureDone
              .flatMap { _ =>
                consumerControl.shutdown()
              }
              .recoverWith { case _ => consumerControl.shutdown() }
        }
        .runWith(Sink.ignore)
    }
  }
  .runWith(Sink.ignore)
Full source at GitHub
Java
RestartSource.withBackoff(
    java.time.Duration.of(3, ChronoUnit.SECONDS),
    java.time.Duration.of(30, ChronoUnit.SECONDS),
    0.2,
    () ->
         Source.fromCompletionStage(
              Consumer
                .plainSource(consumerSettings, Subscriptions.topics("topic1"))
                .via(business())
                .watchTermination(
                         (control, completionStage) ->
                             completionStage.handle((res, ex) -> control.shutdown()).thenCompose(Function.identity())
                 )
                .runWith(Sink.ignore(), materializer)
         )
);
Full source at GitHub

When a stream fails, library internals will handle all underlying resources.

(de)serialization

If reading from Kafka failure is caused by other reasons, like deserialization problems, then the stage will fail immediately. If you expect such cases, consider consuming raw byte arrays and deserializing in a subsequent map stage where you can use supervision to skip failed elements. See also the “At least once” page for more suggestions.

The source code for this page can be found here.