having issues with BlockingQueue java , implementation of Storm (Distributed computing)?
-
19-06-2021 - |
문제
This is the code snippt of my input spout for emmiting tuple to a processing noded for stream processing over a cluster. The problem is The BlockingQueue is throwing InterruptedException .
private SpoutOutputCollector collector;
public BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<String>();
public boolean isDistributed() {
return true;
}
public void open(@SuppressWarnings("rawtypes") final Map conf, final TopologyContext context,
final SpoutOutputCollector collector) {
this.collector=collector;
}
@Override
public void nextTuple() {
try {
//Utils.sleep(100);
collector.emit(new Values("Single Temperature Reading", blockingQueue.take()));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void readInputfile() throws IOException, InterruptedException{
FileInputStream file = new FileInputStream("/home/529076/Desktop/Temperature");
DataInputStream readDate=new DataInputStream(file);
BufferedReader readText=new BufferedReader(new InputStreamReader(readDate));
String line;
String singleReading = null;
while((line=readText.readLine())!=null){
singleReading=line;
blockingQueue.add(singleReading);
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("Single Temperature Reading"));
}
The exception description is as followes :---
java.lang.InterruptedException10930 [Thread-20] INFO backtype.storm.util - Async loop interrupted!
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.reportInterruptAfterWait(AbstractQueuedSynchronizer.java:1961)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1996)
at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:399)
at com.tcs.storm.test.InputStreamSpout.nextTuple(InputStreamSpout.java:65)
at backtype.storm.daemon.task$fn__3349$fn__3404.invoke(task.clj:413)
And The nextTuple(InputStreamSpout.java:65 is ------>
collector.emit(new Values("Single Temperature Reading", blockingQueue.take()));
Thanks
해결책
The error is due to the fact -- The BlockingQueue was not initilaized in the output collector;
다른 팁
In Storm, it is recommendable to initialize fields of Bolts/Spouts in the prepare()/open() methods.
The reason for this is that Bolts/Spouts are created on the node from which you upload the topology: your BlockingQueue would be constructed there. Bolts/Spouts are then then serialized and distributed to the worker nodes, where they are deserialized. It's possible that this process of serializing/deserializing fails to preserve all the properties of fields that are set at construction of the Bolt/Spout. Fields initialized in prepare() or open() do not have this issue.