diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/spec/JobExecutionPlan.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/spec/JobExecutionPlan.java
index 8a6f5e1d496..2e2dbb5ee16 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/spec/JobExecutionPlan.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/spec/JobExecutionPlan.java
@@ -61,7 +61,7 @@
 public class JobExecutionPlan {
   public static final String JOB_MAX_ATTEMPTS = "job.maxAttempts";
   public static final String JOB_PROPS_KEY = "job.props";
-  private static final int MAX_JOB_NAME_LENGTH = 255;
+  private static final int MAX_JOB_NAME_LENGTH = 128;
 
   private final JobSpec jobSpec;
   private final SpecExecutor specExecutor;
@@ -112,10 +112,10 @@ private static JobSpec buildJobSpec(FlowSpec flowSpec, Config jobConfig, Long fl
       // job names are assumed to be unique within a dag.
       int hash = flowInputPath.hashCode();
       jobName = Joiner.on(JOB_NAME_COMPONENT_SEPARATION_CHAR).join(flowGroup, flowName, jobName, edgeId, hash);
-      // jobNames are commonly used as a directory name, which is limited to 255 characters
+      // jobNames are commonly used as a directory name, which is limited to 255 characters (account for potential prefixes added/file name lengths)
       if (jobName.length() >= MAX_JOB_NAME_LENGTH) {
-        // shorten job length to be 128 characters (flowGroup) + (hashed) flowName, hashCode length
-        jobName = Joiner.on(JOB_NAME_COMPONENT_SEPARATION_CHAR).join(flowGroup, flowName.hashCode(), hash);
+        // shorten job length but make it uniquely identifiable in multihop flows or concurrent jobs, max length 139 characters (128 flow group  + hash)
+        jobName = Joiner.on(JOB_NAME_COMPONENT_SEPARATION_CHAR).join(flowGroup, jobName.hashCode());
       }
       JobSpec.Builder jobSpecBuilder = JobSpec.builder(jobSpecURIGenerator(flowGroup, jobName, flowSpec)).withConfig(jobConfig)
           .withDescription(flowSpec.getDescription()).withVersion(flowSpec.getVersion());
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/spec/JobExecutionPlanDagFactoryTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/spec/JobExecutionPlanDagFactoryTest.java
index 83cb05f028c..a97b04d2444 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/spec/JobExecutionPlanDagFactoryTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/spec/JobExecutionPlanDagFactoryTest.java
@@ -232,7 +232,8 @@ public void testCreateDagLongName() throws Exception {
 
     Dag<JobExecutionPlan> dag1 = new JobExecutionPlanDagFactory().createDag(Arrays.asList(jobExecutionPlan));
 
-    Assert.assertEquals(dag1.getStartNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY).length(), 142);
+    Assert.assertEquals(dag1.getStartNodes().get(0).getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY).length(), 139);
+
   }
 
   @Test