Hello,
i've had hard time trying to poll Kafka messages using Camel 2.20.1.
.setHeader(KafkaConstants.OFFSET, constant("smallest"))
.pollEnrich("kafka:myTopic?brokers=MyHost:1111&groupId=reader", 5000L)
.log("${body}");
All I can find in the logs is
INFO 10688 --- [ qtp43368234-22] o.a.c.c.k.KafkaConsumer :
Starting Kafka consumer on topic: myTopic with breakOnFirstError: false
DEBUG 10688 --- [ qtp43368234-22] o.a.c.c.k.KafkaConsumer
: Starting consumer:
Consumer[kafka://myTopic?brokers=MyHost:1111&groupId=reader]
DEBUG 10688 --- [ qtp43368234-22] o.a.c.c.k.KafkaConsumer
: Kafka consumer groupId is reader
After five seconds timeout the body is null. If I would use same URI in
from("kafka:..") it would work fine.
>From logs it would seem like KafkaFetchRecords is not invoked.
I wasn't able to find kafka-camel tests testing pollEnrich or any newer
examples on-line, can anyone confirm that it's working? what might be
possible cause of my problem?
Besides that can you guys explain how camel-kafka component is designed?
pollEnrich can return single message only, but to my understanding
KafkaConsumer polls messages for some time in infinite loop and gets
multiple messages. Does camel cache those messages ? Or does it disconnect
from Kafka so that it wouldn't consume too much memory after receiving too
many messages?
