We are using apache camel to 2.17.* version inorder to utilize the maxPollRecords parameter I am trying to upgrade to 2.18.0. Post upgrade to 2.18.0 The consumer doesn't seems to be recognized by the broker anymore. Following is the sample consumer I tried to create. I could produce the message from cli to the topic and if I create a consumer in the cli I could see the consumer created in cli consumes the message but not the consumer created through apache camel.
Also with the consumer group describe cli command I could see the consumer-id as blank if I run only apache camel consumer instance. While I was running with 2.17.5 the broker used to recognize and assign that to the partition. I can't find the example please help.
package com.test;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.properties.PropertiesComponent;
import org.apache.camel.impl.DefaultCamelContext;
public class CamelConsumer {
public static void main(String argv[]){
CamelContext camelContext = new DefaultCamelContext();
// Add route to send messages to Kafka
try {
camelContext.addRoutes(new RouteBuilder() {
public void configure() {
PropertiesComponent pc = getContext().getComponent("properties", PropertiesComponent.class);
pc.setLocation("classpath:application.properties");
System.out.println("About to start route: Kafka Server -> Log ");
from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}"
+ "&maxPollRecords={{consumer.maxPollRecords}}" + "&consumersCount={{consumer.consumersCount}}"
+ "&groupId={{consumer.group}}").routeId("FromKafka")
.process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
Message message = exchange.getIn();
Object data = message.getBody();
System.out.println(data);
}
});
}
});
camelContext.start();
Thread.sleep(5 * 60 * 1000);
camelContext.stop();
} catch (Exception e) {
e.printStackTrace();
}
}
}
I donot get any exception either. I donot find any documentation related to this. Please help.
consumer.topic=test kafka.host=localhost kafka.port=9092 consumer.maxPollRecords=1 consumer.consumersCount=1 consumer.group=test