Thank you Srinath, Will check that out. Wondering if I need to write a custom serializer.
Thanks! Vinay On Mar 21, 2014 5:34 PM, "Srinath C" <[email protected]> wrote: > Vinay, > The exception is raised from *KryoTupleSerializer*, so one of the > values in your tuple directly or indirectly reference instances of > *ByteArrayLongString.* This is a class from the RabbitMQ client library. > One of the possibilities could be that you are adding all client > properties<http://grepcode.com/file/repo1.maven.org/maven2/com.rabbitmq/amqp-client/3.2.1/com/rabbitmq/client/impl/AMQConnection.java#75>into > the tuple. > > Regards, > Srinath. > > > > On Sat, Mar 22, 2014 at 4:56 AM, Vinay Pothnis <[email protected]>wrote: > >> Hello, >> >> This seems to happen when the spout and the bolt are on different JVMs >> (worker process) - which is understandable as there would be some sort of >> serialization for the inter jvm communication. >> >> But we are not able to find out what data this is and the logs do not say >> much. Have turned on "topology.debug = true" - but the logs dont contain >> any other useful information. >> >> Any idea how to troubleshoot this? >> >> Thanks! >> Vinay >> >> >> On Fri, Mar 21, 2014 at 3:07 PM, Vinay Pothnis >> <[email protected]>wrote: >> >>> Hello, >>> >>> I have a storm cluster of 3 nodes and a topology with 3 workers. I have >>> rabbit-mq from which I read messages and also post back to a queue. >>> >>> I seem to get this exception when I am using more than 1 worker. When I >>> have just 1 worker, I do not get this exception. But when i change it to 3 >>> workers, this exception arises. >>> >>> *java.lang.RuntimeException: java.lang.RuntimeException: >>> java.io.NotSerializableException: >>> com.rabbitmq.client.impl.LongStringHelper$ByteArrayLongString* >>> * at >>> backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:90)* >>> * at >>> backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:61)* >>> * at >>> backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:62)* >>> * at >>> backtype.storm.disruptor$consume_loop_STAR_$fn__2975.invoke(disruptor.clj:74)* >>> * at backtype.storm.util$async_loop$fn__444.invoke(util.clj:403)* >>> * at clojure.lang.AFn.run(AFn.java:24)* >>> * at java.lang.Thread.run(Thread.java:744)* >>> *Caused by: java.lang.RuntimeException: >>> java.io.NotSerializableException: >>> com.rabbitmq.client.impl.LongStringHelper$ByteArrayLongString* >>> * at >>> backtype.storm.serialization.SerializableSerializer.write(SerializableSerializer.java:24)* >>> * at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:554)* >>> * at >>> com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:77)* >>> * at >>> com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:18)* >>> * at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:472)* >>> * at >>> backtype.storm.serialization.KryoValuesSerializer.serializeInto(KryoValuesSerializer.java:27)* >>> * at >>> backtype.storm.serialization.KryoTupleSerializer.serialize(KryoTupleSerializer.java:27)* >>> * at >>> backtype.storm.daemon.worker$mk_transfer_fn$fn__5686$fn__5690.invoke(worker.clj:108)* >>> * at backtype.storm.util$fast_list_map.invoke(util.clj:801)* >>> * at >>> backtype.storm.daemon.worker$mk_transfer_fn$fn__5686.invoke(worker.clj:108)* >>> * at >>> backtype.storm.daemon.executor$start_batch_transfer__GT_worker_handler_BANG_$fn__3328.invoke(executor.clj:240)* >>> * at >>> backtype.storm.disruptor$clojure_handler$reify__2962.onEvent(disruptor.clj:43)* >>> * at >>> backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:87)* >>> * ... 6 more* >>> *Caused by: java.io.NotSerializableException: >>> com.rabbitmq.client.impl.LongStringHelper$ByteArrayLongString* >>> * at >>> java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1183)* >>> * at >>> java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347)* >>> * at java.util.HashMap.writeObject(HashMap.java:1133)* >>> * at sun.reflect.GeneratedMethodAccessor44.invoke(Unknown Source)* >>> * at >>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)* >>> * at java.lang.reflect.Method.invoke(Method.java:606)* >>> * at >>> java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:988)* >>> * at >>> java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1495)* >>> * at >>> java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)* >>> * at >>> java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)* >>> * at >>> java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)* >>> * at >>> java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)* >>> * at >>> java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)* >>> * at >>> java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)* >>> * at >>> java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347)* >>> * at >>> backtype.storm.serialization.SerializableSerializer.write(SerializableSerializer.java:21)* >>> >>> >>> Any pointers on what could be the reason and how to troubleshoot this? >>> >>> Thanks! >>> Vinay >>> >>> >>> >> >
