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

["Rick Kellogg (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/STORM-353?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Rick Kellogg updated STORM-353:
-------------------------------
    Component/s: storm-kafka

> Add configuration per kafka-bolt 
> ---------------------------------
>
>                 Key: STORM-353
>                 URL: https://issues.apache.org/jira/browse/STORM-353
>             Project: Apache Storm
>          Issue Type: Improvement
>          Components: storm-kafka
>            Reporter: Haralds Ulmanis
>            Priority: Trivial
>         Attachments: storm-353.patch
>
>
> 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.3.4#6332)