djcoder100
diff --git a/‎src/clj/backtype/storm/daemon/common.clj
+20-13 b/‎src/clj/backtype/storm/daemon/common.clj
+20-13
diff --git a/‎src/clj/backtype/storm/daemon/nimbus.clj
+6-2 b/‎src/clj/backtype/storm/daemon/nimbus.clj
+6-2
diff --git a/‎src/jvm/backtype/storm/generated/Nimbus.java
+104-6 b/‎src/jvm/backtype/storm/generated/Nimbus.java
+104-6
@@ -109,20 +109,33 @@
                   (str id " is not a valid stream id"))))))
     ))
 
+(defn all-components [^StormTopology topology]
+  (apply merge {}
+         (for [f thrift/STORM-TOPOLOGY-FIELDS]
+           (.getFieldValue topology f)
+           )))
+
+(defn component-conf [component]
+  (->> component
+      .get_common
+      .get_json_conf
+      from-json))
+
 (defn validate-basic! [^StormTopology topology]
   (validate-ids! topology)
   (doseq [f thrift/SPOUT-FIELDS
           obj (->> f (.getFieldValue topology) vals)]
     (if-not (empty? (-> obj .get_common .get_inputs))
-      (throw (InvalidTopologyException. "May not declare inputs for a spout"))
+      (throw (InvalidTopologyException. "May not declare inputs for a spout"))))
+  (doseq [[comp-id comp] (all-components topology)
+          :let [conf (component-conf comp)
+                p (-> comp .get_common thrift/parallelism-hint)]]
+    (when (and (> (conf TOPOLOGY-TASKS) 0)
+               p
+               (<= p 0))
+      (throw (InvalidTopologyException. "Number of executors must be greater than 0 when number of tasks is greater than 0"))
       )))
 
-(defn all-components [^StormTopology topology]
-  (apply merge {}
-         (for [f thrift/STORM-TOPOLOGY-FIELDS]
-           (.getFieldValue topology f)
-           )))
-
 (defn validate-structure! [^StormTopology topology]
   ;; validate all the component subscribe from component+stream which actually exists in the topology
   ;; and if it is a fields grouping, validate the corresponding field exists  
@@ -144,12 +157,6 @@
                   (when-not (empty? diff-fields)
                     (throw (InvalidTopologyException. (str "Component: [" id "] subscribes from stream: [" source-stream-id "] of component [" source-component-id "] with non-existent fields: " diff-fields)))))))))))))
 
-(defn component-conf [component]
-  (->> component
-      .get_common
-      .get_json_conf
-      from-json))
-
 (defn acker-inputs [^StormTopology topology]
   (let [bolt-ids (.. topology get_bolts keySet)
         spout-ids (.. topology get_spouts keySet)
 
@@ -744,10 +744,14 @@
                          (.get_wait_secs options))
               num-workers (if (.is_set_num_workers options)
                             (.get_num_workers options))
-              executor-overrrides (if (.is_set_num_executors options)
+              executor-overrides (if (.is_set_num_executors options)
                                     (.get_num_executors options)
                                     {})]
-          (transition-name! nimbus storm-name [:rebalance wait-amt num-workers executor-overrrides] true)
+          (doseq [[c num-executors] executor-overrides]
+            (when (<= num-executors 0)
+              (throw (InvalidTopologyException. "Number of executors must be greater than 0"))
+              ))
+          (transition-name! nimbus storm-name [:rebalance wait-amt num-workers executor-overrides] true)
           ))
 
       (activate [this storm-name]