EmrDaoImpl implements EmrDao { @Override public synchronized ClusterSummary getActiveEmrClusterByNameAndAccountId(String clusterName, String accountId, AwsParamsDto awsParams) { LOGGER.info("Entering synchronized method."); ClusterSummary clusterSummary = null; Map<EmrClusterCacheKey, String> emrClusterCache = getEmrClusterCacheByAccountId(accountId); LOGGER.debug("EMR cluster cache retrieved. emrClusterCache=\"{}\" emrClusterCacheContents=\"{}\"", System.identityHashCode(emrClusterCache), emrClusterCache.toString()); if (StringUtils.isNotBlank(clusterName)) { EmrClusterCacheKey emrClusterCacheKey = new EmrClusterCacheKey(clusterName.toUpperCase(), accountId); LOGGER.info("EMR cluster cache key. emrClusterCacheKey=\"{}\"", emrClusterCacheKey.toString()); if (emrClusterCache.containsKey(emrClusterCacheKey)) { String clusterId = emrClusterCache.get(emrClusterCacheKey); Cluster cluster = getEmrClusterById(clusterId, awsParams); ClusterStatus clusterStatus = cluster == null ? null : cluster.getStatus(); String status = clusterStatus == null ? null : clusterStatus.getState(); LOGGER.info("Found the EMR cluster name in the EMR cluster cache. emrClusterName=\"{}\" emrClusterId=\"{}\" emrClusterStatus=\"{}\"", clusterName.toUpperCase(), clusterId, status); if (status != null && Arrays.asList(getActiveEmrClusterStates()).contains(status)) { LOGGER.info("Exiting synchronized block."); return new ClusterSummary().withId(clusterId).withName(clusterName).withStatus(clusterStatus); } else { LOGGER.info("Removing cluster from EMR cluster cache. emrClusterName=\"{}\" emrClusterId=\"{}\" emrClusterStatus=\"{}\"", clusterName.toUpperCase(), clusterId, status); emrClusterCache.remove(emrClusterCacheKey); } } LOGGER.info("The cluster name was not in the cluster cache. Make a list cluster request to find the cluster id. emrClusterName=\"{}\"", clusterName.toUpperCase()); EmrClusterCacheTimestamps emrClusterCacheTimestamps = getEmrClusterCacheTimestampsByAccountId(accountId); LocalDateTime lastFullReload = emrClusterCacheTimestamps.getLastFullReload(); LocalDateTime lastDeltaUpdate = emrClusterCacheTimestamps.getLastDeltaUpdate(); LocalDateTime newLastFullReload; LocalDateTime newLastDeltaUpdate; Date createdAfter = null; if (lastDeltaUpdate == null || lastFullReload == null || Duration.between(lastFullReload, LocalDateTime.now(ZoneId.systemDefault())).toMinutes() > FULL_RELOAD_CACHE_TIME_PERIOD_IN_MINUTES) { newLastFullReload = LocalDateTime.now(ZoneId.systemDefault()); emrClusterCache.clear(); LOGGER.info("EMR cluster cache cleared. Starting a full reload of the EMR cluster cache. " + "newLastFullReload=\"{}\" lastFullReload=\"{}\" emrClusterCache=\"{}\"", newLastFullReload, lastFullReload, System.identityHashCode(emrClusterCache)); } else { createdAfter = Date.from(lastDeltaUpdate.minusMinutes(DELTA_UPDATE_BUFFER_IN_MINUTES).atZone(ZoneId.systemDefault()).toInstant()); newLastFullReload = lastFullReload; LOGGER.info("Beginning a delta reload of the EMR cluster cache. lastFullReload=\"{}\" lastDeltaUpdate=\"{}\"", lastFullReload, lastDeltaUpdate); } newLastDeltaUpdate = LocalDateTime.now(ZoneId.systemDefault()); LOGGER .info("The new last delta update is newLastDeltaUpdate=\"{}\" and the created after is createdAfter=\"{}\"", newLastDeltaUpdate, createdAfter); ListClustersRequest listClustersRequest = new ListClustersRequest().withClusterStates(getActiveEmrClusterStates()); String markerForListClusters = listClustersRequest.getMarker(); do { ListClustersResult clusterResult = emrOperations .listEmrClusters(getEmrClient(awsParams), listClustersRequest.withMarker(markerForListClusters).withCreatedAfter(createdAfter)); for (ClusterSummary clusterInstance : clusterResult.getClusters()) { LOGGER .info("Adding EMR cluster to the EMR Cluster Cache. emrClusterName=\"{}\" emrClusterId=\"{}\"", clusterInstance.getName().toUpperCase(), clusterInstance.getId()); emrClusterCache.put(new EmrClusterCacheKey(clusterInstance.getName().toUpperCase(), accountId), clusterInstance.getId()); if (StringUtils.isNotBlank(clusterInstance.getName()) && clusterInstance.getName().equalsIgnoreCase(clusterName)) { clusterSummary = clusterInstance; } } markerForListClusters = clusterResult.getMarker(); } while (markerForListClusters != null); emrClusterCacheTimestamps.setLastFullReload(newLastFullReload); emrClusterCacheTimestamps.setLastDeltaUpdate(newLastDeltaUpdate); } LOGGER.info("Returning clusterSummary=\"{}\"", clusterSummary == null ? null : clusterSummary.toString()); LOGGER.debug("State of cache after calling getActiveEmrClusterByNameAndAccountId. emrClusterCache=\"{}\" emrClusterCacheContents=\"{}\"", System.identityHashCode(emrClusterCache), emrClusterCache.toString()); LOGGER.info("Exiting synchronized method."); return clusterSummary; }  @Override String addEmrStep(String clusterId, StepConfig emrStepConfig, AwsParamsDto awsParamsDto); @Override String createEmrCluster(String clusterName, EmrClusterDefinition emrClusterDefinition, EmrParamsDto emrParamsDto); @Override synchronized ClusterSummary getActiveEmrClusterByNameAndAccountId(String clusterName, String accountId, AwsParamsDto awsParams); List<Application> getApplications(List<EmrClusterDefinitionApplication> emrClusterDefinitionApplications); @Override StepSummary getClusterActiveStep(String clusterId, AwsParamsDto awsParamsDto); @Override Step getClusterStep(String clusterId, String stepId, AwsParamsDto awsParamsDto); @Override AmazonElasticMapReduceClient getEmrClient(AwsParamsDto awsParamsDto); @Override Cluster getEmrClusterById(String clusterId, AwsParamsDto awsParams); @Override String getEmrClusterStatusById(String clusterId, AwsParamsDto awsParams); @Override Instance getEmrMasterInstance(String clusterId, AwsParamsDto awsParams); @Override ListInstanceFleetsResult getListInstanceFleetsResult(String clusterId, AwsParamsDto awsParams); @Override void terminateEmrCluster(String clusterId, boolean overrideTerminationProtection, AwsParamsDto awsParams);  }
@Test public void testGetActiveEmrClusterByNameWhenClusterNameIsBlank() { AwsParamsDto awsParamsDto = new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT, AWS_REGION_NAME_US_EAST_1); ClusterSummary result = emrDaoImpl.getActiveEmrClusterByNameAndAccountId(BLANK_TEXT, null, awsParamsDto); verifyNoMoreInteractionsHelper(); assertNull(result); }