[jira] [Created] (STORM-353) Add configuration per kafka-bolt

["Haralds Ulmanis (JIRA)" <ji...@apache.org>]

Haralds Ulmanis created STORM-353:
-------------------------------------

             Summary: Add configuration per kafka-bolt 
                 Key: STORM-353
                 URL: https://issues.apache.org/jira/browse/STORM-353
             Project: Apache Storm (Incubating)
          Issue Type: Improvement
            Reporter: Haralds Ulmanis
            Priority: Trivial


Currently kafka bolt configuration is passed through storm configuration - which limits possibility to use more than one storm bolt/topic per topology.

I can suggest something like this:
diff --git a/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java b/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
index b9ea948..2a78f84 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
@@ -55,14 +55,20 @@ public class KafkaBolt<K, V> extends BaseRichBolt {
     private OutputCollector collector;
     private String topic;
 
+    private Map boltConfig;
+
+    public KafkaBolt(Map boltConfig) {
+        this.boltConfig = boltConfig;
+    }
+
     @Override
     public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
-        Map configMap = (Map) stormConf.get(KAFKA_BROKER_PROPERTIES);
+        Map configMap = (Map) boltConfig.get(KAFKA_BROKER_PROPERTIES);
         Properties properties = new Properties();
         properties.putAll(configMap);
         ProducerConfig config = new ProducerConfig(properties);
         producer = new Producer<K, V>(config);
-        this.topic = (String) stormConf.get(TOPIC);
+        this.topic = (String) boltConfig.get(TOPIC);
         this.collector = collector;
     }
 
After which you can initialize each bolt with own topic and even own zk servers.

Use case: 
I'm using several streams for output and want each of stream to be published in own topic. 



--
This message was sent by Atlassian JIRA
(v6.2#6252)