Hi Camel Users,
I can set key for an individual message using KafkaConstants.KEY in the header.
However, the whole process took extremely long time (more than 20 mins) to
produce 500K messages with key into Kafka topic. I also tried to produce
messages in batch without assigning the key which took 14 seconds for 500K
messages. So my question is how to assign the key for the batch messages?
Processor(GsonStreamProcessor):
if (numberOfRecords % batchSize == 0) {
List<String> batchList = new ArrayList<String>();
exchange.getIn().setBody(batchList);
exchange.getIn().setHeader(KafkaConstants.KEY, "key-"+keyCount); //This
line not working with batch messages
ProducerTemplate template = exchange.getContext().createProducerTemplate();
template.send("vm:msg", exchange);
}
Route:
from("direct:start")
.process("GsonStreamProcessor")
.log("Done");
from("vm:msg")
.to("kafka:.....");
Thank you,
