@Test public void versionNotFoundTest() { String clusterName = "versionNotFoundTest"; List<Label> labels = new ArrayList<>(); Label label = new Label("l1", "lv1"); labels.add(label); List<Parameter> parameters = new ArrayList<>(); Parameter parameter = new Parameter("paramName", "paramValue"); parameters.add(parameter); final JobClusterDefinitionImpl fakeJobCluster = createFakeJobClusterDefn(clusterName, labels, parameters); IJobClusterMetadata jobClusterMetadata = new JobClusterMetadataImpl(fakeJobCluster,1,false); String version = "0.0.2"; try { JobDefinition givenJobDefn = new JobDefinition.Builder().withName(clusterName).withVersion(version).build(); JobDefinitionResolver resolver = new JobDefinitionResolver(); JobDefinition resolvedJobDefinition = resolver.getResolvedJobDefinition("user", givenJobDefn, jobClusterMetadata); fail(); } catch (Exception e) { e.printStackTrace(); } }
JobDefinition getResolvedJobDefinition(final String user, final JobDefinition givenJobDefnOp, final IJobClusterMetadata jobClusterMetadata) throws Exception { Preconditions.checkNotNull(givenJobDefnOp, "JobDefinition cannot be null"); Preconditions.checkNotNull(jobClusterMetadata, "JobClusterMetadata cannot be null"); JobDefinition resolvedJobDefn = givenJobDefnOp; logger.info("Given JobDefn {}", resolvedJobDefn); List<Parameter> parameters = (resolvedJobDefn.getParameters() != null && !resolvedJobDefn.getParameters().isEmpty()) ? resolvedJobDefn.getParameters() : jobClusterMetadata.getJobClusterDefinition().getParameters(); List<Label> labels = (resolvedJobDefn.getLabels() != null && !resolvedJobDefn.getLabels().isEmpty()) ? resolvedJobDefn.getLabels() : jobClusterMetadata.getJobClusterDefinition().getLabels(); String artifactName = resolvedJobDefn.getArtifactName(); SchedulingInfo schedulingInfo = resolvedJobDefn.getSchedulingInfo(); String version = resolvedJobDefn.getVersion(); JobClusterConfig jobClusterConfig = null; if(!isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { } else if(!isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact is specified %s. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(!isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { version = String.valueOf(System.currentTimeMillis()); } else if(!isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact %s is specified. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); artifactName = jobClusterConfig.getArtifactName(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s which is %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName(), jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig().getSchedulingInfo()); logger.warn(msg); throw new Exception(msg); } } else if(isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)){ jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else { throw new Exception(String.format("Invalid case for resolveJobDefinition artifactName %s version %s schedulingInfo %s", artifactName, version, schedulingInfo)); } logger.info("Resolved version {}, schedulingInfo {}, artifactName {}", version, schedulingInfo, artifactName); if(isNull(artifactName) || isNull(version) || schedulingInfoNotValid(schedulingInfo)) { String msg = String.format(" SchedulingInfo %s or artifact %s or version %s could not be resolved in JobCluster %s. Job Submit fails", schedulingInfo, artifactName, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } return new JobDefinition.Builder() .from(resolvedJobDefn) .withParameters(parameters) .withLabels(labels) .withSchedulingInfo(schedulingInfo) .withUser(user) .withVersion(version) .withArtifactName(artifactName) .build(); }
JobDefinitionResolver { JobDefinition getResolvedJobDefinition(final String user, final JobDefinition givenJobDefnOp, final IJobClusterMetadata jobClusterMetadata) throws Exception { Preconditions.checkNotNull(givenJobDefnOp, "JobDefinition cannot be null"); Preconditions.checkNotNull(jobClusterMetadata, "JobClusterMetadata cannot be null"); JobDefinition resolvedJobDefn = givenJobDefnOp; logger.info("Given JobDefn {}", resolvedJobDefn); List<Parameter> parameters = (resolvedJobDefn.getParameters() != null && !resolvedJobDefn.getParameters().isEmpty()) ? resolvedJobDefn.getParameters() : jobClusterMetadata.getJobClusterDefinition().getParameters(); List<Label> labels = (resolvedJobDefn.getLabels() != null && !resolvedJobDefn.getLabels().isEmpty()) ? resolvedJobDefn.getLabels() : jobClusterMetadata.getJobClusterDefinition().getLabels(); String artifactName = resolvedJobDefn.getArtifactName(); SchedulingInfo schedulingInfo = resolvedJobDefn.getSchedulingInfo(); String version = resolvedJobDefn.getVersion(); JobClusterConfig jobClusterConfig = null; if(!isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { } else if(!isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact is specified %s. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(!isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { version = String.valueOf(System.currentTimeMillis()); } else if(!isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact %s is specified. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); artifactName = jobClusterConfig.getArtifactName(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s which is %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName(), jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig().getSchedulingInfo()); logger.warn(msg); throw new Exception(msg); } } else if(isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)){ jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else { throw new Exception(String.format("Invalid case for resolveJobDefinition artifactName %s version %s schedulingInfo %s", artifactName, version, schedulingInfo)); } logger.info("Resolved version {}, schedulingInfo {}, artifactName {}", version, schedulingInfo, artifactName); if(isNull(artifactName) || isNull(version) || schedulingInfoNotValid(schedulingInfo)) { String msg = String.format(" SchedulingInfo %s or artifact %s or version %s could not be resolved in JobCluster %s. Job Submit fails", schedulingInfo, artifactName, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } return new JobDefinition.Builder() .from(resolvedJobDefn) .withParameters(parameters) .withLabels(labels) .withSchedulingInfo(schedulingInfo) .withUser(user) .withVersion(version) .withArtifactName(artifactName) .build(); } }
JobDefinitionResolver { JobDefinition getResolvedJobDefinition(final String user, final JobDefinition givenJobDefnOp, final IJobClusterMetadata jobClusterMetadata) throws Exception { Preconditions.checkNotNull(givenJobDefnOp, "JobDefinition cannot be null"); Preconditions.checkNotNull(jobClusterMetadata, "JobClusterMetadata cannot be null"); JobDefinition resolvedJobDefn = givenJobDefnOp; logger.info("Given JobDefn {}", resolvedJobDefn); List<Parameter> parameters = (resolvedJobDefn.getParameters() != null && !resolvedJobDefn.getParameters().isEmpty()) ? resolvedJobDefn.getParameters() : jobClusterMetadata.getJobClusterDefinition().getParameters(); List<Label> labels = (resolvedJobDefn.getLabels() != null && !resolvedJobDefn.getLabels().isEmpty()) ? resolvedJobDefn.getLabels() : jobClusterMetadata.getJobClusterDefinition().getLabels(); String artifactName = resolvedJobDefn.getArtifactName(); SchedulingInfo schedulingInfo = resolvedJobDefn.getSchedulingInfo(); String version = resolvedJobDefn.getVersion(); JobClusterConfig jobClusterConfig = null; if(!isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { } else if(!isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact is specified %s. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(!isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { version = String.valueOf(System.currentTimeMillis()); } else if(!isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact %s is specified. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); artifactName = jobClusterConfig.getArtifactName(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s which is %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName(), jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig().getSchedulingInfo()); logger.warn(msg); throw new Exception(msg); } } else if(isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)){ jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else { throw new Exception(String.format("Invalid case for resolveJobDefinition artifactName %s version %s schedulingInfo %s", artifactName, version, schedulingInfo)); } logger.info("Resolved version {}, schedulingInfo {}, artifactName {}", version, schedulingInfo, artifactName); if(isNull(artifactName) || isNull(version) || schedulingInfoNotValid(schedulingInfo)) { String msg = String.format(" SchedulingInfo %s or artifact %s or version %s could not be resolved in JobCluster %s. Job Submit fails", schedulingInfo, artifactName, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } return new JobDefinition.Builder() .from(resolvedJobDefn) .withParameters(parameters) .withLabels(labels) .withSchedulingInfo(schedulingInfo) .withUser(user) .withVersion(version) .withArtifactName(artifactName) .build(); }  }
JobDefinitionResolver { JobDefinition getResolvedJobDefinition(final String user, final JobDefinition givenJobDefnOp, final IJobClusterMetadata jobClusterMetadata) throws Exception { Preconditions.checkNotNull(givenJobDefnOp, "JobDefinition cannot be null"); Preconditions.checkNotNull(jobClusterMetadata, "JobClusterMetadata cannot be null"); JobDefinition resolvedJobDefn = givenJobDefnOp; logger.info("Given JobDefn {}", resolvedJobDefn); List<Parameter> parameters = (resolvedJobDefn.getParameters() != null && !resolvedJobDefn.getParameters().isEmpty()) ? resolvedJobDefn.getParameters() : jobClusterMetadata.getJobClusterDefinition().getParameters(); List<Label> labels = (resolvedJobDefn.getLabels() != null && !resolvedJobDefn.getLabels().isEmpty()) ? resolvedJobDefn.getLabels() : jobClusterMetadata.getJobClusterDefinition().getLabels(); String artifactName = resolvedJobDefn.getArtifactName(); SchedulingInfo schedulingInfo = resolvedJobDefn.getSchedulingInfo(); String version = resolvedJobDefn.getVersion(); JobClusterConfig jobClusterConfig = null; if(!isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { } else if(!isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact is specified %s. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(!isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { version = String.valueOf(System.currentTimeMillis()); } else if(!isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact %s is specified. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); artifactName = jobClusterConfig.getArtifactName(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s which is %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName(), jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig().getSchedulingInfo()); logger.warn(msg); throw new Exception(msg); } } else if(isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)){ jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else { throw new Exception(String.format("Invalid case for resolveJobDefinition artifactName %s version %s schedulingInfo %s", artifactName, version, schedulingInfo)); } logger.info("Resolved version {}, schedulingInfo {}, artifactName {}", version, schedulingInfo, artifactName); if(isNull(artifactName) || isNull(version) || schedulingInfoNotValid(schedulingInfo)) { String msg = String.format(" SchedulingInfo %s or artifact %s or version %s could not be resolved in JobCluster %s. Job Submit fails", schedulingInfo, artifactName, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } return new JobDefinition.Builder() .from(resolvedJobDefn) .withParameters(parameters) .withLabels(labels) .withSchedulingInfo(schedulingInfo) .withUser(user) .withVersion(version) .withArtifactName(artifactName) .build(); }   }
JobDefinitionResolver { JobDefinition getResolvedJobDefinition(final String user, final JobDefinition givenJobDefnOp, final IJobClusterMetadata jobClusterMetadata) throws Exception { Preconditions.checkNotNull(givenJobDefnOp, "JobDefinition cannot be null"); Preconditions.checkNotNull(jobClusterMetadata, "JobClusterMetadata cannot be null"); JobDefinition resolvedJobDefn = givenJobDefnOp; logger.info("Given JobDefn {}", resolvedJobDefn); List<Parameter> parameters = (resolvedJobDefn.getParameters() != null && !resolvedJobDefn.getParameters().isEmpty()) ? resolvedJobDefn.getParameters() : jobClusterMetadata.getJobClusterDefinition().getParameters(); List<Label> labels = (resolvedJobDefn.getLabels() != null && !resolvedJobDefn.getLabels().isEmpty()) ? resolvedJobDefn.getLabels() : jobClusterMetadata.getJobClusterDefinition().getLabels(); String artifactName = resolvedJobDefn.getArtifactName(); SchedulingInfo schedulingInfo = resolvedJobDefn.getSchedulingInfo(); String version = resolvedJobDefn.getVersion(); JobClusterConfig jobClusterConfig = null; if(!isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { } else if(!isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact is specified %s. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(!isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { version = String.valueOf(System.currentTimeMillis()); } else if(!isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)) { String msg = String.format("Scheduling info is not specified during Job Submit for cluster %s while new artifact %s is specified. Job Submit fails", jobClusterMetadata.getJobClusterDefinition().getName(), artifactName); logger.warn(msg); throw new Exception(msg); } else if(isNull(artifactName) && !isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && !isNull(version) && schedulingInfoNotValid(schedulingInfo)) { Optional<JobClusterConfig> clusterConfigForVersion = getJobClusterConfigForVersion(jobClusterMetadata, version); if(!clusterConfigForVersion.isPresent()) { String msg = String.format("No Job Cluster config could be found for version %s in JobCluster %s. Job Submit fails", version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } jobClusterConfig = clusterConfigForVersion.get(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else if(isNull(artifactName) && isNull(version) && !schedulingInfoNotValid(schedulingInfo)) { jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); artifactName = jobClusterConfig.getArtifactName(); if(!validateSchedulingInfo(schedulingInfo, jobClusterConfig.getSchedulingInfo(), jobClusterMetadata)) { String msg = String.format("Given SchedulingInfo %s is incompatible with that associated with the given version %s in JobCluster %s which is %s. Job Submit fails", schedulingInfo, version, jobClusterMetadata.getJobClusterDefinition().getName(), jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig().getSchedulingInfo()); logger.warn(msg); throw new Exception(msg); } } else if(isNull(artifactName) && isNull(version) && schedulingInfoNotValid(schedulingInfo)){ jobClusterConfig = jobClusterMetadata.getJobClusterDefinition().getJobClusterConfig(); version = jobClusterConfig.getVersion(); schedulingInfo = jobClusterConfig.getSchedulingInfo(); artifactName = jobClusterConfig.getArtifactName(); } else { throw new Exception(String.format("Invalid case for resolveJobDefinition artifactName %s version %s schedulingInfo %s", artifactName, version, schedulingInfo)); } logger.info("Resolved version {}, schedulingInfo {}, artifactName {}", version, schedulingInfo, artifactName); if(isNull(artifactName) || isNull(version) || schedulingInfoNotValid(schedulingInfo)) { String msg = String.format(" SchedulingInfo %s or artifact %s or version %s could not be resolved in JobCluster %s. Job Submit fails", schedulingInfo, artifactName, version, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); throw new Exception(msg); } return new JobDefinition.Builder() .from(resolvedJobDefn) .withParameters(parameters) .withLabels(labels) .withSchedulingInfo(schedulingInfo) .withUser(user) .withVersion(version) .withArtifactName(artifactName) .build(); }    }