Repository: storm Updated Branches: refs/heads/1.0.x-branch fc4179c0f -> 01c9923d0
STORM-2003 : Make sure config contains TOPIC before get it Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/c11f2b1c Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c11f2b1c Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c11f2b1c Branch: refs/heads/1.0.x-branch Commit: c11f2b1c9b85f0a43dc5c23472dff951cf45c22a Parents: fc4179c Author: darionyaphet <darion.yap...@gmail.com> Authored: Mon Jul 25 22:26:35 2016 +0800 Committer: Jungtaek Lim <kabh...@gmail.com> Committed: Fri Sep 23 10:20:47 2016 +0900 ---------------------------------------------------------------------- .../src/jvm/org/apache/storm/kafka/bolt/KafkaBolt.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/c11f2b1c/external/storm-kafka/src/jvm/org/apache/storm/kafka/bolt/KafkaBolt.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/bolt/KafkaBolt.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/bolt/KafkaBolt.java index 0ceac3a..31eaac5 100644 --- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/bolt/KafkaBolt.java +++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/bolt/KafkaBolt.java @@ -98,7 +98,11 @@ public class KafkaBolt<K, V> extends BaseRichBolt { //for backward compatibility. if(topicSelector == null) { - this.topicSelector = new DefaultTopicSelector((String) stormConf.get(TOPIC)); + if(stormConf.containsKey(TOPIC)) { + this.topicSelector = new DefaultTopicSelector((String) stormConf.get(TOPIC)); + } else { + throw new IllegalArgumentException("topic should be specified in bolt's configuration"); + } } producer = new KafkaProducer<>(boltSpecfiedProperties);
