Hello,
I am trying to use the pollEnrich to consume messages from a kafka server in
the following manner:
A "timer" based route runs periodically
Use "pollEnrich" with an AggregatorStrategy to get as much messages from the
topic until the pollEnricher expires.
The issues is that the pollEnrich will only get one message at a time, no
matter what type of AggregatorStrategy I use.I saw the following in the
documentation for pollEnrich :
pollEnrich only accept one message as response. That means that if you
target to enrich your original message with the enricher collecting messages
from a seda, ... components using an aggregation strategy. Only one response
message will be aggregated with the original message.
Is there any way that I can get more than one message from kafka in this
manner.
I use the following route:
from("timer://foo?fixedRate=true&period=10000")
.log("IN THE TIMER")
.setHeader("kafka.OFFSET",constant("smallest"))
.pollEnrich("kafka:192.168.111.54:9092?topic=test&"+
"zookeeperHost=192.168.111.54&zookeeperPort=2181&"+
"groupId=groupBatch", 5000L, new
EnrichAggregatorStrategy() )
.beanRef("commonHelper","aggregate") .log("FROM THE TIMER
[${body}]")
The EnrichAggregatorStrategy is a very simple class that extends the
Aggregator Strategy so I will not include that here (unless requested).
Is there any way I can get all the messages that I can within the 5000L time
period?
Thank You,
sunil.
--
View this message in context:
http://camel.465427.n5.nabble.com/pollEnricher-Kafka-Aggregator-tp5775433.html
Sent from the Camel Development mailing list archive at Nabble.com.
