Storm-Kafka喷口失效

Question

我使用storm0.9.4和storm-kafka:0.9.0-wip16a-scala292作为从kafka0.7读取的依赖项。

• 我们的Kafka保留政策是7天。
• 我从经纪人的最新偏移量开始读取。

kafka.common.OffsetOutOfRangeException: null
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[na:1.7.0_75]
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57) ~[na:1.7.0_75]
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[na:1.7.0_75]
        at java.lang.reflect.Constructor.newInstance(Constructor.java:526) ~[na:1.7.0_75]
        at java.lang.Class.newInstance(Class.java:379) ~[na:1.7.0_75]
        at kafka.common.ErrorMapping$.maybeThrowException(ErrorMapping.scala:53) ~[stormjar.jar:na]
        at kafka.message.ByteBufferMessageSet.kafka$message$ByteBufferMessageSet$$internalIterator(ByteBufferMessageSet.scala:99) ~[stormjar.jar:na]
        at kafka.message.ByteBufferMessageSet.iterator(ByteBufferMessageSet.scala:82) ~[stormjar.jar:na]
        at scala.collection.IterableLike$class.foreach(IterableLike.scala:73) ~[stormjar.jar:na]
        at kafka.message.MessageSet.foreach(MessageSet.scala:87) ~[stormjar.jar:na]
        at scala.collection.TraversableOnce$class.size(TraversableOnce.scala:104) ~[stormjar.jar:na]
        at kafka.message.MessageSet.size(MessageSet.scala:87) ~[stormjar.jar:na]
        at storm.kafka.PartitionManager.fill(PartitionManager.java:113) ~[stormjar.jar:na]
        at storm.kafka.PartitionManager.next(PartitionManager.java:83) ~[stormjar.jar:na]
        at storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:106) ~[stormjar.jar:na]
        at backtype.storm.daemon.executor$fn__4654$fn__4669$fn__4698.invoke(executor.clj:565) ~[storm-core-0.9.4.jar:0.9.4]
        at backtype.storm.util$async_loop$fn__458.invoke(util.clj:463) ~[storm-core-0.9.4.jar:0.9.4]
        at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
        at java.lang.Thread.run(Thread.java:745) [na:1.7.0_75] 2015-04-30T01:49:15.118-0500 backtype.storm.daemon.executor [ERROR] kafka.common.OffsetOutOfRangeException: null
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[na:1.7.0_75]
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57) ~[na:1.7.0_75]
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[na:1.7.0_75]
        at java.lang.reflect.Constructor.newInstance(Constructor.java:526) ~[na:1.7.0_75]
        at java.lang.Class.newInstance(Class.java:379) ~[na:1.7.0_75]
        at kafka.common.ErrorMapping$.maybeThrowException(ErrorMapping.scala:53) ~[stormjar.jar:na]
        at kafka.message.ByteBufferMessageSet.kafka$message$ByteBufferMessageSet$$internalIterator(ByteBufferMessageSet.scala:99) ~[stormjar.jar:na]
        at kafka.message.ByteBufferMessageSet.iterator(ByteBufferMessageSet.scala:82) ~[stormjar.jar:na]
        at scala.collection.IterableLike$class.foreach(IterableLike.scala:73) ~[stormjar.jar:na]
        at kafka.message.MessageSet.foreach(MessageSet.scala:87) ~[stormjar.jar:na]
        at scala.collection.TraversableOnce$class.size(TraversableOnce.scala:104) ~[stormjar.jar:na]
        at kafka.message.MessageSet.size(MessageSet.scala:87) ~[stormjar.jar:na]
        at storm.kafka.PartitionManager.fill(PartitionManager.java:113) ~[stormjar.jar:na]
        at storm.kafka.PartitionManager.next(PartitionManager.java:83) ~[stormjar.jar:na]
        at storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:106) ~[stormjar.jar:na]
        at backtype.storm.daemon.executor$fn__4654$fn__4669$fn__4698.invoke(executor.clj:565) ~[storm-core-0.9.4.jar:0.9.4]
        at backtype.storm.util$async_loop$fn__458.invoke(util.clj:463) ~[storm-core-0.9.4.jar:0.9.4]
        at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
        at java.lang.Thread.run(Thread.java:745) [na:1.7.0_75] 2015-04-30T01:49:15.129-0500 backtype.storm.util [ERROR] Halting process: ("Worker died") java.lang.RuntimeException: ("Worker died")
        at backtype.storm.util$exit_process_BANG_.doInvoke(util.clj:325) [storm-core-0.9.4.jar:0.9.4]
        at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.5.1.jar:na]
        at backtype.storm.daemon.worker$fn__5102$fn__5103.invoke(worker.clj:495) [storm-core-0.9.4.jar:0.9.4]
        at backtype.storm.daemon.executor$mk_executor_data$fn__4555$fn__4556.invoke(executor.clj:240) [storm-core-0.9.4.jar:0.9.4]
        at backtype.storm.util$async_loop$fn__458.invoke(util.clj:473) [storm-core-0.9.4.jar:0.9.4]
        at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
        at java.lang.Thread.run(Thread.java:745) [na:1.7.0_75]

Answer 1

喷口的配置方式有问题。我们有用于初始化SpoutConfig对象的自定义属性，我们始终将forceFromStart和startOffsetTime（设置为latest）。问题是，与后者相关的属性配置了错误的键，因此有时一个spout的zk条目具有一个在kafka中不再存在的较早的偏移量条目，或者引用了storm拓扑启动时存在的条目，但在storm完成积压之前从kafka中删除了该条目。由于我们无论如何都不想满足这个场景，我们只是修正了配置及其工作。