kryo serializer throws java.util.ConcurrentModificationException

[IvyTang <iv...@gmail.com>]

Hi all，

    My storm project is for calculating pv and uv . So i write two classes  .

    class UidCountState implements java.io.Serializable 
       private Set<String> dimens = new ConcurrentSkipListSet<String>();
       private long count = 0;

   The ConcurrentSkipListSet is for the uv ,and count for pv.



   class RedisState implements IBackingMap<UidCountState>
   This class is to store UidCountState into Redis.

    Because i need to put UidCountState  into Redis, so in UidCountState i have two methods serialize and deserialize.  The serialize method put every fields of UidCountState into a byte[] , of course it needs iterate the Set.
    When multiPut, serialize  UidCountState and put the bytes to redis.
    When multiGet, get the bytes and deserialize it.

    At the beginning,the set in UidCountState is HasSet, and storm throws:

    java.lang.RuntimeException: java.util.ConcurrentModificationException
	at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:87)
	at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:58)
	at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:62)
	at backtype.storm.disruptor$consume_loop_STAR_$fn__1619.invoke(disruptor.clj:73)
	at backtype.storm.util$async_loop$fn__465.invoke(util.clj:377)
	at clojure.lang.AFn.run(AFn.java:24)
	at java.lang.Thread.run(Thread.java:662)
Caused by: java.util.ConcurrentModificationException
	at java.util.HashMap$HashIterator.nextEntry(HashMap.java:793)
	at java.util.HashMap$KeyIterator.next(HashMap.java:828)
	at java.util.HashSet.writeObject(HashSet.java:267)
	at sun.reflect.GeneratedMethodAccessor30.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
	at java.lang.reflect.Method.invoke(Method.java:597)
	at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:940)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1469)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1400)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1158)
	at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1518)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1483)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1400)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1158)
	at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:330)
	at backtype.storm.serialization.SerializableSerializer.write(SerializableSerializer.java:21)
	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__4152$fn__4156.invoke(worker.clj:99)
	at backtype.storm.util$fast_list_map.invoke(util.clj:771)
	at backtype.storm.daemon.worker$mk_transfer_fn$fn__4152.invoke(worker.clj:99)
	at backtype.storm.daemon.executor$start_batch_transfer__GT_worker_handler_BANG_$fn__3914.invoke(executor.clj:238)
	at backtype.storm.disruptor$clojure_handler$reify__1606.onEvent(disruptor.clj:43)
	at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:84)
	... 6 more

    I thought it was because  the HashSet is not thread-safe,so i changed it to ConcurrentSkipListSet. But the exception still exists.

    Need i custom serialize the UidCountState ?
     
-- 
Best regards,

Ivy Tang