support error handling for KafkaActorSubscriber about alpakka-kafka HOT 6 CLOSED

like add a invoker to props
and then :

  private def processElement(element: T) = {
props.invoker.invoke{
    producer.send(props.encoder.toBytes(element), props.partitionizer(element))
}
  }

from alpakka-kafka.

You can supervise a KafkaActorSubscriber and prevent it from restarting using code like this:
https://github.com/softwaremill/reactive-kafka/blob/master/src/test/scala/examples/examples.scala#L40
It's not perfect and I'm hoping get this refined but I guess that it can should work for now.

from alpakka-kafka.

yes I just found the refactory here
#23
will close it for now.

from alpakka-kafka.

There's another way to handle subscriber errors:

   val sinkDecider: Supervision.Decider = {
      case _ => Supervision.Resume // Your error handling
    }

    Source(publisher)
      .map(_.message().toUpperCase)
      .to(Sink(subscriber).withAttributes(ActorAttributes.supervisionStrategy(sinkDecider)))
      .run()

from alpakka-kafka.

@kciesielski once I realized that I could return the props ,I just attach it as a child actor and the override the supervison strategy.
but from the doc,said that:

ZipWith, FlexiMerge, FlexiRoute junction, ActorPublisher source and ActorSubscriber sink components do not honour the supervision strategy attribute yet.

will the solution you described here works?

from alpakka-kafka.

@hepin1989 Thanks for finding this :) Indeed, I wrote some tests and looks like this decider will never be called. So it looks like the only way to catch exceptions now is to use something similar to the method described in the ReactiveKafkaIntegrationSpec:"close producer, kill actor and log error when in trouble" test.
I'll update the examples and documentation accordingly.

from alpakka-kafka.