apache-storm - KafkaSpout 中的异常
问题描述
我正在关注异常风暴拓扑。
java.lang.NoSuchMethodError: org.apache.kafka.common.network.NetworkSend.<init>(Ljava/lang/String;[Ljava/nio/ByteBuffer;)V
at kafka.network.RequestOrResponseSend.<init>(RequestOrResponseSend.scala:41) ~[stormjar.jar:?]
at kafka.network.RequestOrResponseSend.<init>(RequestOrResponseSend.scala:44) ~[stormjar.jar:?]
at kafka.network.BlockingChannel.send(BlockingChannel.scala:112) ~[stormjar.jar:?]
at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:98) ~[stormjar.jar:?]
at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:83) ~[stormjar.jar:?]
at kafka.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:149) ~[stormjar.jar:?]
at kafka.javaapi.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:79) ~[stormjar.jar:?]
at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:81) ~[stormjar.jar:?]
at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:71) ~[stormjar.jar:?]
at org.apache.storm.kafka.PartitionManager.<init>(PartitionManager.java:135) ~[stormjar.jar:?]
at org.apache.storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:110) ~[stormjar.jar:?]
at org.apache.storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:71) ~[stormjar.jar:?]
at org.apache.storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:135) ~[stormjar.jar:?]
at org.apache.storm.daemon.executor$fn__10727$fn__10742$fn__10773.invoke(executor.clj:654) ~[storm-core-1.2.2.jar:1.2.2]
at org.apache.storm.util$async_loop$fn__553.invoke(util.clj:484) [storm-core-1.2.2.jar:1.2.2]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_151]
POM 配置:
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<!-- <version>0.10.0</version> -->
<version>1.2.2</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<artifactId>log4j-core</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-api</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<!-- <version>0.10.0</version> -->
<version>1.2.2</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>1.0.1</version>
<exclusions>
<exclusion>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
我正在使用已弃用的storm-kafka 库。如果这是上述异常的原因,那么让我知道如何使用storm-kafka-client 库创建kafka spout 并将自定义方案传递给它。
谢谢。
解决方案
您能否尝试将org.apache.kafka:kafka-clients
工件也放入您的依赖项中,与 相同的版本kafka_2.11
?
关于如何使用storm-kafka-client,在 https://storm.apache.org/releases/2.0.0-SNAPSHOT/storm-kafka-client.html 的 Storm 页面上有文档,在https://github有示例.com/apache/storm/blob/master/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutTopologyMainNamedTopics.java
具体来说,您想要的是RecordTranslator
.
ByTopicRecordTranslator<String, String> trans = new ByTopicRecordTranslator<>(
(r) -> new Values(r.topic(), r.partition(), r.offset(), r.key(), r.value()),
new Fields("topic", "partition", "offset", "key", "value"), TOPIC_0_1_STREAM);
trans.forTopic(TOPIC_2,
(r) -> new Values(r.topic(), r.partition(), r.offset(), r.key(), r.value()),
new Fields("topic", "partition", "offset", "key", "value"), TOPIC_2_STREAM);
return KafkaSpoutConfig.builder(bootstrapServers, new String[]{TOPIC_0, TOPIC_1, TOPIC_2})
.setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup")
.setRetry(getRetryService())
.setRecordTranslator(trans)
.setOffsetCommitPeriodMs(10_000)
.setFirstPollOffsetStrategy(EARLIEST)
.setMaxUncommittedOffsets(250)
.build();
例如,这会从列出的字段中的每条记录中输出主题、分区、偏移量、键和值,并将从 TOPIC_2 发出元组到与其他订阅主题不同的流。如果您不需要针对不同主题使用不同的方案,则可以SimpleRecordTranslator
改用。
推荐阅读
- qt - Felgo模拟器中的元素尺寸与真机上的显示尺寸相差较大
- architecture - 两个组件的方向,财务报表生成器如何从数据库中获取数据?
- javascript - 移动视图图像上的滑块未显示
- fullcalendar - fullcalendar 更改月份视图的开始和结束
- gnuplot - 如何在 gnuplot 中使用图像命令绘制具有不同颜色圆圈的 100x100 数据集?
- php - 在laravel中使用表中的对象数组
- flutter - 如何创建在使用 AnimationController 创建的倒数计时器期间发出的警报?
- java - Java中的基本继承帮助
- c# - 在配置文件中填写部分
- ios - 无法在 iOS 12 的文档选择器中选择文件夹