openstack/monasca-thresh
Code Issues Proposed changes

Have to use a different consumer.id for each instance of a KafkaSpout so use the storm taskId. Otherwise, zookeeper complains about a conflicted ephemeral node when there is more than one spout reading from a topic

Browse Source 
Have two metric and event spouts to make sure the messages stop coming out. More like the real configuration.
This commit is contained in:
Craig Bryant 2014-05-01 13:44:43 -06:00
parent a3429536d2
commit 9050956a3b
2 changed files with 8 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/config/mon-thresh-sample-config.yml
View File

@ -1,5 +1,5 @@
metricSpoutThreads: 1
metricSpoutTasks: 1
metricSpoutThreads: 2
metricSpoutTasks: 2
metricSpoutConfig:
@ -29,8 +29,8 @@ metricSpoutConfig:
zookeeperConnectionTimeoutMs : 6000
zookeeperSyncTimeMs: 2000
eventSpoutThreads: 1
eventSpoutTasks: 1
eventSpoutThreads: 2
eventSpoutTasks: 2
eventSpoutConfig:
#Kafka settings.

4
src/main/java/com/hpcloud/mon/infrastructure/thresholding/KafkaSpout.java
View File

@ -53,6 +53,10 @@ public abstract class KafkaSpout extends BaseRichSpout {
LOG.info(" topic = " + kafkaConsumerConfig.getTopic());
Properties kafkaProperties = KafkaConsumerProperties.createKafkaProperties(kafkaConsumerConfig);
// Have to use a different consumer.id for each spout so use the storm taskId. Otherwise,
// zookeeper complains about a conflicted ephemeral node when there is more than one spout
// reading from a topic
kafkaProperties.setProperty("consumer.id", String.valueOf(context.getThisTaskId()));
ConsumerConfig consumerConfig = new ConsumerConfig(kafkaProperties);
this.consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);
}