how to shut down a stream cleanly

[xelax]

I would like to see an example of how to shut down kafka source cleanly and how to detect errors (for example if the kafka server goes down or if the connection fails)

[kciesielski]

I'm working on this now as a part of #21.
As I mentioned there:
Underneath the subscriber is an Akka actor. Errors can be handled with custom behavior using DeathWatch. I'm checking how one can pass more details on exceptions with that.

[kciesielski]

@xelax Error handling can be realized using the API like in following example:
https://github.com/softwaremill/reactive-kafka/blob/master/src/test/scala/examples/examples.scala#L40
You can shut down the source by sending a ActorSubscriberMessage.OnComplete to a KafkaActorSubscriber actor.

[kciesielski]

@xelax you can also use more idiomatic way to handle errors. I'm hoping to add some documentation soon. Here's a quick example:

      val decider: Supervision.Decider = {
        case _ => Supervision.Resume
      }

      Source(kafkaPublisher)
        .withAttributes(ActorAttributes.supervisionStrategy(decider))
        .runWith(someSink)

[kciesielski]

@xelax looks like I gave a wrong example. A source will not restart on errors, it will just propagate the error further by calling onError() on its subscribers. Here's a better example of custom error, handling, which is meaningful for a Sink:

val decider: Supervision.Decider = {
  case _ => Supervision.Resume // Your custom handling
}
Source(publisher)
  .map(_.message().toUpperCase)
  .to(Sink(kafkaSubscribe).withAttributes(ActorAttributes.supervisionStrategy(decider)))
  .run()