Kafka 0.9.0.1 Consumer

MyConsumer.scala

class MyConsumer(
    groupId: String,
    broker: String
) extends Logging {

  val props = new Properties();
  props.put(KafkaConsumerConf.BOOTSTRAP_SERVERS_CONFIG, broker)
  props.put(KafkaConsumerConf.GROUP_ID_CONFIG, groupId)
  props.put(KafkaConsumerConf.AUTO_OFFSET_RESET_CONFIG, "earliest")

  val consumer = new KafkaConsumer(
    props,
    new org.apache.kafka.common.serialization.ByteArrayDeserializer(),
    new org.example.kafka.MyDeserializer
  )

  consumer.subscribe(List("mytopic").asJava)

  def read(): Seq[ConsumerRecord[Array[Byte], Try[MyEventStruct]]] = {
    consumer.listTopics() //<- DOESN'T WORK WITHOUT IT
    consumer.poll(100).iterator.asScala.toSeq
  }
}