nathanmarz/storm-starter — Issue #5: java.util.ConcurrentModificationException from storm.starter.bolt.MergeObjects
Author: rickdu
Created: 2012-01-10 01:48:28+00:00
State: open

When I was learning storm.starter.RollingTopWords.java, I changed the spout to generate 1 million random words continuously without sleeping, so the code will be like this:

public class PerformanceTestingTopology {

public static void main(String[] args) throws Exception {

final int TOP_N = 20;

TopologyBuilder builder = new TopologyBuilder();

LocalCluster cluster = new LocalCluster();

builder.setSpout("word", new RandomWordSpout(1000000), 1); // my spout, I just copy from TestWordSpout, add counter and remove the sleeping

builder.setBolt("count", new RollingCountObjects(60, 10), 5)
.fieldsGrouping("word", new Fields("word"));
builder.setBolt("rank", new RankObjects(TOP_N), 5)
.fieldsGrouping("count", new Fields("obj"));
builder.setBolt("merge", new MergeObjects(TOP_N))
.globalGrouping("rank");
Config conf = new Config();
cluster.submitTopology("rolling-demo", conf, builder.createTopology());
}
}

Then I got errors like this:

5511 [Thread-53] ERROR backtype.storm.util - Async loop died!
java.util.ConcurrentModificationException
at java.util.AbstractList$Itr.checkForComodification(AbstractList.java:372)
at java.util.AbstractList$Itr.next(AbstractList.java:343)
at com.esotericsoftware.kryo.serialize.CollectionSerializer.writeObjectData(CollectionSerializer.java:85)
at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:489)
at com.esotericsoftware.kryo.serialize.CollectionSerializer.writeObjectData(CollectionSerializer.java:86)
at com.esotericsoftware.kryo.Serializer.writeObject(Serializer.java:43)
at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:520)
at com.esotericsoftware.kryo.ObjectBuffer.writeObject(ObjectBuffer.java:159)
at backtype.storm.serialization.KryoValuesSerializer.serializeInto(KryoValuesSerializer.java:25)
at backtype.storm.serialization.KryoTupleSerializer.serialize(KryoTupleSerializer.java:32)
at backtype.storm.daemon.worker$fn__3181$exec_fn__945__auto____3182$fn__3242$fn__3243.invoke(worker.clj:200)
at backtype.storm.daemon.worker$fn__3181$exec_fn__945__auto____3182$fn__3242.invoke(worker.clj:195)
at clojure.lang.AFn.applyToHelper(AFn.java:165)
at clojure.lang.AFn.applyTo(AFn.java:151)
at clojure.core$apply.invoke(core.clj:540)
at backtype.storm.util$async_loop$fn__451.invoke(util.clj:226)
at clojure.lang.AFn.run(AFn.java:24)
at java.lang.Thread.run(Thread.java:619)
5518 [Thread-53] INFO backtype.storm.util - Halting process: ("Async loop died!")

I found that it's caused by MergeObjects. After I changed
private List< List> _rankings = new ArrayList();
to:
private List< String> _rankings = new ArrayList();
and modified the comparator, it works.

As a reference, my merging bolt is:

package storm.starter.bolt;

import org.apache.log4j.Logger;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MergeRankings implements IBasicBolt {
public static Logger LOG = Logger.getLogger(MergeRankings.class);

private List _rankings = new ArrayList();
int _count = 10;
Long _lastTime;

public MergeRankings(int n) {
_count = n;
}

private int _compare(String one, String two) {
long valueOne = Long.parseLong(one.substring(one.indexOf(" ") + 1));
long valueTwo = Long.parseLong(two.substring(two.indexOf(" ") + 1));
long delta = valueTwo - valueOne;
if(delta > 0) {
return 1;
} else if (delta < 0) {
return -1;
} else {
return 0;
}
}

private Integer _find(Object tag) {
for(int i = 0; i < _rankings.size(); ++i) {
Object cur = _rankings.get(i).substring(0, _rankings.get(i).indexOf(" "));
if (cur.equals(tag)) {
return i;
}
}
return null;
}

public void prepare(Map stormConf, TopologyContext context) {

}

public void execute(Tuple tuple, BasicOutputCollector collector) {
List

merging = (List) tuple.getValue(0);
for(List pair : merging) {
Integer existingIndex = _find(pair.get(0));
if (null != existingIndex) {
_rankings.set(existingIndex, pair.get(0) + " " + pair.get(1));
} else {
_rankings.add(pair.get(0) + " " + pair.get(1));
}

Collections.sort(_rankings, new Comparator() {
public int compare(String o1, String o2) {
return _compare(o1, o2);
}
});

if (_rankings.size() > _count) {
_rankings.subList(_count, _rankings.size()).clear();
}
}

long currentTime = System.currentTimeMillis();
if(_lastTime==null || currentTime >= _lastTime + 2000) {
collector.emit(new Values(_rankings));
LOG.info("Rankings: " + _rankings);
_lastTime = currentTime;
}
}

public void cleanup() {
}

public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("list"));
}
}