You should do like this in your bolt execute method
if (tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) &&
tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID)) {
// Processing for tick tuple
} else {
// Processing for non tick tuple
}
import backtype.storm.Constants;
On 1/3/14, Shane <shane.lisy@gmail.com> wrote:
> Hi,
> I found TransactionalSpoutBatchExecutor source code like this:
> @Override
> public void execute(Tuple input) {
> TransactionAttempt attempt = (TransactionAttempt)
> input.getValue(0);
> ......
> }
>
> so if I use tick tuple frequently to do someting , I get the error:
>
> 12139 [Thread-23-userflow_spout] ERROR backtype.storm.daemon.executor -
> java.lang.RuntimeException: java.lang.ClassCastException: java.lang.Long
> cannot be cast to backtype.storm.transactional.TransactionAttempt
> at
> backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:90)
> ~[storm-core-0.9.0.1.jar:na]
> at
> backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:61)
> ~[storm-core-0.9.0.1.jar:na]
> at
> backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:62)
> ~[storm-core-0.9.0.1.jar:na]
> at
> backtype.storm.daemon.executor$fn__3498$fn__3510$fn__3557.invoke(executor.clj:730)
> ~[storm-core-0.9.0.1.jar:na]
> at backtype.storm.util$async_loop$fn__444.invoke(util.clj:403)
> ~[storm-core-0.9.0.1.jar:na]
> at clojure.lang.AFn.run(AFn.java:24) [clojure-1.4.0.jar:na]
> at java.lang.Thread.run(Thread.java:662) [na:1.6.0_45]
> --
> -----------------------------------------------------------------------
> Yours sincerely,
> Shane
>
|