StoragePolicySelectorServiceImpl implements StoragePolicySelectorService { @Override public List<StoragePolicySelection> execute(String sqsQueueName, int maxResult) { List<StoragePolicySelection> storagePolicySelections = new ArrayList<>(); Timestamp currentTimestamp = herdDao.getCurrentTimestamp(); int updatedOnThresholdInDays = herdStringHelper.getConfigurationValueAsInteger(ConfigurationValue.STORAGE_POLICY_PROCESSOR_BDATA_UPDATED_ON_THRESHOLD_DAYS); 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); Timestamp updatedOnThresholdTimestamp = HerdDateUtils.addDays(currentTimestamp, -updatedOnThresholdInDays); Set<BusinessObjectDataEntity> selectedBusinessObjectDataEntities = new LinkedHashSet<>(); for (StoragePolicyPriorityLevel storagePolicyPriorityLevel : STORAGE_POLICY_PRIORITY_LEVELS) { 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(); if (!selectedBusinessObjectDataEntities.contains(businessObjectDataEntity)) { boolean createStoragePolicySelection = false; selectedBusinessObjectDataEntities.add(businessObjectDataEntity); StoragePolicyEntity storagePolicyEntity = entry.getValue(); String storagePolicyRuleType = storagePolicyEntity.getStoragePolicyRuleType().getCode(); Integer storagePolicyRuleValue = storagePolicyEntity.getStoragePolicyRuleValue(); if (StoragePolicyRuleTypeEntity.DAYS_SINCE_BDATA_REGISTERED.equals(storagePolicyRuleType)) { Timestamp createdOnThresholdTimestamp = HerdDateUtils.addDays(currentTimestamp, -storagePolicyRuleValue); createStoragePolicySelection = (businessObjectDataEntity.getCreatedOn().compareTo(createdOnThresholdTimestamp) <= 0); } else if (StoragePolicyRuleTypeEntity.DAYS_SINCE_BDATA_PRIMARY_PARTITION_VALUE.equals(storagePolicyRuleType)) { if (businessObjectDataEntity.getUpdatedOn().compareTo(updatedOnThresholdTimestamp) <= 0) { Date primaryPartitionValue = businessObjectDataHelper.getDateFromString(businessObjectDataEntity.getPartitionValue()); if (primaryPartitionValue != null) { Date primaryPartitionValueThreshold = new Date(HerdDateUtils.addDays(currentTimestamp, -storagePolicyRuleValue).getTime()); createStoragePolicySelection = (primaryPartitionValue.compareTo(primaryPartitionValueThreshold) <= 0); } } } else { throw new IllegalStateException(String.format("Storage policy type \"%s\" is not supported.", storagePolicyRuleType)); } if (createStoragePolicySelection) { BusinessObjectDataKey businessObjectDataKey = businessObjectDataHelper.getBusinessObjectDataKey(businessObjectDataEntity); StoragePolicyKey storagePolicyKey = new StoragePolicyKey(storagePolicyEntity.getNamespace().getCode(), storagePolicyEntity.getName()); 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()); if (storagePolicySelections.size() >= maxResult) { break; } } } } if (storagePolicySelections.size() >= maxResult || map.isEmpty()) { break; } startPosition += maxResult; } if (storagePolicySelections.size() >= maxResult) { break; } } sendStoragePolicySelectionToSqsQueue(sqsQueueName, storagePolicySelections); return storagePolicySelections; }  @Override List<StoragePolicySelection> execute(String sqsQueueName, int maxResult); static final List<StoragePolicyPriorityLevel> STORAGE_POLICY_PRIORITY_LEVELS; static final List<String> SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES; }
@Test public void testCompleteStoragePolicyTransitionImpl() { Timestamp currentTimestamp = new Timestamp(LONG_VALUE); Integer maxResults = 10; Integer updatedOnThresholdInDays = 90; Integer storagePolicyTransitionMaxAllowedAttempts = 3; Map<BusinessObjectDataEntity, StoragePolicyEntity> noMatchingBusinessObjectDataEntities = new HashMap<>(); List<StoragePolicySelection> storagePolicySelections = new ArrayList<>(); when(herdDao.getCurrentTimestamp()).thenReturn(currentTimestamp); when(herdStringHelper.getConfigurationValueAsInteger(ConfigurationValue.STORAGE_POLICY_PROCESSOR_BDATA_UPDATED_ON_THRESHOLD_DAYS)) .thenReturn(updatedOnThresholdInDays); when(businessObjectDataDao.getBusinessObjectDataEntitiesMatchingStoragePolicies(StoragePolicySelectorServiceImpl.STORAGE_POLICY_PRIORITY_LEVELS.get(0), StoragePolicySelectorServiceImpl.SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES, storagePolicyTransitionMaxAllowedAttempts, 0, maxResults)) .thenReturn(noMatchingBusinessObjectDataEntities); when(businessObjectDataDao.getBusinessObjectDataEntitiesMatchingStoragePolicies(StoragePolicySelectorServiceImpl.STORAGE_POLICY_PRIORITY_LEVELS.get(1), StoragePolicySelectorServiceImpl.SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES, storagePolicyTransitionMaxAllowedAttempts, 0, maxResults)) .thenReturn(noMatchingBusinessObjectDataEntities); when(businessObjectDataDao.getBusinessObjectDataEntitiesMatchingStoragePolicies(StoragePolicySelectorServiceImpl.STORAGE_POLICY_PRIORITY_LEVELS.get(2), StoragePolicySelectorServiceImpl.SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES, storagePolicyTransitionMaxAllowedAttempts, 0, maxResults)) .thenReturn(noMatchingBusinessObjectDataEntities); when(businessObjectDataDao.getBusinessObjectDataEntitiesMatchingStoragePolicies(StoragePolicySelectorServiceImpl.STORAGE_POLICY_PRIORITY_LEVELS.get(3), StoragePolicySelectorServiceImpl.SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES, storagePolicyTransitionMaxAllowedAttempts, 0, maxResults)) .thenReturn(noMatchingBusinessObjectDataEntities); when(herdStringHelper.getConfigurationValueAsInteger(ConfigurationValue.STORAGE_POLICY_TRANSITION_MAX_ALLOWED_ATTEMPTS)) .thenReturn(storagePolicyTransitionMaxAllowedAttempts); List<StoragePolicySelection> result = storagePolicySelectorServiceImpl.execute(AWS_SQS_QUEUE_NAME, maxResults); verify(herdDao).getCurrentTimestamp(); verify(herdStringHelper).getConfigurationValueAsInteger(ConfigurationValue.STORAGE_POLICY_PROCESSOR_BDATA_UPDATED_ON_THRESHOLD_DAYS); verify(herdStringHelper).getConfigurationValueAsInteger(ConfigurationValue.STORAGE_POLICY_TRANSITION_MAX_ALLOWED_ATTEMPTS); verify(businessObjectDataDao) .getBusinessObjectDataEntitiesMatchingStoragePolicies(StoragePolicySelectorServiceImpl.STORAGE_POLICY_PRIORITY_LEVELS.get(0), StoragePolicySelectorServiceImpl.SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES, storagePolicyTransitionMaxAllowedAttempts, 0, maxResults); verify(businessObjectDataDao) .getBusinessObjectDataEntitiesMatchingStoragePolicies(StoragePolicySelectorServiceImpl.STORAGE_POLICY_PRIORITY_LEVELS.get(1), StoragePolicySelectorServiceImpl.SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES, storagePolicyTransitionMaxAllowedAttempts, 0, maxResults); verify(businessObjectDataDao) .getBusinessObjectDataEntitiesMatchingStoragePolicies(StoragePolicySelectorServiceImpl.STORAGE_POLICY_PRIORITY_LEVELS.get(2), StoragePolicySelectorServiceImpl.SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES, storagePolicyTransitionMaxAllowedAttempts, 0, maxResults); verify(businessObjectDataDao) .getBusinessObjectDataEntitiesMatchingStoragePolicies(StoragePolicySelectorServiceImpl.STORAGE_POLICY_PRIORITY_LEVELS.get(3), StoragePolicySelectorServiceImpl.SUPPORTED_BUSINESS_OBJECT_DATA_STATUSES, storagePolicyTransitionMaxAllowedAttempts, 0, maxResults); verifyNoMoreInteractionsHelper(); assertEquals(storagePolicySelections, result); }