Sachin Pasalkar created STORM-1006:
--------------------------------------
Summary: A big memory hit
Key: STORM-1006
URL: https://issues.apache.org/jira/browse/STORM-1006
Project: Apache Storm
Issue Type: Improvement
Components: storm-kafka
Affects Versions: 0.9.3
Environment: Mac
Reporter: Sachin Pasalkar
We are reading whole file in memory around 5 MB, which is send through Kafaka
to Storm. In next bolt, we performs the operation on file and sends out tuple
to next bolt. After profiling we found that file (bytes of file) does not get
garbage collected. So after further investigation we found that
backtype.storm.coordination.CoordinatedBolt.CoordinatedOutputCollector.emit(String,
Collection<Tuple>, List<Object>) API gets the first object and use it for
tracking :(. Can you confirm reason behind this? Is there any way we can send
different unique id as first element in list or the unique id of tuple used as
indicator.
However, for time being we have made changes in schema assigned to KafkaSpout,
so that it will parse the file and send out list of values.
If you below code CoordinatedBolt, "Object id = tuple.getValue(0);” takes the
1st element from tuple instead of taking id of tuple. This "id" is then saved
to _tracked hashhMap(TimeCache). In our case the 0th element is files byte
data. This gets stored in the _tracked map till tree of tuple doesn’t get
complete. As we are processing huge data we run outofMemory issue.
Code:
public void execute(Tuple tuple) {
Object id = tuple.getValue(0);
TrackingInfo track;
TupleType type = getTupleType(tuple);
synchronized(_tracked) {
track = _tracked.get(id);
if(track==null) {
track = new TrackingInfo();
if(_idStreamSpec==null) track.receivedId = true;
_tracked.put(id, track);
}
}
if(type==TupleType.ID) {
synchronized(_tracked) {
track.receivedId = true;
}
checkFinishId(tuple, type);
} else if(type==TupleType.COORD) {
int count = (Integer) tuple.getValue(1);
synchronized(_tracked) {
track.reportCount++;
track.expectedTupleCount+=count;
}
checkFinishId(tuple, type);
} else {
synchronized(_tracked) {
_delegate.execute(tuple);
}
}
}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
