Kafka consumer not returning any events

问题:

The below Scala kafka consumer is not returning any events from the poll call.

However, the topic is correct, and I can see events being sent to the topic using the console consumer:

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

I also see the topic in my Scala code sample below when I step through it with a debugger and invoke kafkaConsumer.listTopics()

Also, this is called from a single unit test, so I'm only creating one instance of this trait and consumer (i.e. another consumer instance can't be consuming the messages). I'm also using a random group_id.

Is there anything wrong with the below code/configuration?

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?
    ...
  }
}

Increasing the poll timeout doesn't help either.

回答1:

To read from beginning AUTO_OFFSET_RESET_CONFIG property has to be set to earliest, by default it "latest"

kafkaConsumerProperties.put(
    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, 
    OffsetResetStrategy.EARLIEST.toString().toLowerCase())