### Question:
JobDefinitionResolver { Optional<JobClusterConfig> getJobClusterConfigForVersion(final IJobClusterMetadata jobClusterMetadata, final String version) { Preconditions.checkNotNull(jobClusterMetadata, "JobClusterMetadata cannot be null"); Preconditions.checkNotNull(version, "Version cannot be null"); final String versionToFind = version; List<JobClusterConfig> configList = jobClusterMetadata.getJobClusterDefinition() .getJobClusterConfigs() .stream() .filter((cfg) -> cfg.getVersion().equals(versionToFind)) .collect(Collectors.toList()); if(!configList.isEmpty()) { return of(configList.get(0)); } else { String msg = String.format("No config with version %s found for Job Cluster %s. Job Submit fails", versionToFind, jobClusterMetadata.getJobClusterDefinition().getName()); logger.warn(msg); return empty(); } }    }### Answer:
@Test public void lookupJobClusterConfigTest() { String clusterName = "lookupJobClusterConfigTest"; JobClusterConfig clusterConfig1 = new JobClusterConfig.Builder() .withArtifactName(DEFAULT_ARTIFACT_NAME) .withSchedulingInfo(SINGLE_WORKER_SCHED_INFO) .withVersion(DEFAULT_VERSION) .build(); JobClusterConfig clusterConfig2 = new JobClusterConfig.Builder() .withArtifactName("artifact2") .withSchedulingInfo(TWO_WORKER_SCHED_INFO) .withVersion("0.0.2") .build(); List<JobClusterConfig> configList = new ArrayList<>(); configList.add(clusterConfig1); configList.add(clusterConfig2); JobClusterDefinitionImpl jobClusterDefinition = new JobClusterDefinitionImpl.Builder() .withJobClusterConfigs(configList) .withName(clusterName) .withParameters(Lists.newArrayList()) .withLabels(Lists.newArrayList()) .withUser("user") .withIsReadyForJobMaster(true) .withOwner(DEFAULT_JOB_OWNER) .withMigrationConfig(WorkerMigrationConfig.DEFAULT) .withSla(NO_OP_SLA) .build(); IJobClusterMetadata jobClusterMetadata = new JobClusterMetadataImpl.Builder().withJobClusterDefinition(jobClusterDefinition).withLastJobCount(1).withIsDisabled(false).build(); JobDefinitionResolver resolver = new JobDefinitionResolver(); Optional<JobClusterConfig> config = resolver.getJobClusterConfigForVersion(jobClusterMetadata, DEFAULT_VERSION); assertTrue(config.isPresent()); assertEquals(DEFAULT_ARTIFACT_NAME, config.get().getArtifactName()); assertEquals(DEFAULT_VERSION, config.get().getVersion()); assertEquals(SINGLE_WORKER_SCHED_INFO, config.get().getSchedulingInfo()); Optional<JobClusterConfig> config2 = resolver.getJobClusterConfigForVersion(jobClusterMetadata, "0.0.2"); assertTrue(config2.isPresent()); assertEquals("artifact2", config2.get().getArtifactName()); assertEquals("0.0.2", config2.get().getVersion()); assertEquals(TWO_WORKER_SCHED_INFO, config2.get().getSchedulingInfo()); try { Optional<JobClusterConfig> config3 = resolver.getJobClusterConfigForVersion(jobClusterMetadata, "0.0.3"); assertTrue(!config3.isPresent()); } catch(Exception e) { e.printStackTrace(); fail(); } }