Fix group-by error when task has a group-by task and a non-group by task

downstream.
punit-naik · Apr 24, 2017 · 424a14f · 424a14f
1 parent 5e26b35
commit 424a14f
Showing 1 changed file with 5 additions and 7 deletions.
diff --git a/src/onyx/plugin/messaging_output.clj b/src/onyx/plugin/messaging_output.clj
@@ -77,14 +77,12 @@
                   replica messenger]
     (.clear ^java.util.ArrayList segments)
     (let [;; generate this on each new replica / messenger
-          get-pub-fn (if (empty? task->group-by-fn)
-                       (fn [segment dst-task-id]
-                         (rand-nth (m/task->publishers messenger dst-task-id)))            
-                       (fn [segment dst-task-id]
-                         (let [group-fn (task->group-by-fn dst-task-id) 
-                               hsh (hash (group-fn segment))
+          get-pub-fn (fn [segment dst-task-id]
+                       (if-let [group-fn (task->group-by-fn dst-task-id)]
+                         (let [hsh (hash (group-fn segment))
                                dest-pubs (m/task->publishers messenger dst-task-id)]
-                           (get dest-pubs (mod hsh (count dest-pubs))))))
+                           (get dest-pubs (mod hsh (count dest-pubs))))
+                         (rand-nth (m/task->publishers messenger dst-task-id))))
           ;; TODO, avoid initial flattening preprocessing step
           _ (run! (fn [{:keys [leaves] :as result}]
                     (run! (fn [seg]