class EmrDaoImpl implements EmrDao {

    @Override
    public synchronized ClusterSummary getActiveEmrClusterByNameAndAccountId(String clusterName, String accountId, AwsParamsDto awsParams)
    {
        LOGGER.info("Entering synchronized method.");

        // Initialize a cluster summary to null for the case that the cluster is not found in the list.
        ClusterSummary clusterSummary = null;

        // Get the cluster cache using the accountId.
        Map<EmrClusterCacheKey, String> emrClusterCache = getEmrClusterCacheByAccountId(accountId);

        LOGGER.debug("EMR cluster cache retrieved. emrClusterCache=\"{}\" emrClusterCacheContents=\"{}\"", System.identityHashCode(emrClusterCache),
            emrClusterCache.toString());

        if (StringUtils.isNotBlank(clusterName))
        {
            // Build the EMR cluster cache key
            EmrClusterCacheKey emrClusterCacheKey = new EmrClusterCacheKey(clusterName.toUpperCase(), accountId);

            LOGGER.info("EMR cluster cache key. emrClusterCacheKey=\"{}\"", emrClusterCacheKey.toString());

            // First check to see if this cluster id is stored locally in the EMR Cluster Cache.
            // If the EMR cluster cache does contain the cluster key use the id found in the EMR cluster cache.
            // Else the EMR cluster cache does not contain the cluster key then move on to do a list cluster.
            if (emrClusterCache.containsKey(emrClusterCacheKey))
            {
                // Get the cluster id value from the EMR cluster cache with the cluster name key.
                String clusterId = emrClusterCache.get(emrClusterCacheKey);

                // Retrieve the cluster status to validate the cluster.
                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 the status is not null and the status is in one of the active EMR cluster states,
                // then return the cluster summary with the cluster id from the EMR cluster cache.
                // Else remove the cluster from the EMR cluster cache and then move on to do a list cluster.
                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);

                    // Remove the cluster from the cache.
                    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());

            // Get the EMR cluster cache timeout values.
            EmrClusterCacheTimestamps emrClusterCacheTimestamps = getEmrClusterCacheTimestampsByAccountId(accountId);
            LocalDateTime lastFullReload = emrClusterCacheTimestamps.getLastFullReload();
            LocalDateTime lastDeltaUpdate = emrClusterCacheTimestamps.getLastDeltaUpdate();

            // New cache timeout values.
            LocalDateTime newLastFullReload;
            LocalDateTime newLastDeltaUpdate;

            // Default the created after date to null for the full update case.
            Date createdAfter = null;

            // If the last delta update is null, or the last full reload is null, or the if the difference between the current time and the lastFullReload is
            // greater than FULL_RELOAD_CACHE_TIME_PERIOD_IN_MINUTES, then do a full reload.
            if (lastDeltaUpdate == null || lastFullReload == null ||
                Duration.between(lastFullReload, LocalDateTime.now(ZoneId.systemDefault())).toMinutes() > FULL_RELOAD_CACHE_TIME_PERIOD_IN_MINUTES)
            {
                // Set the new last full reload time to the current time.
                newLastFullReload = LocalDateTime.now(ZoneId.systemDefault());

                // Clear the EMR cluster cache
                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
            {
                // Set the created after date to the last delta update minus the delta update safety buffer time.
                createdAfter = Date.from(lastDeltaUpdate.minusMinutes(DELTA_UPDATE_BUFFER_IN_MINUTES).atZone(ZoneId.systemDefault()).toInstant());

                // Keep the last full reload the same.
                newLastFullReload = lastFullReload;

                LOGGER.info("Beginning a delta reload of the EMR cluster cache. lastFullReload=\"{}\" lastDeltaUpdate=\"{}\"", lastFullReload, lastDeltaUpdate);
            }

            // Set the new last delta update to the current time.
            newLastDeltaUpdate = LocalDateTime.now(ZoneId.systemDefault());

            LOGGER
                .info("The new last delta update is newLastDeltaUpdate=\"{}\" and the created after is createdAfter=\"{}\"", newLastDeltaUpdate, createdAfter);

            /**
             * Call AWSOperations for ListClusters API. Need to list all the active clusters that are in
             * BOOTSTRAPPING/RUNNING/STARTING/WAITING states
             */
            ListClustersRequest listClustersRequest = new ListClustersRequest().withClusterStates(getActiveEmrClusterStates());

            /**
             * ListClusterRequest returns only 50 clusters at a time. However, this returns a marker
             * that can be used for subsequent calls to listClusters to get all the clusters
             */
            String markerForListClusters = listClustersRequest.getMarker();

            // Loop through all the available clusters and look for the given cluster id
            do
            {
                /**
                 * Call AWSOperations for ListClusters API.
                 * Need to include the Marker returned by the previous iteration
                 */
                ListClustersResult clusterResult = emrOperations
                    .listEmrClusters(getEmrClient(awsParams), listClustersRequest.withMarker(markerForListClusters).withCreatedAfter(createdAfter));

                // Loop through all the active clusters returned by AWS
                for (ClusterSummary clusterInstance : clusterResult.getClusters())
                {
                    LOGGER
                        .info("Adding EMR cluster to the EMR Cluster Cache. emrClusterName=\"{}\" emrClusterId=\"{}\"", clusterInstance.getName().toUpperCase(),
                            clusterInstance.getId());

                    // Add this cluster instance to the EMR cluster cache.
                    emrClusterCache.put(new EmrClusterCacheKey(clusterInstance.getName().toUpperCase(), accountId), clusterInstance.getId());

                    // If the cluster name matches, then set the clusterSummary to the clusterInstance
                    if (StringUtils.isNotBlank(clusterInstance.getName()) && clusterInstance.getName().equalsIgnoreCase(clusterName))
                    {
                        clusterSummary = clusterInstance;
                    }
                }
                markerForListClusters = clusterResult.getMarker();
            }
            while (markerForListClusters != null);

            // Update the cluster cache timestamps
            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;
    }

}