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 currently configured number of max-wakeups is reached, the source stage will fail with an exception and stop.

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

val control = new AtomicReference[Consumer.Control](Consumer.NoopControl)

val result = RestartSource
  .onFailuresWithBackoff(
    minBackoff = 3.seconds,
    maxBackoff = 30.seconds,
    randomFactor = 0.2
  ) { () =>
    Consumer
      .plainSource(consumerSettings, Subscriptions.topics(topic))
      // this is a hack to get access to the Consumer.Control
      // instances of the latest Kafka Consumer source
      .mapMaterializedValue(c => control.set(c))
      .via(businessFlow)
  }
  .runWith(Sink.seq)

control.get().shutdown()

Java

AtomicReference<Consumer.Control> control = new AtomicReference<>(Consumer.createNoopControl());

RestartSource.onFailuresWithBackoff(
        java.time.Duration.ofSeconds(3),
        java.time.Duration.ofSeconds(30),
        0.2,
        () ->
            Consumer.plainSource(consumerSettings, Subscriptions.topics(topic))
                .mapMaterializedValue(
                    c -> {
                      control.set(c);
                      return c;
                    })
                .via(business()))
    .runWith(Sink.ignore(), materializer);

control.get().shutdown();

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