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);

Reply via email to