[GOBBLIN-268] Generate unique job name and job uri for GaaS

Closes #2121 from yukuai518/gaas_job
apache · Sep 28, 2017 · ae0ba28 · ae0ba28
1 parent 859fadc
commit ae0ba28
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/...src/main/java/org/apache/gobblin/service/modules/flow/MultiHopsFlowToJobSpecCompiler.java b/...src/main/java/org/apache/gobblin/service/modules/flow/MultiHopsFlowToJobSpecCompiler.java
@@ -289,7 +289,10 @@ private JobSpec buildJobSpec (ServiceNode sourceNode, ServiceNode targetNode, UR
     // Add job.name and job.group
     if (flowSpec.getConfig().hasPath(ConfigurationKeys.FLOW_NAME_KEY)) {
       jobSpec.setConfig(jobSpec.getConfig()
-          .withValue(ConfigurationKeys.JOB_NAME_KEY, flowSpec.getConfig().getValue(ConfigurationKeys.FLOW_NAME_KEY)));
+          .withValue(ConfigurationKeys.JOB_NAME_KEY,  ConfigValueFactory.fromAnyRef(
+              flowSpec.getConfig().getValue(ConfigurationKeys.FLOW_NAME_KEY).unwrapped().toString()
+                  + "-" + sourceNode.getNodeName()
+                  + "-" + targetNode.getNodeName())));
     }
     if (flowSpec.getConfig().hasPath(ConfigurationKeys.FLOW_GROUP_KEY)) {
       jobSpec.setConfig(jobSpec.getConfig()
@@ -329,8 +332,9 @@ private URI getTemplateURI (ServiceNode sourceNode, ServiceNode targetNode, Flow
    */
   public static URI jobSpecURIGenerator(FlowSpec flowSpec, ServiceNode sourceNode, ServiceNode targetNode) {
     try {
-      return new URI(flowSpec.getUri().getScheme(), flowSpec.getUri().getAuthority(),
-          "/" + sourceNode.getNodeName() + "-" + targetNode.getNodeName(), null);
+      return new URI(JobSpec.Builder.DEFAULT_JOB_CATALOG_SCHEME, flowSpec.getUri().getAuthority(),
+          StringUtils.appendIfMissing(StringUtils.prependIfMissing(flowSpec.getUri().getPath(), "/"),"/")
+              + sourceNode.getNodeName() + "-" + targetNode.getNodeName(), null);
     } catch (URISyntaxException e) {
       log.error(
           "URI construction failed when jobSpec from " + sourceNode.getNodeName() + " to " + targetNode.getNodeName());