You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by "Robert Joseph Evans (JIRA)" <ji...@apache.org> on 2014/06/16 19:22:01 UTC
[jira] [Commented] (STORM-353) Add configuration per kafka-bolt
[ https://issues.apache.org/jira/browse/STORM-353?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14032655#comment-14032655 ]
Robert Joseph Evans commented on STORM-353:
-------------------------------------------
My biggest problem with the proposed solution is that it breaks backwards compatibility. I would like to see a default constructor, and configs falling back to the storm conf if not found in the passed in. Also if you could turn this into a github pull request to apache/incubator-storm that would be great. Please be sure to include STORM-353 in the title of the pull request.
> 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)