class StoragePolicySelectorServiceImpl implements StoragePolicySelectorService {

    @Override
    public List<StoragePolicySelection> execute(String sqsQueueName, int maxResult)
    {
        // Create a result list.
        List<StoragePolicySelection> storagePolicySelections = new ArrayList<>();

        // Get the current timestamp from the database.
        Timestamp currentTimestamp = herdDao.getCurrentTimestamp();

        // Get the threshold in days since business object data registration update for business object data to be selectable
        // by a storage policy with DAYS_SINCE_BDATA_PRIMARY_PARTITION_VALUE storage policy rule type.
        int updatedOnThresholdInDays =
            herdStringHelper.getConfigurationValueAsInteger(ConfigurationValue.STORAGE_POLICY_PROCESSOR_BDATA_UPDATED_ON_THRESHOLD_DAYS);

        // Get the maximum number of failed storage policy transition attempts before the relative storage unit gets excluded from
        // being selected per storage policies by the storage policy selector system job. 0 means the maximum is not set.
        int maxAllowedTransitionAttempts = herdStringHelper.getConfigurationValueAsInteger(ConfigurationValue.STORAGE_POLICY_TRANSITION_MAX_ALLOWED_ATTEMPTS);

        LOGGER.info("{}={} {}={}", ConfigurationValue.STORAGE_POLICY_PROCESSOR_BDATA_UPDATED_ON_THRESHOLD_DAYS.getKey(), updatedOnThresholdInDays,
            ConfigurationValue.STORAGE_POLICY_TRANSITION_MAX_ALLOWED_ATTEMPTS.getKey(), maxAllowedTransitionAttempts);

        // Compute business object data "updated on" threshold timestamp based on
        // the current database timestamp and the threshold value configured in the system.
        Timestamp updatedOnThresholdTimestamp = HerdDateUtils.addDays(currentTimestamp, -updatedOnThresholdInDays);

        // Keep track of all business object data entities selected per storage policies. This is need to avoid a lower priority selection policy
        // to be executed ahead of a higher priority one.
        Set<BusinessObjectDataEntity> selectedBusinessObjectDataEntities = new LinkedHashSet<>();

        // Separately process all possible storage policy priority levels in order of priorities. This is done to assure that higher priority level storage
        // policies will be listed earlier in the final result map.
        for (StoragePolicyPriorityLevel storagePolicyPriorityLevel : STORAGE_POLICY_PRIORITY_LEVELS)
        {
            // Until we reach maximum number of results or run out of entities to select, retrieve and process business object data entities mapped to their
            // corresponding storage policy entities, where the business object data status is supported by the storage policy feature and the business object
            // data alternate key values match storage policy's filter and transition (not taking into account storage policy rules).
            int startPosition = 0;
            while (true)
            {
                Map<BusinessObjectDataEntity, StoragePolicyEntity> map = businessObjectDataDao
                    .getBusinessObjectDataEntitiesMatchingStoragePolicies(storagePolicyPriorityLevel, SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES,
                        maxAllowedTransitionAttempts, startPosition, maxResult);

                for (Map.Entry<BusinessObjectDataEntity, StoragePolicyEntity> entry : map.entrySet())
                {
                    BusinessObjectDataEntity businessObjectDataEntity = entry.getKey();

                    // Process this storage policy selection, only if this business object data has not been selected earlier.
                    if (!selectedBusinessObjectDataEntities.contains(businessObjectDataEntity))
                    {
                        boolean createStoragePolicySelection = false;

                        // Remember that we got this business object data entity as matching to a storage policy.
                        // This is done so we would not try to select this business object data again later by a lower level storage policy.
                        selectedBusinessObjectDataEntities.add(businessObjectDataEntity);

                        // Get the storage policy entity, so we can validate the storage policy rule against this business object data.
                        StoragePolicyEntity storagePolicyEntity = entry.getValue();

                        // Get a storage policy rule type and value.
                        String storagePolicyRuleType = storagePolicyEntity.getStoragePolicyRuleType().getCode();
                        Integer storagePolicyRuleValue = storagePolicyEntity.getStoragePolicyRuleValue();

                        // For DAYS_SINCE_BDATA_REGISTERED storage policy rule type, select business object data based on it's "created on" timestamp.
                        if (StoragePolicyRuleTypeEntity.DAYS_SINCE_BDATA_REGISTERED.equals(storagePolicyRuleType))
                        {
                            // Compute "created on" threshold timestamp based on the current timestamp and storage policy rule value.
                            Timestamp createdOnThresholdTimestamp = HerdDateUtils.addDays(currentTimestamp, -storagePolicyRuleValue);

                            // Select this business object data if it has "created on" timestamp before or equal to the threshold timestamp.
                            createStoragePolicySelection = (businessObjectDataEntity.getCreatedOn().compareTo(createdOnThresholdTimestamp) <= 0);
                        }
                        // For DAYS_SINCE_BDATA_PRIMARY_PARTITION_VALUE storage policy rule type, select business object data based on both it's primary
                        // partition value compared against storage policy rule value and "updated on" timestamp being below the threshold.
                        else if (StoragePolicyRuleTypeEntity.DAYS_SINCE_BDATA_PRIMARY_PARTITION_VALUE.equals(storagePolicyRuleType))
                        {
                            // For this storage policy rule, we ignore this business object data if it was updated earlier than the threshold value of days ago.
                            if (businessObjectDataEntity.getUpdatedOn().compareTo(updatedOnThresholdTimestamp) <= 0)
                            {
                                // Try to convert business object data primary partition value to a timestamp.
                                // If it is not a date, the storage policy rule is not matching this business object data.
                                Date primaryPartitionValue = businessObjectDataHelper.getDateFromString(businessObjectDataEntity.getPartitionValue());

                                // For this storage policy rule, we ignore this business data if primary partition value is not a date.
                                if (primaryPartitionValue != null)
                                {
                                    // Compute the relative primary partition value threshold date based on the current timestamp and storage policy rule value.
                                    Date primaryPartitionValueThreshold = new Date(HerdDateUtils.addDays(currentTimestamp, -storagePolicyRuleValue).getTime());

                                    // Select this business object data if it has it's primary partition value before or equal to the threshold date.
                                    createStoragePolicySelection = (primaryPartitionValue.compareTo(primaryPartitionValueThreshold) <= 0);
                                }
                            }
                        }
                        // Fail on an un-supported storage policy rule type.
                        else
                        {
                            throw new IllegalStateException(String.format("Storage policy type \"%s\" is not supported.", storagePolicyRuleType));
                        }

                        // If this business object data got selected, create a storage policy selection and add it to the result list.
                        if (createStoragePolicySelection)
                        {
                            // Create business object data key and storage policy key per selected entities.
                            BusinessObjectDataKey businessObjectDataKey = businessObjectDataHelper.getBusinessObjectDataKey(businessObjectDataEntity);
                            StoragePolicyKey storagePolicyKey =
                                new StoragePolicyKey(storagePolicyEntity.getNamespace().getCode(), storagePolicyEntity.getName());

                            // Create and add a storage policy selection to the result list.
                            storagePolicySelections.add(new StoragePolicySelection(businessObjectDataKey, storagePolicyKey, storagePolicyEntity.getVersion()));

                            LOGGER.info("Selected business object data for storage policy processing: " +
                                "businessObjectDataKey={} storagePolicyKey={} storagePolicyVersion={}", jsonHelper.objectToJson(businessObjectDataKey),
                                jsonHelper.objectToJson(storagePolicyKey), storagePolicyEntity.getVersion());

                            // Stop adding storage policy selections to the result list if we reached the maximum results limit.
                            if (storagePolicySelections.size() >= maxResult)
                            {
                                break;
                            }
                        }
                    }
                }

                // Stop processing storage policies if we reached the max result limit or there are no more business object data to select.
                if (storagePolicySelections.size() >= maxResult || map.isEmpty())
                {
                    break;
                }

                // Increment start position for the next select.
                startPosition += maxResult;
            }

            // Stop processing storage policies if we reached the max result limit.
            if (storagePolicySelections.size() >= maxResult)
            {
                break;
            }
        }

        // Send all storage policy selections to the specified SQS queue.
        sendStoragePolicySelectionToSqsQueue(sqsQueueName, storagePolicySelections);

        return storagePolicySelections;
    }

}