block8.kt

val config = HashMap<String, String>()
config["bootstrap.servers"] = "kafka:9092"
config["key.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
config["value.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
config["group.id"] = "my_group"
config["auto.offset.reset"] = "latest" //earliest --> startet am anfang
config["enable.auto.commit"] = "false"

val consumer: KafkaConsumer<String, String> = KafkaConsumer.create(vertx, config)
consumer.subscribe("kafkaqueue")

consumer.handler{
        println(it.value())
      }