akka-stream - Akka Kafka Producer でのエラー処理

Question

私は react-kafka-core 0.10.1 を使用しています (Kafka 0.9.x を対象としています)。コールバック関数からエラーが発生するたびに、Kafka プロデューサー アクターが停止しているようです。この動作をカスタマイズする方法はありますか? 私たちの使用例は、メッセージを回復して再送信しようとすることです。

private def processElement(element: ProducerMessage[K, V]) = {
  val record = richProducer.props.partitionizer(element.value) match {
  case Some(partitionId) => new ProducerRecord(richProducer.props.topic, partitionId, element.key, element.value)
  case None => new ProducerRecord(richProducer.props.topic, element.key, element.value)
}
richProducer.producer.send(record, new Callback {
  override def onCompletion(metadata: RecordMetadata, exception: Exception) = {
    if (exception != null) {
      handleError(exception)
    }
  }
})
()} private def handleError(ex: Throwable) = {
log.error(ex, "Stopping Kafka subscriber due to fatal error.")
stop()

}