Worker dies (bolt)
Hi
I am using apache-storm-0.9.1-incubating.
I have simple topology: Spout reads from kafka topic and Bolt writes
lines from spout to HBase.
recently we did a test - we send 300 000 000 messages over kafka-rest ->
kafka-queue -> storm topology -> hbase.
I noticed that around one hour and around 2500 messages worker died. PID
is there and process is up but bolt's execute method hangs.
Bolts code is:
package storm;
2
3 import java.util.Map;
4 import java.util.UUID;
5
6 import org.apache.hadoop.conf.Configuration;
7 import org.apache.hadoop.hbase.HBaseConfiguration;
8 import org.apache.hadoop.hbase.client.HTableInterface;
9 import org.apache.hadoop.hbase.client.HTablePool;
10 import org.apache.hadoop.hbase.client.Put;
11 import org.apache.hadoop.hbase.util.Bytes;
12
13 import backtype.storm.task.TopologyContext;
14 import backtype.storm.topology.BasicOutputCollector;
15 import backtype.storm.topology.OutputFieldsDeclarer;
16 import backtype.storm.topology.base.BaseBasicBolt;
17 import backtype.storm.tuple.Fields;
18 import backtype.storm.tuple.Tuple;
19 import backtype.storm.tuple.Values;
public class HBaseWriterBolt extends BaseBasicBolt
22 {
23
24 HTableInterface usersTable;
25 HTablePool pool;
26 int count = 0;
27
28 @Override
29 public void prepare(Map stormConf, TopologyContext context) {
30 Configuration conf = HBaseConfiguration.create();
31 conf.set("hbase.defaults.for.version","0.96.0.2.0.6.0-76-hadoop2");
32 conf.set("hbase.defaults.for.version.skip","true");
33 conf.set("hbase.zookeeper.quorum", "vm24,vm37,vm38");
34 conf.set("hbase.zookeeper.property.clientPort",
"2181");
35 conf.set("hbase.rootdir",
"hdfs://vm38:8020/user/hbase/data");
36 //conf.set("zookeeper.znode.parent",
"/hbase-unsecure");
37
38 pool = new HTablePool(conf, 1);
39 usersTable = pool.getTable("kafkademo1");
40 }
41
42 @Override
43 public void execute(Tuple tuple, BasicOutputCollector
collector)
44 {
45 String line = tuple.getString(0);
46
47 Put p = new
Put(Bytes.toBytes(UUID.randomUUID().toString()));
48 p.add(Bytes.toBytes("info"), Bytes.toBytes("line"),
Bytes.toBytes(line));
49
50 try {
51 usersTable.put(p);
52 count ++;
53 System.out.println("Count: "+ count);
54 }
55 catch (Exception e){
56 e.printStackTrace();
57 }
58 collector.emit(new Values(line));
59
60 }
61
62 @Override
63 public void declareOutputFields(OutputFieldsDeclarer declarer)
64 {
65 declarer.declare(new Fields("line"));
66 }
67
68 @Override
69 public void cleanup()
70 {
71 try {
72 usersTable.close();
73 }
74 catch (Exception e){
75 e.printStackTrace();
76 }
77 }
78
79 }
line in execute method: System.out.println("Count: "+ count); added in
debug purpose to see in log that bolt is running.
In to Spout in method nextTuple()
I added debug line: System.out.println("Message from the Topic ...");
After some time around 50minutes in log file I can see that Spout is
working but Bolt is died.
Any ideas?
--
Best regards, Margus (Margusja) Roo
+372 51 48 780
http://margus.roo.ee
http://ee.linkedin.com/in/margusroo
skype: margusja
ldapsearch -x -h ldap.sk.ee -b c=EE "(serialNumber=37303140314)"
Re: Worker dies (bolt)
Hei
I have made a new test and discovered that in my environment a very
simple bolt will die too after around 2500 cycle.
Bolt's code:
1 package storm;
2
3 import backtype.storm.task.TopologyContext;
4 import backtype.storm.topology.BasicOutputCollector;
5 import backtype.storm.topology.OutputFieldsDeclarer;
6 import backtype.storm.topology.base.BaseBasicBolt;
7 import backtype.storm.tuple.Fields;
8 import backtype.storm.tuple.Tuple;
9 import backtype.storm.tuple.Values;
10
11 import java.util.Map;
12 import java.util.UUID;
13
14 public class DummyBolt extends BaseBasicBolt
15 {
16 int count = 0;
17
18 @Override
19 public void prepare(Map stormConf, TopologyContext context) {
20 }
21
22 @Override
23 public void execute(Tuple tuple, BasicOutputCollector
collector)
24 {
25 String line = tuple.getString(0);
26
27 count ++;
28 System.out.println("Dummy count: "+ count);
29 collector.emit(new Values(line));
30
31 }
32
33 @Override
34 public void declareOutputFields(OutputFieldsDeclarer declarer)
35 {
36 declarer.declare(new Fields("line"));
37 }
38
39 @Override
40 public void cleanup()
41 {
42 }
43
44 }
after around 2500 cycles there is no output from execute methods.
What I do after this.
[root@dlvm2 sysconfig]# ls /var/lib/storm/workers/*/pids
[root@dlvm2 sysconfig]# kill -9 4179
after it new worker is coming up and it works again around 2500 cycles
and stops and I have to kill pid again.
Any ideas?
Best regards, Margus (Margusja) Roo
+372 51 48 780
http://margus.roo.ee
http://ee.linkedin.com/in/margusroo
skype: margusja
ldapsearch -x -h ldap.sk.ee -b c=EE "(serialNumber=37303140314)"
On 02/06/14 13:36, Margusja wrote:
Hi
I am using apache-storm-0.9.1-incubating.
I have simple topology: Spout reads from kafka topic and Bolt writes
lines from spout to HBase.
recently we did a test - we send 300 000 000 messages over kafka-rest
-> kafka-queue -> storm topology -> hbase.
I noticed that around one hour and around 2500 messages worker died.
PID is there and process is up but bolt's execute method hangs.
Bolts code is:
package storm;
2
3 import java.util.Map;
4 import java.util.UUID;
5
6 import org.apache.hadoop.conf.Configuration;
7 import org.apache.hadoop.hbase.HBaseConfiguration;
8 import org.apache.hadoop.hbase.client.HTableInterface;
9 import org.apache.hadoop.hbase.client.HTablePool;
10 import org.apache.hadoop.hbase.client.Put;
11 import org.apache.hadoop.hbase.util.Bytes;
12
13 import backtype.storm.task.TopologyContext;
14 import backtype.storm.topology.BasicOutputCollector;
15 import backtype.storm.topology.OutputFieldsDeclarer;
16 import backtype.storm.topology.base.BaseBasicBolt;
17 import backtype.storm.tuple.Fields;
18 import backtype.storm.tuple.Tuple;
19 import backtype.storm.tuple.Values;
public class HBaseWriterBolt extends BaseBasicBolt
22 {
23
24 HTableInterface usersTable;
25 HTablePool pool;
26 int count = 0;
27
28 @Override
29 public void prepare(Map stormConf, TopologyContext
context) {
30 Configuration conf = HBaseConfiguration.create();
31 conf.set("hbase.defaults.for.version","0.96.0.2.0.6.0-76-hadoop2");
32 conf.set("hbase.defaults.for.version.skip","true");
33 conf.set("hbase.zookeeper.quorum",
"vm24,vm37,vm38");
34 conf.set("hbase.zookeeper.property.clientPort", "2181");
35 conf.set("hbase.rootdir",
"hdfs://vm38:8020/user/hbase/data");
36 //conf.set("zookeeper.znode.parent",
"/hbase-unsecure");
37
38 pool = new HTablePool(conf, 1);
39 usersTable = pool.getTable("kafkademo1");
40 }
41
42 @Override
43 public void execute(Tuple tuple, BasicOutputCollector
collector)
44 {
45 String line = tuple.getString(0);
46
47 Put p = new
Put(Bytes.toBytes(UUID.randomUUID().toString()));
48 p.add(Bytes.toBytes("info"),
Bytes.toBytes("line"), Bytes.toBytes(line));
49
50 try {
51 usersTable.put(p);
52 count ++;
53 System.out.println("Count: "+ count);
54 }
55 catch (Exception e){
56 e.printStackTrace();
57 }
58 collector.emit(new Values(line));
59
60 }
61
62 @Override
63 public void declareOutputFields(OutputFieldsD
Re: Worker dies (bolt)
Some new information.
Set debug true and from active worker log I can see:
if worker is ok:
2014-06-03 11:04:55 b.s.d.task [INFO] Emitting: hbasewriter __ack_ack
[7197822474056634252 -608920652033678418]
2014-06-03 11:04:55 b.s.d.executor [INFO] Processing received message
source: hbasewriter:3, stream: __ack_ack, id: {}, [7197822474056634252
-608920652033678418]
2014-06-03 11:04:55 b.s.d.task [INFO] Emitting direct: 1; __acker
__ack_ack [7197822474056634252]
2014-06-03 11:04:55 b.s.d.executor [INFO] Processing received message
source: KafkaConsumerSpout:1, stream: default, id:
{4344988213623161794=-5214435544383558411}, my message...
and after worker dies there are only rows about spout like:
2014-06-03 11:06:30 b.s.d.task [INFO] Emitting: KafkaConsumerSpout
__ack_init [3399515592775976300 5357635772515085965 1]
2014-06-03 11:06:30 b.s.d.task [INFO] Emitting: KafkaConsumerSpout default
Best regards, Margus (Margusja) Roo
+372 51 48 780
http://margus.roo.ee
http://ee.linkedin.com/in/margusroo
skype: margusja
ldapsearch -x -h ldap.sk.ee -b c=EE "(serialNumber=37303140314)"
On 03/06/14 09:58, Margusja wrote:
Hei
I have made a new test and discovered that in my environment a very
simple bolt will die too after around 2500 cycle.
Bolt's code:
1 package storm;
2
3 import backtype.storm.task.TopologyContext;
4 import backtype.storm.topology.BasicOutputCollector;
5 import backtype.storm.topology.OutputFieldsDeclarer;
6 import backtype.storm.topology.base.BaseBasicBolt;
7 import backtype.storm.tuple.Fields;
8 import backtype.storm.tuple.Tuple;
9 import backtype.storm.tuple.Values;
10
11 import java.util.Map;
12 import java.util.UUID;
13
14 public class DummyBolt extends BaseBasicBolt
15 {
16 int count = 0;
17
18 @Override
19 public void prepare(Map stormConf, TopologyContext
context) {
20 }
21
22 @Override
23 public void execute(Tuple tuple, BasicOutputCollector
collector)
24 {
25 String line = tuple.getString(0);
26
27 count ++;
28 System.out.println("Dummy count: "+ count);
29 collector.emit(new Values(line));
30
31 }
32
33 @Override
34 public void declareOutputFields(OutputFieldsDeclarer
declarer)
35 {
36 declarer.declare(new Fields("line"));
37 }
38
39 @Override
40 public void cleanup()
41 {
42 }
43
44 }
after around 2500 cycles there is no output from execute methods.
What I do after this.
[root@dlvm2 sysconfig]# ls /var/lib/storm/workers/*/pids
[root@dlvm2 sysconfig]# kill -9 4179
after it new worker is coming up and it works again around 2500
cycles and stops and I have to kill pid again.
Any ideas?
Best regards, Margus (Margusja) Roo
+372 51 48 780
http://margus.roo.ee
http://ee.linkedin.com/in/margusroo
skype: margusja
ldapsearch -x -h ldap.sk.ee -b c=EE "(serialNumber=37303140314)"
On 02/06/14 13:36, Margusja wrote:
Hi
I am using apache-storm-0.9.1-incubating.
I have simple topology: Spout reads from kafka topic and Bolt writes
lines from spout to HBase.
recently we did a test - we send 300 000 000 messages over kafka-rest
-> kafka-queue -> storm topology -> hbase.
I noticed that around one hour and around 2500 messages worker died.
PID is there and process is up but bolt's execute method hangs.
Bolts code is:
package storm;
2
3 import java.util.Map;
4 import java.util.UUID;
5
6 import org.apache.hadoop.conf.Configuration;
7 import org.apache.hadoop.hbase.HBaseConfiguration;
8 import org.apache.hadoop.hbase.client.HTableInterface;
9 import org.apache.hadoop.hbase.client.HTablePool;
10 import org.apache.hadoop.hbase.client.Put;
11 import org.apache.hadoop.hbase.util.Bytes;
12
13 import backtype.storm.task.TopologyContext;
14 import backtype.storm.topology.BasicOutputCollector;
15 import backtype.storm.topology.OutputFieldsDeclarer;
16 import backtype.storm.topology.base.BaseBasicBolt;
17 import backtype.storm.tuple.Fields;
18 import backtype.storm.tuple.Tuple;
19 import backtype.storm.tuple.Values;
public class HBaseWriterBolt extends BaseBasicBolt
22 {
23
24 HTableInterface usersTable;
25 HTablePool pool;
26 int count = 0;
27
28 @Override
29 public void prepare(Map stormConf, TopologyContext
context) {
30 Configuration conf = HBaseConfiguration.create();
31 conf.set("hbase.defaults.for.version","0.96.0.2.0.6.0-76-hadoop2");
32 conf.set("hbase.defaults.for.version.skip","true");
33 conf.set("hbase.zookeeper.quorum",
"vm24,vm37,vm38");
34 conf.set("hbase.zookeeper.property.clientPort", "2181");
35 conf.set("hbase.rootdir",
&qu
Re: Worker dies (bolt)
Ok got more info.
Looks like the problem is related with spout.
I changed spout:
32 public void open(Map conf, TopologyContext
context,SpoutOutputCollector collector)
33 {
34 this.collector = collector;
35
36 Properties props = new Properties();
37 props.put("zookeeper.connect",
"vm24:2181,vm37:2181,vm38:2181");
38 props.put("group.id", "testgroup");
39 props.put("zookeeper.session.timeout.ms", "500");
40 props.put("zookeeper.sync.time.ms", "250");
41 props.put("auto.commit.interval.ms", "1000");
42 consumer = Consumer.createJavaConsumerConnector(new
ConsumerConfig(props));
43 this.topic = "kafkademo1";
44
45
46 }
to
32 public void open(Map conf, TopologyContext
context,SpoutOutputCollector collector)
33 {
34 this.collector = collector;
35
36 Properties props = new Properties();
37 props.put("zookeeper.connect",
"vm24:2181,vm37:2181,vm38:2181");
38 props.put("group.id", "testgroup");
39 //props.put("zookeeper.session.timeout.ms", "500");
40 //props.put("zookeeper.sync.time.ms", "250");
41 //props.put("auto.commit.interval.ms", "1000");
42 consumer = Consumer.createJavaConsumerConnector(new
ConsumerConfig(props));
43 this.topic = "kafkademo1";
44
45
46 }
and
48 public void nextTuple()
49 {
50
55
56 Map topicCount = new HashMapInteger>();
57 // Define single thread for topic
58 topicCount.put(topic, new Integer(1));
59 Map>>
consumerStreams = consumer.createMessageStreams(topicCount);
60 List> streams =
consumerStreams.get(topic);
61 for (final KafkaStream stream : streams) {
62 ConsumerIterator consumerIte =
stream.iterator();
63 while (consumerIte.hasNext())
64 {
65 // System.out.println("Message from the Topic ...");
66 String line = new
String(consumerIte.next().message());
67 this.collector.emit(new Values(line), line);
69 }
70
71
72 }
73 if (consumer != null)
74 consumer.shutdown();
75 }
to
48 public void nextTuple()
49 {
50
55
56 Map topicCount = new HashMapInteger>();
57 // Define single thread for topic
58 topicCount.put(topic, new Integer(1));
59 Map>>
consumerStreams = consumer.createMessageStreams(topicCount);
60 List> streams =
consumerStreams.get(topic);
61 for (final KafkaStream stream : streams) {
62 ConsumerIterator consumerIte =
stream.iterator();
63 while (consumerIte.hasNext())
64 {
65 // System.out.println("Message from the Topic ...");
66 String line = new
String(consumerIte.next().message());
67 //this.collector.emit(new Values(line), line);
68 this.collector.emit(new Values(line));
69 }
70
71
72 }
73 if (consumer != null)
74 consumer.shutdown();
75 }
And now it is running.
Strange because when worker died then I see log rows from spout. But I
think it is related somehow with the internal stuff in storm.
Best regards, Margus (Margusja) Roo
+372 51 48 780
http://margus.roo.ee
http://ee.linkedin.com/in/margusroo
skype: margusja
ldapsearch -x -h ldap.sk.ee -b c=EE "(serialNumber=37303140314)"
On 03/06/14 11:09, Margusja wrote:
Some new information.
Set debug true and from active worker log I can see:
if worker is ok:
2014-06-03 11:04:55 b.s.d.task [INFO] Emitting: hbasewriter __ack_ack
[7197822474056634252 -608920652033678418]
2014-06-03 11:04:55 b.s.d.executor [INFO] Processing received message
source: hbasewriter:3, stream: __ack_ack, id: {}, [7197822474056634252
-608920652033678418]
2014-06-03 11:04:55 b.s.d.task [INFO] Emitting direct: 1; __acker
__ack_ack [7197822474056634252]
2014-06-03 11:04:55 b.s.d.executor [INFO] Processing received message
source: KafkaConsumerSpout:1, stream: default, id:
{4344988213623161794=-5214435544383558411}, my message...
and after worker dies there are only rows about spout like:
2014-06-03 11:06:30 b.s.d.task [INFO] Emitting: KafkaConsumerSpout
__ack_init [3399515592775976300 5357635772515085965 1]
2014-06-03 11:06:30 b.s.d.t
