     (fn [task-id ^List values load]
       (.chooseTasks grouping task-id values))))
 
-(defn mk-shuffle-grouper [^List target-tasks topo-conf ^WorkerTopologyContext context ^String component-id ^String stream-id]
+(defn mk-shuffle-grouper
+  [^List target-tasks topo-conf ^WorkerTopologyContext context ^String component-id ^String stream-id]
   (if (.get topo-conf TOPOLOGY-DISABLE-LOADAWARE-MESSAGING)
     (mk-custom-grouper (ShuffleGrouping.) context component-id stream-id target-tasks)
     (mk-custom-grouper (LoadAwareShuffleGrouping.) context component-id stream-id target-tasks)))

          (mod num-tasks)
          task-getter))))

(defn- mk-custom-grouper [^CustomStreamGrouping grouping ^WorkerTopologyContext context ^String component-id ^String stream-id target-tasks]
  (.prepare grouping context (GlobalStreamId. component-id stream-id) target-tasks)
  (if (instance? LoadAwareCustomStreamGrouping grouping)
    (fn [task-id ^List values load]

        :direct
      )))

(defn- outbound-groupings
  [^WorkerTopologyContext worker-context this-component-id stream-id out-fields component->grouping topo-conf]
  (->> component->grouping
       (filter-key #(-> worker-context
                        (.getComponentTasks %)
