Error handling

[z4f1r0v]

We streaming Twitter data from a service called GNIP suddenly I get a

ERROR [lt-dispatcher-2] c.s.r.k.KafkaActorSubscriber Stopping Kafka subscriber due to fatal error. WARNING arguments left: 1

As I read in order to understand what is happening I need to implement error handling with actors.
I would like some help on the example. Can someone show me how I can create a sink from the actor that supervises the KafkaActorSubscriber child?

[kciesielski]

There's an example in our Activator Template: https://github.com/softwaremill/activator-reactive-kafka-scala/blob/master/src/main/scala/sample/reactivekafka/KafkaWriterCoordinator.scala
Let me know if you need any further explanation on how to plug in your error handling.

[z4f1r0v]

Made some progress on the matter and ended up with this. Unfortunately I'm still getting the above mentioned error. I would like to know what is this "arguments left: 1"

import akka.actor.SupervisorStrategy.Resume
import akka.actor.{ Props, Actor, ActorLogging, OneForOneStrategy, SupervisorStrategy }
import com.softwaremill.react.kafka.{ ProducerProperties, ReactiveKafka }
import com.typesafe.config.Config
import kafka.serializer.StringEncoder

import scala.collection.JavaConversions._

class ActorSupervisor(config: Config)(implicit val materializer: akka.stream.Materializer) extends Actor with ActorLogging {
  val subscriber: String = "Subscriber"

  override val supervisorStrategy: SupervisorStrategy = OneForOneStrategy() {
    case e =>
      log.error("Exception: {}", e)
      Resume
  }

  private def createSupervisedSubscriberActor() = {
    val kafka = new ReactiveKafka()

    val subscriberProperties = ProducerProperties(
      brokerList = config.getStringList("kafka.brokers").toList.mkString(","),
      topic = config.getString("kafka.topic"),
      clientId = config.getString("kafka.clientId"),
      encoder = new StringEncoder())

    val subscriberActorProps: Props = kafka.producerActorProps(subscriberProperties)
    context.actorOf(subscriberActorProps, subscriber)
  }

  override def aroundPreStart() = {
    createSupervisedSubscriberActor()
  }

  override def receive = {
    case msg => context.child(subscriber) match {
      case Some(ca) => ca ! msg
      case None =>
    }
  }
}

And the sink is created like this

val kafkaSink = Sink.actorSubscriber(Props(new ActorSupervisor(config)))

[z4f1r0v]

According to this you are missing a {} in KafkaActorSubscriber.
Should I make a PR?

[kciesielski]

Sure, a PR will be most welcome :)

[z4f1r0v]

Ok, I see that someone already changed the logging. I guess this will be part of 0.8.2.? @kciesielski any estimates on when you will put that up?

[kciesielski]

@alexanderzafirov I need to verify this one as well: #43, looks like a serious issue. As soon as it's resolved I'll release 0.8.2.

[z4f1r0v]

@kciesielski thanks!

[z4f1r0v]

@kciesielski Just to clarify - by labeling the issue as a bug are you referring to the fact that we cannot catch certain exceptions with the ActorSupervisor?

[kciesielski]

@alexanderzafirov Yes, because we cannot see the cause.

[kciesielski]

Fixed by 35f7a49