Exception is not propagated if there is an uncaught error in Sender

[nemzsom]

Hi!
I found an issue when tried to use the library (version 0.11) with a kafka 0.9.x server. I got this exception:

14:24:13.262 [kafka-producer-network-thread | producer-1] ERROR org.apache.kafka.clients.producer.internals.Sender - Uncaught error in kafka producer I/O thread:
org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'topic_metadata': Error reading array of size 423023, only 34 bytes available
    at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:73)
    at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:380)
    at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:449)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:269)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:229)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:134)
    at java.lang.Thread.run(Thread.java:745)

I know this is caused by the incompatibility between the client (0.10) and the server (0.9) but the issue is that the stream was not terminated. It just hangs. I would expect an exception that a supervisor can handle.

When I run this code:

implicit val system = ActorSystem("TEST")
  implicit val materializer = ActorMaterializer(ActorMaterializerSettings(system))

  val supervisor: Supervision.Decider = {
    case e =>
      println(s"Exception in supervisor: ${e.getMessage}")
      e.printStackTrace()
      Supervision.stop
  }

  val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)
    .withBootstrapServers("localhost:9092")

  val done = Source(1 to 100)
    .map(_.toString)
    .map { elem =>
      new ProducerRecord[String, String]("topic1", elem)
    }
    .runWith(Producer.plainSink(producerSettings).withAttributes(ActorAttributes.supervisionStrategy(supervisor)))

  done.onComplete {
    case _ => println("Stream done")
  }

neither the Exception in supervisor nor the Stream done line is printed.

If I switch to a kafka 0.10 server the code works, but the issue is if an error happens in the communication I would like to get an exception.
I attached a little sbt project to reproduce the issue.

Thank you for this great library!

reactiveKafkaIssue.zip