CyanogenMod CyanogenMod - 11 months ago 60
Java Question

Storm Kafka-Spout not work properly

General: I'm a student who want to run some performance tests (WordCount) on Storm / Kafka / Flink / MS Azure SA / Spark. I want to use the Kafka Broker as an input source.

I used the WordCount Example from the Storm-Starter project and added Kafka as a spout:

public class WordCountKafkaTopology {
public static class SplitSentence extends ShellBolt implements IRichBolt {

public SplitSentence() {
super("python", "splitsentence.py");
}

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}

@Override
public Map<String, Object> getComponentConfiguration() {
return null;
}
}

public static class WordCount extends BaseBasicBolt {
Map<String, Integer> counts = new HashMap<String, Integer>();

@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
String word = tuple.getString(0);
Integer count = counts.get(word);
if (count == null)
count = 0;
count++;
counts.put(word, count);
collector.emit(new Values(word, count));
}

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word", "count"));
}
}

public static void main(String[] args) {

String zkIp = "localhost";

String topicName = "perfTest";

List<String> nimbus_seeds = new ArrayList<String>();
nimbus_seeds.add("localhost");

String zookeeperHost = zkIp +":2181";

ZkHosts zkHosts = new ZkHosts(zookeeperHost);

SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, topicName, "/" + topicName, topicName);
kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
KafkaSpout kafkaSpout = new KafkaSpout(kafkaConfig);


TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("kafkaPerfTestSpout", kafkaSpout, 8);

builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("kafkaPerfTestSpout");
builder.setBolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));

Config config = new Config();

config.setMaxTaskParallelism(5);
config.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 2);
config.put(Config.NIMBUS_SEEDS, nimbus_seeds);
config.put(Config.NIMBUS_THRIFT_PORT, 6627);
config.put(Config.STORM_ZOOKEEPER_PORT, 2181);
config.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList(zkIp));

try {
StormSubmitter.submitTopology("my-kafka-topology", config, builder.createTopology());
} catch (Exception e) {
throw new IllegalStateException("Couldn't initialize the topology", e);
}
}

}


By running the topolgy I get serveral error messages. The spout says:


java.lang.ExceptionInInitializerError at kafka.metrics.KafkaMetricsGroup$class.newTimer(KafkaMetricsGroup.scala:89) at kafka.consumer.FetchRequestAndResponseMetrics.newTimer(FetchRequestAndResponseStats.scala:26) at kafka.consumer.FetchRequestAndResponseMetrics.(FetchRequestAndResponseStats.scala:35) at kafka.consumer.FetchRequestAndResponseStats.(FetchRequestAndResponseStats.scala:47) at kafka.consumer.FetchRequestAndResponseStatsRegistry$$anonfun$2.apply(FetchRequestAndResponseStats.scala:60) at kafka.consumer.FetchRequestAndResponseStatsRegistry$$anonfun$2.apply(FetchRequestAndResponseStats.scala:60) at kafka.utils.Pool.getAndMaybePut(Pool.scala:59) at kafka.consumer.FetchRequestAndResponseStatsRegistry$.getFetchRequestAndResponseStats(FetchRequestAndResponseStats.scala:64) at kafka.consumer.SimpleConsumer.(SimpleConsumer.scala:44) at kafka.javaapi.consumer.SimpleConsumer.(SimpleConsumer.scala:34) at org.apache.storm.kafka.DynamicPartitionConnections.register(DynamicPartitionConnections.java:60) at org.apache.storm.kafka.PartitionManager.(PartitionManager.java:74) at org.apache.storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:98) at org.apache.storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:69) at org.apache.storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:129) at org.apache.storm.daemon.executor$fn__7990$fn__8005$fn__8036.invoke(executor.clj:648) at org.apache.storm.util$async_loop$fn__624.invoke(util.clj:484) at clojure.lang.AFn.run(AFn.java:22) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.IllegalStateException: Shutdown in progress at java.lang.ApplicationShutdownHooks.add(ApplicationShutdownHooks.java:66) at java.lang.Runtime.addShutdownHook(Runtime.java:211) at com.yammer.metrics.Metrics.(Metrics.java:21) ... 19 more


At the split bolt:


java.lang.RuntimeException: java.lang.RuntimeException: java.lang.RuntimeException: pid:3973, name:split exitCode:0, errorString: at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:464) at org.apache.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:430) at org.apache.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:73) at org.apache.storm.daemon.executor$fn__8058$fn__8071$fn__8124.invoke(executor.clj:850) at org.apache.storm.util$async_loop$fn__624.invoke(util.clj:484) at clojure.lang.AFn.run(AFn.java:22) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.RuntimeException: java.lang.RuntimeException: pid:3973, name:split exitCode:0, errorString: at org.apache.storm.task.ShellBolt.execute(ShellBolt.java:150) at org.apache.storm.daemon.executor$fn__8058$tuple_action_fn__8060.invoke(executor.clj:731) at org.apache.storm.daemon.executor$mk_task_receiver$fn__7979.invoke(executor.clj:464) at org.apache.storm.disruptor$clojure_handler$reify__7492.onEvent(disruptor.clj:40) at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:451) ... 6 more Caused by: java.lang.RuntimeException: pid:3973, name:split exitCode:0, errorString: at org.apache.storm.task.ShellBolt.die(ShellBolt.java:295) at org.apache.storm.task.ShellBolt.access$400(ShellBolt.java:70) at org.apache.storm.task.ShellBolt$BoltWriterRunnable.run(ShellBolt.java:398) ... 1 more Caused by: java.io.IOException: Broken pipe at java.io.FileOutputStream.writeBytes(Native Method) at java.io.FileOutputStream.write(FileOutputStream.java:326) at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82) at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140) at sun.nio.cs.StreamEncoder.implFlush(StreamEncoder.java:297) at sun.nio.cs.StreamEncoder.flush(StreamEncoder.java:141) at java.io.OutputStreamWriter.flush(OutputStreamWriter.java:229) at java.io.BufferedWriter.flush(BufferedWriter.java:254) at org.apache.storm.multilang.JsonSerializer.writeString(JsonSerializer.java:99) at org.apache.storm.multilang.JsonSerializer.writeMessage(JsonSerializer.java:93) at org.apache.storm.multilang.JsonSerializer.writeBoltMsg(JsonSerializer.java:78) at org.apache.storm.utils.ShellProcess.writeBoltMsg(ShellProcess.java:127) at org.apache.storm.task.ShellBolt$BoltWriterRunnable.run(ShellBolt.java:387) ... 1 more


I use the kafka-console-producer to generate some messages. I hope someone can help me. I'am a rookie in programming storm...

Answer Source

Deleting "config.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 2);" did the job!

Recommended from our users: Dynamic Network Monitoring from WhatsUp Gold from IPSwitch. Free Download