El consumidor de Scala kafka a continuación no devuelve ningún evento depoll llamada

Sin embargo, el tema es correcto, y puedo ver los eventos que se envían al tema utilizando el consumidor de la consola:

/opt/kafka_2.11-0.10.1.0/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic my_topic --from-beginning

También veo el tema en mi ejemplo de código Scala a continuación cuando lo reviso con un depurador e invocokafkaConsumer.listTopics()

Además, esto se llama desde una sola prueba de unidad, por lo que solo estoy creando una instancia de este rasgo y consumidor (es decir, otra instancia de consumidor no puede estar consumiendo los mensajes). También estoy usando un group_id aleatorio.

¿Hay algún problema con el siguiente código / configuración?

import java.util.Properties

import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.serialization.{ByteArrayDeserializer, StringDeserializer}

import scala.util.Random

trait KafkaTest {

  val kafkaConsumerProperties = new Properties()

  kafkaConsumerProperties.put("bootstrap.servers", "kafka:9092")

  kafkaConsumerProperties.put("group.id", Random.alphanumeric.take(10).mkString)

  kafkaConsumerProperties.put("key.deserializer", classOf[ByteArrayDeserializer])

  kafkaConsumerProperties.put("value.deserializer", classOf[StringDeserializer])

  val kafkaConsumer = new KafkaConsumer[String, String](kafkaConsumerProperties)

kafkaConsumer.subscribe(java.util.Collections.singletonList("my_topic"))

  def checkKafkaHasReceivedEvent(): Assertion = {

    val kafkaEvents = kafkaConsumer.poll(2000) // Always returns 0 events?
    ...
  }
}

Aumentar el tiempo de espera de la encuesta tampoco ayuda.