[jira] [Work logged] (GOBBLIN-776) Add a utility method to return Helix WorflowId given a Gobblin job name

["ASF GitHub Bot (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/GOBBLIN-776?focusedWorklogId=252780&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-252780 ]

ASF GitHub Bot logged work on GOBBLIN-776:
------------------------------------------

                Author: ASF GitHub Bot
            Created on: 01/Jun/19 20:54
            Start Date: 01/Jun/19 20:54
    Worklog Time Spent: 10m 
      Work Description: shirshanka commented on pull request #2654: GOBBLIN-776: Add a utility method to return Helix WorflowId given a Gobblin job name.
URL: https://github.com/apache/incubator-gobblin/pull/2654#discussion_r289618070
 
 

 ##########
 File path: gobblin-cluster/src/main/java/org/apache/gobblin/cluster/HelixUtils.java
 ##########
 @@ -256,37 +257,37 @@ private static void deleteStoppedHelixJob(HelixManager helixManager, String work
     new TaskDriver(helixManager).deleteAndWaitForCompletion(workFlowName, 10000L);
     log.info("Workflow deleted.");
   }
-
   /**
    * Returns the Helix Workflow Ids given {@link Iterable} of Gobblin job names. The method returns a
-   * {@link Map} from Gobblin job name to the corresponding Helix Workflow Id. This method iterates
-   * each Helix workflow, and obtains the jobs of each workflow from its jobDag.
+   * {@link java.util.Map} from Gobblin job name to the corresponding Helix Workflow Id. This method iterates
+   * over all Helix workflows, and obtains the jobs of each workflow from its jobDag.
+   *
+   * NOTE: This call is expensive as it results in listing of znodes and subsequently, multiple ZK calls to get the job
+   * configuration for each HelixJob. Ideally, this method should be called infrequently e.g. when a job is deleted/cancelled.
+   *
    * @param jobNames a list of Gobblin job names.
    * @return a map from jobNames to their Helix Workflow Ids.
    */
-  public static Map<String, String> getWorkflowIdsFromJobNames(HelixManager helixManager, Iterable<String> jobNames) {
+  public static Map<String, String> getWorkflowIdsFromJobNames(HelixManager helixManager, Collection<String> jobNames) {
     Map<String, String> jobNameToWorkflowId = new HashMap<>();
     TaskDriver taskDriver = new TaskDriver(helixManager);
     Map<String, WorkflowConfig> workflowConfigMap = taskDriver.getWorkflows();
-    for (String workflow: workflowConfigMap.keySet()) {
+    for (String workflow : workflowConfigMap.keySet()) {
       WorkflowConfig workflowConfig = taskDriver.getWorkflowConfig(workflow);
       Set<String> helixJobs = workflowConfig.getJobDag().getAllNodes();
-
-      for (String helixJob: helixJobs) {
+      for (String helixJob : helixJobs) {
         Iterator<TaskConfig> taskConfigIterator = taskDriver.getJobConfig(helixJob).getTaskConfigMap().values().iterator();
         if (taskConfigIterator.hasNext()) {
           TaskConfig taskConfig = taskConfigIterator.next();
-          for (String jobName: jobNames) {
-            if (jobName.equals(taskConfig.getConfigMap().get(ConfigurationKeys.JOB_NAME_KEY))) {
-              if (!jobNameToWorkflowId.containsKey(jobName)) {
-                jobNameToWorkflowId.put(jobName, workflow);
-              } else {
-                String workflowId = jobNameToWorkflowId.get(jobName);
-                log.warn("JobName {} previously found to have WorkflowId {}; found "
-                    + " a different WorkflowId {} for the job; Skipping this entry", jobName, workflowId);
-              }
-              break;
+          String jobName = taskConfig.getConfigMap().get(ConfigurationKeys.JOB_NAME_KEY);
+          if (jobNames.contains(jobName)) {
+            if (!jobNameToWorkflowId.containsKey(jobName)) {
+              jobNameToWorkflowId.put(jobName, workflow);
+            } else {
+              String workflowId = jobNameToWorkflowId.get(jobName);
+              log.warn("JobName {} previously found to have WorkflowId {}; found " + " a different WorkflowId {} for the job; Skipping this entry", jobName, workflowId);
 
 Review comment:
   missing the argument for the current "different" workflow id in the log line. 
 
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Issue Time Tracking
-------------------

    Worklog Id:     (was: 252780)
    Time Spent: 40m  (was: 0.5h)

> Add a utility method to return Helix WorflowId given a Gobblin job name 
> ------------------------------------------------------------------------
>
>                 Key: GOBBLIN-776
>                 URL: https://issues.apache.org/jira/browse/GOBBLIN-776
>             Project: Apache Gobblin
>          Issue Type: Improvement
>          Components: gobblin-cluster
>    Affects Versions: 0.15.0
>            Reporter: Sudarshan Vasudevan
>            Assignee: Hung Tran
>            Priority: Major
>             Fix For: 0.15.0
>
>          Time Spent: 40m
>  Remaining Estimate: 0h
>
> This task adds a utility method to return Helix WorflowId given a Gobblin job name. This allows cancelling Gobblin jobs running on a Gobblin Helix cluster using a Spec containing the job name, instead of requiring the spec to provide the Helix Workflow Id.  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)