You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@storm.apache.org by Banias H <ba...@gmail.com> on 2014/12/19 20:32:54 UTC
Help: KafkaSpout not getting data from Kafka
Hi folks,
I am new to both Kafka and Storm and I have problem having KafkaSpout to
get data from Kafka in our three-node environment with Kafka 0.8.1.1 and
Storm 0.9.3.
What is working:
- I have a Kafka producer (a java application) to generate random string to
a topic and I was able to run the following command in one of the nodes to
read the random strings on the console while the Kafka producer is running:
<kaffa folder>/bin/kafka-console-consumer.sh --zookeeper
<node1>:2181,<node2>:2181,<node3>:2181/kafka --topic test_topic
- I was also able to run WordCountTopology in Storm.
What is not working:
- I tried running the following code based on KafkaTopology.java:
String zkNodes = "<node1>:2181,<node2>:2181,<node3>:2181";
String brokerZkPath = "/kafka/brokers";
String topicName = "test_topic";
String zkRoot = "/kafka";
String topoName = "test_topology";
ZkHosts zkhost = new ZkHosts(zkNodes, brokerZkPath);
SpoutConfig kafkaConf = new SpoutConfig(zkhost, topicName, zkRoot,
"discovery");
kafkaConf.scheme = new SchemeAsMultiScheme(new StringScheme());
kafkaConf.forceStartOffsetTime(-2);
KafkaSpout kafkaSpout = new KafkaSpout(kafkaConf);
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", kafkaSpout, 1);
builder.setBolt("printer", new
PrinterBolt()).shuffleGrouping("spout2");
Config config = new Config();
config.setDebug(true);
config.setNumWorkers(3);
StormSubmitter.submitTopology(topoName, config,
builder.createTopology());
- Result:
In Storm UI, the numbers of emitted and transferred are always 0,
regardless of whether Kafka producer is running. See attached image. I
- Command to run:
<storm folder>/bin/storm jar storm-starter-0.9.3-jar-with-dependencies.jar
storm.starter.KafkaTopology
- Zookeeper Path:
[zk: <node1>:2181(CONNECTED) 73] ls /kafka
[consumers, config, controller, admin, brokers, controller_epoch]
[zk: <node1>:2181(CONNECTED) 74] ls /kafka/brokers
[topics, ids]
I have run out of ideas of trying different options and figuring out where
to look. If anyone could shed some light on this topic, I would greatly
appreciate it. Many thanks!
-BH
Re: Help: KafkaSpout not getting data from Kafka
Posted by Ankit Toshniwal <an...@gmail.com>.
Couple of questions:
1. Are the workers JVM starting up? if yes, can you send us a copy of the
worker logs.
2. Also have you tried with forceStartOffsetTime set to -1?
Ankit
On Fri, Dec 19, 2014 at 11:32 AM, Banias H <ba...@gmail.com> wrote:
> Hi folks,
>
> I am new to both Kafka and Storm and I have problem having KafkaSpout to
> get data from Kafka in our three-node environment with Kafka 0.8.1.1 and
> Storm 0.9.3.
>
> What is working:
> - I have a Kafka producer (a java application) to generate random string
> to a topic and I was able to run the following command in one of the nodes
> to read the random strings on the console while the Kafka producer is
> running:
>
> <kaffa folder>/bin/kafka-console-consumer.sh --zookeeper
> <node1>:2181,<node2>:2181,<node3>:2181/kafka --topic test_topic
>
> - I was also able to run WordCountTopology in Storm.
>
> What is not working:
> - I tried running the following code based on KafkaTopology.java:
>
> String zkNodes = "<node1>:2181,<node2>:2181,<node3>:2181";
> String brokerZkPath = "/kafka/brokers";
> String topicName = "test_topic";
> String zkRoot = "/kafka";
> String topoName = "test_topology";
>
> ZkHosts zkhost = new ZkHosts(zkNodes, brokerZkPath);
> SpoutConfig kafkaConf = new SpoutConfig(zkhost, topicName, zkRoot,
> "discovery");
> kafkaConf.scheme = new SchemeAsMultiScheme(new StringScheme());
> kafkaConf.forceStartOffsetTime(-2);
> KafkaSpout kafkaSpout = new KafkaSpout(kafkaConf);
>
> TopologyBuilder builder = new TopologyBuilder();
> builder.setSpout("spout", kafkaSpout, 1);
> builder.setBolt("printer", new
> PrinterBolt()).shuffleGrouping("spout2");
>
> Config config = new Config();
> config.setDebug(true);
> config.setNumWorkers(3);
> StormSubmitter.submitTopology(topoName, config,
> builder.createTopology());
>
> - Result:
> In Storm UI, the numbers of emitted and transferred are always 0,
> regardless of whether Kafka producer is running. See attached image. I
>
> - Command to run:
> <storm folder>/bin/storm jar storm-starter-0.9.3-jar-with-dependencies.jar
> storm.starter.KafkaTopology
>
> - Zookeeper Path:
> [zk: <node1>:2181(CONNECTED) 73] ls /kafka
> [consumers, config, controller, admin, brokers, controller_epoch]
> [zk: <node1>:2181(CONNECTED) 74] ls /kafka/brokers
> [topics, ids]
>
> I have run out of ideas of trying different options and figuring out where
> to look. If anyone could shed some light on this topic, I would greatly
> appreciate it. Many thanks!
>
> -BH
>
Fwd: Help: KafkaSpout not getting data from Kafka
Posted by Banias H <ba...@gmail.com>.
Hi folks,
I am new to both Kafka and Storm and I have problem having KafkaSpout to
get data from Kafka in our three-node environment with Kafka 0.8.1.1 and
Storm 0.9.3.
What is working:
- I have a Kafka producer (a java application) to generate random string to
a topic and I was able to run the following command in one of the nodes to
read the random strings on the console while the Kafka producer is running:
<kaffa folder>/bin/kafka-console-consumer.sh --zookeeper
<node1>:2181,<node2>:2181,<node3>:2181/kafka --topic test_topic
- I was also able to run WordCountTopology in Storm.
What is not working:
- I tried running the following code based on KafkaTopology.java:
String zkNodes = "<node1>:2181,<node2>:2181,<node3>:2181";
String brokerZkPath = "/kafka/brokers";
String topicName = "test_topic";
String zkRoot = "/kafka";
String topoName = "test_topology";
ZkHosts zkhost = new ZkHosts(zkNodes, brokerZkPath);
SpoutConfig kafkaConf = new SpoutConfig(zkhost, topicName, zkRoot,
"discovery");
kafkaConf.scheme = new SchemeAsMultiScheme(new StringScheme());
kafkaConf.forceStartOffsetTime(-2);
KafkaSpout kafkaSpout = new KafkaSpout(kafkaConf);
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", kafkaSpout, 1);
builder.setBolt("printer", new
PrinterBolt()).shuffleGrouping("spout2");
Config config = new Config();
config.setDebug(true);
config.setNumWorkers(3);
StormSubmitter.submitTopology(topoName, config,
builder.createTopology());
- Result:
In Storm UI, the numbers of emitted and transferred are always 0,
regardless of whether Kafka producer is running. See attached image. I
- Command to run:
<storm folder>/bin/storm jar storm-starter-0.9.3-jar-with-dependencies.jar
storm.starter.KafkaTopology
- Zookeeper Path:
[zk: <node1>:2181(CONNECTED) 73] ls /kafka
[consumers, config, controller, admin, brokers, controller_epoch]
[zk: <node1>:2181(CONNECTED) 74] ls /kafka/brokers
[topics, ids]
I have run out of ideas of trying different options and figuring out where
to look. If anyone could shed some light on this topic, I would greatly
appreciate it. Many thanks!
-BH