我试图用apachestorm、kafka和trident编写实时处理的应用程序,但是在tridentkafkanconfig的初始化中我看到了这个错误
Exception in thread "main" java.lang.NoClassDefFoundError: kafka/api/OffsetRequest
at storm.kafka.KafkaConfig.<init>(KafkaConfig.java:43)
at storm.kafka.trident.TridentKafkaConfig.<init>(TridentKafkaConfig.java:30)
at spout.TestSpout.<clinit>(TestSpout.java:22)
at IOTTridentTopology.initializeTridentTopology(IOTTridentTopology.java:31)
at IOTTridentTopology.main(IOTTridentTopology.java:26)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:601)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)
Caused by: java.lang.ClassNotFoundException: kafka.api.OffsetRequest
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:423)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
at java.lang.ClassLoader.loadClass(ClassLoader.java:356)
... 10 more
我的喷壶课是
public class TestSpout extends OpaqueTridentKafkaSpout {
private static TridentKafkaConfig config;
private static BrokerHosts HOSTS = new ZkHosts(TridentConfig.ZKHOSTS);
private static String TOPIC = "test";
private static int BUFFER_SIZE = TridentConfig.BUFFER_SIZE;
static{
config = new TridentKafkaConfig(HOSTS, TOPIC);
config.scheme = new SchemeAsMultiScheme(new RawScheme());
config.bufferSizeBytes = BUFFER_SIZE;
}
public TestSpout(TridentKafkaConfig config) {
super(config);
}
public TestSpout() {
super(config);
}
}
主要类别:
public static void main(String[] args) {
initializeTridentTopology();
}
private static void initializeTridentTopology() {
TridentTopology topology = new TridentTopology();
TestSpout spout = new TestSpout();
//////////////// test //////////////////////
topology.newStream("testspout", spout).each(spout.getOutputFields(), new TestFunction(), new Fields());
/////////////// end test ///////////////////
LocalCluster cluster = new LocalCluster();
Config config = new Config();
config.setDebug(false);
config.setMaxTaskParallelism(1);
config.registerSerialization(storm.kafka.trident.GlobalPartitionInformation.class);
config.registerSerialization(java.util.TreeMap.class);
config.setNumWorkers(5);
config.setFallBackOnJavaSerialization(true);
cluster.submitTopology("KafkaTrident", config, topology.build());
}
以及我的pom.xml:
<?xml version="1.0" encoding="UTF-8"?>
http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0
<groupId>IOT</groupId>
<artifactId>ver0.1</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>0.9.3</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>0.9.3</version>
</dependency>
</dependencies>
我正在尝试不同版本的storm kafka(0.9.3和0.9.4和0.9.5和0.9.6和0.10.0)和storm core(9.3和9.4和9.6)
但我仍然看到我以前的错误
通过google我找到了这个链接,但是。。。
classnotfoundexception:kafka.api.offsetrequest
2条答案
按热度按时间jobtbby31#
如果使用localcluster部署storm拓扑,则需要将kafka库添加到依赖项(对于storm 0.10.0):
缺少kafka.api.offsetrequest类,因为org.apache.kafka提供了风暴kafka的依赖项:http://mvnrepository.com/artifact/org.apache.storm/storm-kafka/0.10.0. 有关详细信息,请参阅提供的依赖项部分。
pod7payv2#
在谷歌搜索之后,我找到了这个链接https://github.com/wurstmeister/storm-kafka-0.8-plus-test 在pom.xml文件中找到了我的答案
通过添加此代码并找到兼容版本的Kafka,所有问题都得到解决