nathanmarz/storm — Issue #70: util [ERROR] Async loop died! what is wrong !!!
Author: wangweichen
Created: 2011-12-13 06:52:01+00:00
State: open

i run a simple spout object ,but i meet a trouble:
spout code follow :

package com.taobao.search.spout.log;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.Map;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
import com.taobao.search.ttclient.*;
import com.taobao.timetunnel.client.Message;
import com.taobao.timetunnel.client.ReadPosition;

public class atpanelSpout implements IRichSpout {

private SpoutOutputCollector _collector;
//private searchlog atpanellog;
//private SimpleDateFormat format;
//private ReadPosition r ;
//private long timestamp;
//private String searchlogname;
//private String router;
//private String user;
//private String pwd;
//private String timeString;

public atpanelSpout(){

}

public void open(Map conf,TopologyContext context, SpoutOutputCollector collector){

this._collector= collector;

}

public boolean isDistributed(){
return true;
}

public void nextTuple(){
Utils.sleep(100);
//this.r = this.atpanellog.getReadPosition(1, this.timestamp);

//	long timestamp;
try {
searchlog atpanelLog = new searchlog("atpanel.atpanel", "ipaddress:9090",
30000, 16, "ads", "ads");
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
long timestamp = format.parse("2011-12-05 10:00:00").getTime();
atpanelLog.SubClientInit();
ReadPosition r = atpanelLog.getReadPosition(1, timestamp);

while(true){
Iterator itr = atpanelLog.getDigital(1, timestamp, 10, r);
while ((itr != null) && (itr.hasNext())) {
Message msg = (Message)itr.next();
//System.out.print(new String(msg.getData()));
_collector.emit(new Values(new String(msg.getData())));
}
}

} catch (ParseException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

//ReadPosition r = atpanelLog.getReadPosition(1, timestamp);
//this.r = atpanellog.getReadPosition(1, timestamp);
//while (true) {

//Iterator itr = atpanellog.getDigital(1, timestamp, 10, r);
//	while ((itr != null) && (itr.hasNext())) {
// Message msg = (Message)itr.next();
// System.out.print(new String(msg.getData()));
// _collector.emit(new Values(msg.getData()));
// }
//	}
//while(true){
//	_collector.emit(new Values("wangqian\n"));
//}

}

public void ack(){}

public void close(){}

public void fail(){}

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

public void ack(Object msgId) {
// TODO Auto-generated method stub

}

public void fail(Object msgId) {
// TODO Auto-generated method stub

}

}

the function nextTuple have a problem:
log follow:
2011-12-12 18:12:16 util [ERROR] Async loop died!
java.lang.NoSuchMethodError: org.apache.thrift.meta_data.FieldValueMetaData.(BZ)V
at com.taobao.timetunnel.thrift.broker.BrokerGetService$getOffsetByTime_result.(BrokerGetService.java:9346)
at com.taobao.timetunnel.thrift.broker.BrokerGetService$Client.recv_getOffsetByTime(BrokerGetService.java:436)
at com.taobao.timetunnel.thrift.broker.BrokerGetService$Client.getOffsetByTime(BrokerGetService.java:409)
at com.taobao.timetunnel.client.impl.DefaultTimeTunnelSubClient$4.execute(DefaultTimeTunnelSubClient.java:221)
at com.taobao.timetunnel.client.impl.DefaultTimeTunnelSubClient$4.execute(DefaultTimeTunnelSubClient.java:218)
at com.taobao.timetunnel.client.impl.BaseTimeTunnelClient.execute(BaseTimeTunnelClient.java:229)
at com.taobao.timetunnel.client.impl.BaseTimeTunnelClient.execute(BaseTimeTunnelClient.java:211)
at com.taobao.timetunnel.client.impl.DefaultTimeTunnelSubClient.getOffset(DefaultTimeTunnelSubClient.java:218)
at com.taobao.search.ttclient.searchlog.getReadPosition(searchlog.java:41)
at com.taobao.search.spout.log.atpanelSpout.nextTuple(atpanelSpout.java:85)
at backtype.storm.daemon.task$fn__2957$fn__2972.invoke(task.clj:383)
at backtype.storm.daemon.task$mk_task$iter__2910__2914$fn__2915$fn__2916$fn__2917.invoke(task.clj:281)
at clojure.lang.AFn.applyToHelper(AFn.java:159)
at clojure.lang.AFn.applyTo(AFn.java:151)
at clojure.core$apply.invoke(core.clj:540)
at backtype.storm.util$async_loop$fn__443.invoke(util.clj:213)
at clojure.lang.AFn.run(AFn.java:24)
at java.lang.Thread.run(Thread.java:619)
2011-12-12 18:12:16 util [INFO] Halting process: ("Task died")

but when i create a new project run code :
package com.taobao.search.ttclient;

import java.util.Iterator;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

import backtype.storm.tuple.Values;

import com.taobao.timetunnel.client.Authentication;
import com.taobao.timetunnel.client.Message;
import com.taobao.timetunnel.client.ReadPosition;
import com.taobao.timetunnel.client.Tunnel;
import com.taobao.timetunnel.client.impl.DefaultTimeTunnelSubClient;
import com.taobao.timetunnel.thrift.common.Partition;

public class searchlog {

private Tunnel ttclient;
private Authentication auth;
private DefaultTimeTunnelSubClient sub;

public searchlog(String name, String router, int timout, int partnum,
String username, String pwd) {
ttclient = new Tunnel(name);
ttclient.setRouter(router);
ttclient.setRouterTimeout(timout);
ttclient.setPartitions(partnum);
auth = new Authentication();
auth.setUsername(username);
auth.setPassword(pwd);
sub = new DefaultTimeTunnelSubClient(ttclient);
sub.setAuthentication(auth);
}

public void SubClientInit() {
sub.init();
}

public ReadPosition getReadPosition(int partnum, long timestamp) {
return sub.getOffset(new Partition((short) partnum), timestamp,
ttclient.getRouterTimeout());
}

public Iterator getDigital(int partitionNum, long timestamp,
int rownum, ReadPosition position) {
// ReadPosition position = sub.getOffset(new
// Partition((short)partitionNum), timestamp,
// ttclient.getRouterTimeout());
// Iterator returnMsg = new Iterator();
if (position != null) {
Iterator itr = sub.iterator(new Partition(
(short) partitionNum), position, rownum, ttclient
.getRouterTimeout());
return itr;

} else {
return null;
}

}

public static void main(String[] args) throws ParseException {
searchlog atpanelLog = new searchlog("atpanel.atpanel", "10.232.130.13:9090",
30000, 16, "ads", "ads");
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
long timestamp = format.parse("2011-12-05 10:00:00").getTime();
atpanelLog.SubClientInit();
ReadPosition r = atpanelLog.getReadPosition(1, timestamp);

while(true){
Iterator itr = atpanelLog.getDigital(1, timestamp, 10, r);
while ((itr != null) && (itr.hasNext())) {
Message msg = (Message)itr.next();
System.out.print(new String(msg.getData()));
// _collector.emit(new Values(new String(msg.getData())));
}
}
}
}

run complete!!!!
console correct!!

i do not know why ????
who can help me !!!