I'm new to Java, and I want to write a simple Kafka consumer application in Java.
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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.zhihu</groupId>
<artifactId>kafka-consumer</artifactId>
<packaging>jar</packaging>
<version>1.0-SNAPSHOT</version>
<name>kafkadev</name>
<url>http://maven.apache.org</url>
<repositories>
<repository>
<!-- Proper URL for Cloudera maven artifactory -->
<id>cloudera</id>
<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.1.0</version>
<scope>compile</scope>
</dependency>
<!-- gson -->
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.8.2</version>
</dependency>
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>5.6.12</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.7.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
java source code:
package zhihu;
import java.util.Arrays;
import java.util.Base64;
import java.util.Properties;
import java.io.UnsupportedEncodingException;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import com.google.gson.Gson;
public class ZhihuCommentConsumer {
public static void main(String[] args) throws InterruptedException, UnsupportedEncodingException {
Properties properties = new Properties();
properties.put("bootstrap.servers", "device1:9092,device2:9092");
properties.put("kafka.topic", "zhihu_comment");
properties.put("group.id", "test");
runMainLoop(args, properties);
}
static void runMainLoop(String[] args, Properties properties) throws InterruptedException, UnsupportedEncodingException {
// Create Kafka producer
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
try {
consumer.subscribe(Arrays.asList(properties.getProperty("kafka.topic")));
System.out.println("Subscribed to topic " + properties.getProperty("kafka.topic"));
while (true)
{
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
System.out.printf("partition = %s, offset = %d, key = %s, value = %s\n", record.partition(), record.offset(), record.key(), record.value() );
}
}
}
finally {
consumer.close();
}
}
}
Here is the hierarchy:
├── pom.xml
├── src
│ └── main
│ └── java
│ └── zhihu
│ └── ZhihuCommentConsumer.java
└── target
├── classes
│ └── zhihu
│ └── ZhihuCommentConsumer.class
The question is:
I run mvn package
to generate the jar file in target directory, then execute java -cp target/kafka-consumer-1.0-SNAPSHOT.jar zhihu.ZhihuCommentConsumer
(I'm not sure if this is right).
But, it turns out:
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/kafka/clients/consumer/KafkaConsumer
at zhihu.ZhihuCommentConsumer.runMainLoop(ZhihuCommentConsumer.java:33)
at zhihu.ZhihuCommentConsumer.main(ZhihuCommentConsumer.java:27)
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.clients.consumer.KafkaConsumer
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 2 more
I checked the maven repository .m2
, there is the kafka-client package downloaded already. What should I do to run the application in the right way?
Thanks a lot, I'll appreciate for any help.