support error handling for KafkaActorSubscriber

[He-Pin]

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

I here the encoder may cause an error,and I don't want the actor subscriber to fail.I can't find anyway to handle it.

the akka doc said :

ActorPublisher source and ActorSubscriber sink components do not honour the supervision strategy attribute yet.

[He-Pin]

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))
}
  }

[kciesielski]

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.

[He-Pin]

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

[kciesielski]

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()

[He-Pin]

@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?

[kciesielski]

@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.