【发布时间】：2018-11-03 10:18:36
【问题描述】：
我正在使用 OpaqueTridentKafkaSpout 来使用来自 Kafka 的消息。下面是代码。我忽略了max spout pending 配置，因为这会导致相同的 kafka 消息分批到达。
TridentKafkaConfig tridentKafkaConfig = new TridentKafkaConfig(hosts,properties.getProperty("topic", "mytopic"));
tridentKafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
OpaqueTridentKafkaSpout kafkaSpout = new OpaqueTridentKafkaSpout(tridentKafkaConfig);
我在 Kafka Spout 启动时收到以下错误，但之后运行顺利。
2018-05-29 09:47:21.703 o.a.s.util Thread-9-spout-myspout-Spout-executor [33 33] [错误] 异步循环死了！ java.lang.RuntimeException: java.lang.NullPointerException 在 org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor（DisruptorQueue.java:522） ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.utils.DisruptorQueue.consumeBatchWhenAvailable（DisruptorQueue.java:487） ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:74) ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.daemon.executor$fn__5043$fn__5056$fn__5109.invoke(executor.clj:861) ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.util$async_loop$fn__557.invoke(util.clj:484) [风暴核心-1.2.1.jar:1.2.1] 在 clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?] 在 java.lang.Thread.run(Thread.java:748) [?:1.8.0_171] 引起：java.lang.NullPointerException 在 org.apache.storm.kafka.spout.trident.KafkaTridentSpoutEmitter.seek(KafkaTridentSpoutEmitter.java:193) ~[stormjar.jar:?] 在 org.apache.storm.kafka.spout.trident.KafkaTridentSpoutEmitter.emitPartitionBatch（KafkaTridentSpoutEmitter.java:127） ~[stormjar.jar:?] 在 org.apache.storm.kafka.spout.trident.KafkaTridentSpoutEmitter.emitPartitionBatch(KafkaTridentSpoutEmitter.java:51) ~[stormjar.jar:?] 在 org.apache.storm.trident.spout.OpaquePartitionedTridentSpoutExecutor$Emitter.emitBatch(OpaquePartitionedTridentSpoutExecutor.java:141) ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.trident.spout.TridentSpoutExecutor.execute(TridentSpoutExecutor.java:82) ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.trident.topology.TridentBoltExecutor.execute（TridentBoltExecutor.java:383） ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.daemon.executor$fn__5043$tuple_action_fn__5045.invoke(executor.clj:739) ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.daemon.executor$mk_task_receiver$fn__4964.invoke(executor.clj:468) ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.disruptor$clojure_handler$reify__4475.onEvent(disruptor.clj:41) ~[storm-core-1.2.1.jar:1.2.1] 在 org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor（DisruptorQueue.java:509） ~[storm-core-1.2.1.jar:1.2.1] ... 6 更多
对此有何建议？
【问题讨论】：
-
能否请您发布整个堆栈跟踪？在您发布的内容下方可能有一个“原因”行，它会告诉您 NPE 的来源。
-
@StigRohdeDøssing 我已经更新了整个堆栈跟踪。谢谢
-
@StigRohdeDøssing 我想我的问题与这个groups.google.com/forum/#!topic/storm-user/wzdduOWul54 密切相关
-
您发布的代码 sn-p 与堆栈跟踪不匹配。代码 sn-p 正在使用storm-kafka Trident spout，而堆栈跟踪似乎正在使用storm-kafka-client Trident spout。你用的是哪个？
-
@StigRohdeDøssing 我正在使用 Storm Kafka 客户端。谢谢
标签： java nullpointerexception apache-kafka apache-storm trident