25
25
26
26
import org .apache .commons .lang3 .StringUtils ;
27
27
28
- import com .google .common .base .Joiner ;
29
28
import com .google .common .base .Optional ;
30
29
import com .google .common .base .Strings ;
31
30
import com .typesafe .config .Config ;
46
45
import org .apache .gobblin .runtime .api .SpecExecutor ;
47
46
import org .apache .gobblin .service .ExecutionStatus ;
48
47
import org .apache .gobblin .service .modules .flowgraph .DagNodeId ;
49
- import org .apache .gobblin .service .modules .flowgraph .DatasetDescriptorConfigKeys ;
50
48
import org .apache .gobblin .service .modules .flowgraph .FlowGraphConfigurationKeys ;
51
49
import org .apache .gobblin .service .modules .orchestration .DagProcessingEngine ;
52
50
import org .apache .gobblin .service .modules .orchestration .DagUtils ;
@@ -111,8 +109,6 @@ private static JobSpec buildJobSpec(FlowSpec flowSpec, Config jobConfig, Long fl
111
109
String flowName = ConfigUtils .getString (flowConfig , ConfigurationKeys .FLOW_NAME_KEY , "" );
112
110
String flowGroup = ConfigUtils .getString (flowConfig , ConfigurationKeys .FLOW_GROUP_KEY , "" );
113
111
String flowFailureOption = ConfigUtils .getString (flowConfig , ConfigurationKeys .FLOW_FAILURE_OPTION , DagProcessingEngine .DEFAULT_FLOW_FAILURE_OPTION );
114
- String flowInputPath = ConfigUtils .getString (flowConfig , DatasetDescriptorConfigKeys .FLOW_INPUT_DATASET_DESCRIPTOR_PREFIX
115
- + "." + DatasetDescriptorConfigKeys .PATH_KEY , "" );
116
112
Long flowModTime = ConfigUtils .getLong (flowConfig , FlowSpec .MODIFICATION_TIME_KEY , 0L );
117
113
118
114
String jobName = ConfigUtils .getString (jobConfig , ConfigurationKeys .JOB_NAME_KEY , "" );
@@ -121,15 +117,9 @@ private static JobSpec buildJobSpec(FlowSpec flowSpec, Config jobConfig, Long fl
121
117
final int gaasJobExecutionIdHash = gaasJobExecutionId .hashCode (); // Passing the hashCode of the uniqueIdentifier to be used as jobExecutionId for backward compatibility
122
118
123
119
if (!ConfigUtils .getBoolean (jobConfig , JOB_MAINTAIN_JOBNAME , false ) || jobName .isEmpty ()) {
124
- // Modify the job name to include the flow group, flow name, edge id, and a random string to avoid collisions since
125
- // job names are assumed to be unique within a dag.
126
- int hash = flowInputPath .hashCode ();
127
- jobName = Joiner .on (JOB_NAME_COMPONENT_SEPARATION_CHAR ).join (flowGroup , flowName , jobName , edgeId , hash );
128
- // jobNames are commonly used as a directory name, which is limited to 255 characters
129
- if (jobName .length () >= MAX_JOB_NAME_LENGTH ) {
130
- // shorten job length to be 128 characters (flowGroup) + (hashed) flowName, hashCode length
131
- jobName = Joiner .on (JOB_NAME_COMPONENT_SEPARATION_CHAR ).join (flowGroup , flowName .hashCode (), hash );
132
- }
120
+ jobName = gaasJobExecutionId ; // Assigning jobName with the value of GaaSJobExecutionId
121
+ // which is a UUID and unique to avoid collisions
122
+
133
123
}
134
124
JobSpec .Builder jobSpecBuilder = JobSpec .builder (jobSpecURIGenerator (flowGroup , jobName , flowSpec )).withConfig (jobConfig )
135
125
.withDescription (flowSpec .getDescription ()).withVersion (flowSpec .getVersion ());
0 commit comments