How to detect consumption error while using KafkaConsumerObservable?

[yeryomenkom]

Hi!
I couldn't find any test in the lib source that tests the behaviour of KafkaConsumerObservable in case of kafka shutdown. Can you explain me when KafkaConsumerObservable should fail?
Currently I am not able to make KafkaConsumerObservable fail...

[Avasil]

It could fail if poll fails so you could try replicating one of the conditions

[yeryomenkom]

Yeah. I saw it in the code.
But in general I am expecting that this test should pass: https://github.com/monix/monix-kafka/pull/183/files

[Avasil]

It seems like consumer.poll hangs if there is no conenction:

  test("observable should become completed after kafka broker shutdown") {
    EmbeddedKafka.start()
    assert(EmbeddedKafka.isRunning)
    val props = new Properties()
    props.setProperty("bootstrap.servers", "127.0.0.1:6001")
    props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    val consumer = new KafkaConsumer[String, String](props)
    consumer.subscribe(topicsRegex.pattern)

    EmbeddedKafka.stop()
    assert(!EmbeddedKafka.isRunning)

    consumer.poll(0)
  }

We probably have to listen for shutdown in a separate thread