ClusterState cluster_state = new DistributedClusterState(conf);
StormClusterState zk = new StormZkClusterState(
cluster_state);
for(Integer taskid:zk.task_ids(stromId))
{
System.out.println("########"+taskid);
for(TaskError err:zk.task_errors(stromId, taskid))
{
System.out.println(err.getError());