This is my route builder. Here i am trying to insert data from my file to topic.Later, i am passing through my main method and using camel context i am running it. I have tried several codes but none helped me. i am working on POC of Apache kafka - Camel.
public class SimpleRouteBuilder extends RouteBuilder {
@Override
public void configure() throws Exception {
String topicName = "test120";
String kafkaServer = "kafka:localhost:9092";
String zooKeeperHost = "zookeeperHost=localhost&zookeeperPort=2181";
String serializerClass = "serializerClass=kafka.serializer.StringEncoder";
String toKafka = "kafka:localhost:9092?topic=test120;zookeeperHost=localhost;zookeeperPort=2181;groupId=group1";
// toKafka = new StringBuilder().append("&").append(serializerClass).toString();
/*new StringBuilder().append(kafkaServer).append("?").append(topicName).append("&")
.append(zooKeeperHost).append("&").append(serializerClass).toString();*/
from("file:C:/inbox?noop=true").to(toKafka);
}
}
This is my pom.xml
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>HelloWorld</groupId>
<artifactId>Pallavi</artifactId>
<version>0.0.1-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-core</artifactId>
<version>2.20.1</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-kafka</artifactId>
<version>2.20.1</version>
</dependency>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.11.0</version>
</dependency>
</dependencies>
</project>
This is my main class:
import org.apache.camel.CamelContext;
import org.apache.camel.impl.DefaultCamelContext;
public class MainApp {
public static void main(String[] args) {
SimpleRouteBuilder routeBuilder = new SimpleRouteBuilder();
CamelContext ctx = new DefaultCamelContext();
try {
ctx.addRoutes(routeBuilder);
ctx.start();
Thread.sleep(5 * 60 * 1000);
ctx.stop();
System.out.println("hi i am working");
}
catch (Exception e) {
e.printStackTrace();
}
}
}
error is :
java.lang.NullPointerException
at java.util.Hashtable.put(Hashtable.java:459)
at org.apache.camel.component.kafka.KafkaProducer.getProps(KafkaProducer.java:63)
at org.apache.camel.component.kafka.KafkaProducer.doStart(KafkaProducer.java:89)
at org.apache.camel.support.ServiceSupport.start(ServiceSupport.java:61)
at org.apache.camel.util.ServiceHelper.startService(ServiceHelper.java:75)
at org.apache.camel.impl.DeferServiceStartupListener.onCamelContextStarted(DeferServiceStartupListener.java:49)
at org.apache.camel.impl.DefaultCamelContext.safelyStartRouteServices(DefaultCamelContext.java:3859)
at org.apache.camel.impl.DefaultCamelContext.doStartOrResumeRoutes(DefaultCamelContext.java:3638)
at org.apache.camel.impl.DefaultCamelContext.doStartCamel(DefaultCamelContext.java:3490)
at org.apache.camel.impl.DefaultCamelContext.access$000(DefaultCamelContext.java:208)
at org.apache.camel.impl.DefaultCamelContext$2.call(DefaultCamelContext.java:3249)
at org.apache.camel.impl.DefaultCamelContext$2.call(DefaultCamelContext.java:3245)
at org.apache.camel.impl.DefaultCamelContext.doWithDefinedClassLoader(DefaultCamelContext.java:3268)
at org.apache.camel.impl.DefaultCamelContext.doStart(DefaultCamelContext.java:3245)
at org.apache.camel.support.ServiceSupport.start(ServiceSupport.java:61)
at org.apache.camel.impl.DefaultCamelContext.start(DefaultCamelContext.java:3168)
at demo.MainApp.main(MainApp.java:13)
Picked up _JAVA_OPTIONS: -Xmx512M -Xms512M
