You are viewing a plain text version of this content. The canonical link for it is here.

Posted to user@storm.apache.org by "Mitchell Rathbun (BLOOMBERG/ 731 LEX)" <mr...@bloomberg.net> on 2019/05/10 20:29:02 UTC

ClosedSelectorException on Topology shutdown

We frequently get ClosedSelectorException from our KafkaSpout implementation on topology shutdown. We are using storm-kafka-client version 1.2.1 and kafka-clients version 0.10.2.0. The reason for the older version of kafka-clients is https://issues.apache.org/jira/browse/STORM-3102. We want to look into building storm-kafka-client 1.2.3 locally until it is released, but would this prevent the ClosedSelectorException from occurring? Or is there a different cause for this exception? The exception trace is:

2019-05-04 11:34:38,420 ERROR executor [Thread-72-customKafkaSpout-executor[19 19]] 
java.lang.RuntimeException: java.nio.channels.ClosedSelectorException
        at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:522) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:487) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.utils.DisruptorQueue.consumeBatch(DisruptorQueue.java:477) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.disruptor$consume_batch.invoke(disruptor.clj:70) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.daemon.executor$fn__4975$fn__4990$fn__5021.invoke(executor.clj:634) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:484) ~[storm-core-1.2.1.jar:1.2.1]
        at clojure.lang.AFn.run(AFn.java:22) ~[clojure-1.7.0.jar:?]
        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_172]
Caused by: java.nio.channels.ClosedSelectorException
        at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:83) ~[?:1.8.0_172]
        at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97) ~[?:1.8.0_172]
        at org.apache.kafka.common.network.Selector.select(Selector.java:489) ~[stormjar.jar:?]
        at org.apache.kafka.common.network.Selector.poll(Selector.java:298) ~[stormjar.jar:?]
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:349) ~[stormjar.jar:?]
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:226) ~[stormjar.jar:?]
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:188) ~[stormjar.jar:?]
        at org.apache.kafka.clients.consumer.internals.Fetcher.retrieveOffsetsByTimes(Fetcher.java:405) ~[stormjar.jar:?]
        at org.apache.kafka.clients.consumer.internals.Fetcher.beginningOrEndOffset(Fetcher.java:448) ~[stormjar.jar:?]
        at org.apache.kafka.clients.consumer.internals.Fetcher.beginningOffsets(Fetcher.java:433) ~[stormjar.jar:?]
        at org.apache.kafka.clients.consumer.KafkaConsumer.beginningOffsets(KafkaConsumer.java:1473) ~[stormjar.jar:?]
        at org.apache.storm.kafka.spout.metrics.KafkaOffsetMetric.getValueAndReset(KafkaOffsetMetric.java:79) ~[stormjar.jar:?]
        at org.apache.storm.daemon.executor$metrics_tick$fn__4899.invoke(executor.clj:345) ~[storm-core-1.2.1.jar:1.2.1]
        at clojure.core$map$fn__4553.invoke(core.clj:2622) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.sval(LazySeq.java:40) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.seq(LazySeq.java:49) ~[clojure-1.7.0.jar:?]
        at clojure.lang.RT.seq(RT.java:507) ~[clojure-1.7.0.jar:?]
        at clojure.core$seq__4128.invoke(core.clj:137) ~[clojure-1.7.0.jar:?]
        at clojure.core$filter$fn__4580.invoke(core.clj:2679) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.sval(LazySeq.java:40) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.seq(LazySeq.java:49) ~[clojure-1.7.0.jar:?]
        at clojure.lang.Cons.next(Cons.java:39) ~[clojure-1.7.0.jar:?]
        at clojure.lang.RT.next(RT.java:674) ~[clojure-1.7.0.jar:?]
        at clojure.core$next__4112.invoke(core.clj:64) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$fn__6523.invoke(protocols.clj:170) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$fn__6478$G__6473__6487.invoke(protocols.clj:19) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$seq_reduce.invoke(protocols.clj:31) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$fn__6506.invoke(protocols.clj:101) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$fn__6452$G__6447__6465.invoke(protocols.clj:13) ~[clojure-1.7.0.jar:?]
        at clojure.core$reduce.invoke(core.clj:6519) ~[clojure-1.7.0.jar:?]
        at clojure.core$into.invoke(core.clj:6600) ~[clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.executor$metrics_tick.invoke(executor.clj:349) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.daemon.executor$fn__4975$tuple_action_fn__4981.invoke(executor.clj:522) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.daemon.executor$mk_task_receiver$fn__4964.invoke(executor.clj:471) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.disruptor$clojure_handler$reify__4475.onEvent(disruptor.clj:41) ~[storm-core-1.2.1.jar:1.2.1]
        at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:509) ~[storm-core-1.2.1.jar:1.2.1]
        ... 7 more

Re: ClosedSelectorException on Topology shutdown

Posted by Stig Rohde Døssing <st...@gmail.com>.

I would just try it. If it's still an issue after the upgrade, we can look
at it.

Den fre. 10. maj 2019 kl. 22.29 skrev Mitchell Rathbun (BLOOMBERG/ 731 LEX)
<mr...@bloomberg.net>:

> We frequently get ClosedSelectorException from our KafkaSpout
> implementation on topology shutdown. We are using storm-kafka-client
> version 1.2.1 and kafka-clients version 0.10.2.0. The reason for the older
> version of kafka-clients is
> https://issues.apache.org/jira/browse/STORM-3102. We want to look into
> building storm-kafka-client 1.2.3 locally until it is released, but would
> this prevent the ClosedSelectorException from occurring? Or is there a
> different cause for this exception? The exception trace is:
>
> 2019-05-04 11:34:38,420 ERROR executor
> [Thread-72-customKafkaSpout-executor[19 19]]
> java.lang.RuntimeException: java.nio.channels.ClosedSelectorException
> at
> org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:522)
> ~[storm-core-1.2.1.jar:1.2.1]
> at
> org.apache.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:487)
> ~[storm-core-1.2.1.jar:1.2.1]
> at
> org.apache.storm.utils.DisruptorQueue.consumeBatch(DisruptorQueue.java:477)
> ~[storm-core-1.2.1.jar:1.2.1]
> at org.apache.storm.disruptor$consume_batch.invoke(disruptor.clj:70)
> ~[storm-core-1.2.1.jar:1.2.1]
> at
> org.apache.storm.daemon.executor$fn__4975$fn__4990$fn__5021.invoke(executor.clj:634)
> ~[storm-core-1.2.1.jar:1.2.1]
> at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:484)
> ~[storm-core-1.2.1.jar:1.2.1]
> at clojure.lang.AFn.run(AFn.java:22) ~[clojure-1.7.0.jar:?]
> at java.lang.Thread.run(Thread.java:748) [?:1.8.0_172]
> Caused by: java.nio.channels.ClosedSelectorException
> at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:83)
> ~[?:1.8.0_172]
> at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97) ~[?:1.8.0_172]
> at org.apache.kafka.common.network.Selector.select(Selector.java:489)
> ~[stormjar.jar:?]
> at org.apache.kafka.common.network.Selector.poll(Selector.java:298)
> ~[stormjar.jar:?]
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:349)
> ~[stormjar.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:226)
> ~[stormjar.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:188)
> ~[stormjar.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.Fetcher.retrieveOffsetsByTimes(Fetcher.java:405)
> ~[stormjar.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.Fetcher.beginningOrEndOffset(Fetcher.java:448)
> ~[stormjar.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.Fetcher.beginningOffsets(Fetcher.java:433)
> ~[stormjar.jar:?]
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.beginningOffsets(KafkaConsumer.java:1473)
> ~[stormjar.jar:?]
> at
> org.apache.storm.kafka.spout.metrics.KafkaOffsetMetric.getValueAndReset(KafkaOffsetMetric.java:79)
> ~[stormjar.jar:?]
> at
> org.apache.storm.daemon.executor$metrics_tick$fn__4899.invoke(executor.clj:345)
> ~[storm-core-1.2.1.jar:1.2.1]
> at clojure.core$map$fn__4553.invoke(core.clj:2622) ~[clojure-1.7.0.jar:?]
> at clojure.lang.LazySeq.sval(LazySeq.java:40) ~[clojure-1.7.0.jar:?]
> at clojure.lang.LazySeq.seq(LazySeq.java:49) ~[clojure-1.7.0.jar:?]
> at clojure.lang.RT.seq(RT.java:507) ~[clojure-1.7.0.jar:?]
> at clojure.core$seq__4128.invoke(core.clj:137) ~[clojure-1.7.0.jar:?]
> at clojure.core$filter$fn__4580.invoke(core.clj:2679)
> ~[clojure-1.7.0.jar:?]
> at clojure.lang.LazySeq.sval(LazySeq.java:40) ~[clojure-1.7.0.jar:?]
> at clojure.lang.LazySeq.seq(LazySeq.java:49) ~[clojure-1.7.0.jar:?]
> at clojure.lang.Cons.next(Cons.java:39) ~[clojure-1.7.0.jar:?]
> at clojure.lang.RT.next(RT.java:674) ~[clojure-1.7.0.jar:?]
> at clojure.core$next__4112.invoke(core.clj:64) ~[clojure-1.7.0.jar:?]
> at clojure.core.protocols$fn__6523.invoke(protocols.clj:170)
> ~[clojure-1.7.0.jar:?]
> at clojure.core.protocols$fn__6478$G__6473__6487.invoke(protocols.clj:19)
> ~[clojure-1.7.0.jar:?]
> at clojure.core.protocols$seq_reduce.invoke(protocols.clj:31)
> ~[clojure-1.7.0.jar:?]
> at clojure.core.protocols$fn__6506.invoke(protocols.clj:101)
> ~[clojure-1.7.0.jar:?]
> at clojure.core.protocols$fn__6452$G__6447__6465.invoke(protocols.clj:13)
> ~[clojure-1.7.0.jar:?]
> at clojure.core$reduce.invoke(core.clj:6519) ~[clojure-1.7.0.jar:?]
> at clojure.core$into.invoke(core.clj:6600) ~[clojure-1.7.0.jar:?]
> at org.apache.storm.daemon.executor$metrics_tick.invoke(executor.clj:349)
> ~[storm-core-1.2.1.jar:1.2.1]
> at
> org.apache.storm.daemon.executor$fn__4975$tuple_action_fn__4981.invoke(executor.clj:522)
> ~[storm-core-1.2.1.jar:1.2.1]
> at
> org.apache.storm.daemon.executor$mk_task_receiver$fn__4964.invoke(executor.clj:471)
> ~[storm-core-1.2.1.jar:1.2.1]
> at
> org.apache.storm.disruptor$clojure_handler$reify__4475.onEvent(disruptor.clj:41)
> ~[storm-core-1.2.1.jar:1.2.1]
> at
> org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:509)
> ~[storm-core-1.2.1.jar:1.2.1]
> ... 7 more
>