public static void main(String[] args) throws Exception {
    VerifiableProperties verifiableProperties = ToolUtils.getVerifiableProperties(args);
    AggregationToolConfig config = new AggregationToolConfig(verifiableProperties);
    Map<String, ClusterMapUtils.DcZkInfo> dataCenterToZKAddress =
        ClusterMapUtils.parseDcJsonAndPopulateDcInfo(Utils.readStringFromFile(config.zkLayoutFilePath));
    String clusterName = config.clusterName;
    String workflowName = config.workflowName;
    long recurrentIntervalInMinutes = config.recurrentIntervalInMinutes;
    boolean isDelete = config.deleteSpecifiedWorkflow;
    boolean isRecurrentWorkflow = recurrentIntervalInMinutes != Utils.Infinite_Time;
    for (ClusterMapUtils.DcZkInfo zkInfo : dataCenterToZKAddress.values()) {
      // If there are multiple ZK endpoints in same dc, we trigger stats aggregation for each of them.
      for (String zkAddress : zkInfo.getZkConnectStrs()) {
        ZkClient zkClient = new ZkClient(zkAddress, SESSION_TIMEOUT, CONNECTION_TIMEOUT, new ZNRecordSerializer());
        TaskDriver taskDriver = new TaskDriver(zkClient, clusterName);
        if (isDelete) {
          try {
            taskDriver.waitToStop(workflowName, TIME_OUT_MILLI_SEC);
            taskDriver.delete(workflowName);
            System.out.println(
                String.format("Successfully deleted the workflow: %s in cluster %s at %s", workflowName, clusterName,
                    zkAddress));
          } catch (Exception | Error e) {
            System.out.println(
                String.format("Failed to delete %s. Workflow not found in cluster %s at %s", workflowName, clusterName,
                    zkAddress));
          }
        } else {
          Workflow.Builder workflowBuilder = new Workflow.Builder(workflowName);
          try {
            switch (config.taskType) {
              case AGGREGATE_TASK:
                buildAggregationTaskWorkflow(workflowBuilder, config, isRecurrentWorkflow);
                break;
              case DEPRECATED_CONTAINER_CLOUD_SYNC_TASK:
                buildDeprecatedContainerCloudSyncTaskWorkflow(workflowBuilder);
                break;
              case PROPERTY_STORE_CLEANUP_TASK:
                buildPropertyStoreCleanUpTaskWorkflow(workflowBuilder);
                break;
              default:
                throw new IllegalArgumentException("Invalid task type: " + config.taskType);
            }
            if (isRecurrentWorkflow) {
              workflowBuilder.setScheduleConfig(
                  ScheduleConfig.recurringFromNow(TimeUnit.MINUTES, recurrentIntervalInMinutes));
              workflowBuilder.setExpiry(TimeUnit.MINUTES.toMillis(recurrentIntervalInMinutes));
            }
            Workflow workflow = workflowBuilder.build();
            taskDriver.start(workflow);
            System.out.println(
                String.format("%s started successfully in cluster %s at %s", workflowName, clusterName, zkAddress));
          } catch (Exception | Error e) {
            System.out.println(
                String.format("Failed to start %s in cluster %s at %s", workflowName, clusterName, zkAddress));
          }
        }
      }
    }
  }
Runs the cluster wide aggregation tool
@param args arguments specifying config file. For example: --propsFile /path/AggregationToolConfig
@throws Exception in case of any error.
ambry-tools/src/main/java/com/github/ambry/clustermap/HelixTaskWorkflowManagerTool.java