code: public void setupZkTaskInfo(Map<Object, Object> conf, String topologyId, StormClusterState stormClusterState) throws Exception {
          Map<Integer, TaskInfo> taskToTaskInfo = mkTaskComponentAssignments(conf, topologyId);
  
          // mkdir /ZK/taskbeats/topoologyId
          int masterId = NimbusUtils.getTopologyMasterId(taskToTaskInfo);
          TopologyTaskHbInfo topoTaskHbInfo = new TopologyTaskHbInfo(topologyId, masterId);
          data.getTasksHeartbeat().put(topologyId, topoTaskHbInfo);
          stormClusterState.topology_heartbeat(topologyId, topoTaskHbInfo);
  
          if (taskToTaskInfo == null || taskToTaskInfo.size() == 0) {
              throw new InvalidTopologyException("Failed to generate TaskIDs map");
          }
          // key is task id, value is task info
          stormClusterState.set_task(topologyId, taskToTaskInfo);
      }
docs: generate TaskInfo for every bolt or spout in ZK /ZK/tasks/topoologyId/xxx