test_class.declarator: storageFileHelper
test_class.modifier: @Autowired
      private
test_class.original_string: @Autowired
      private StorageFileHelper storageFileHelper;
test_class.type: StorageFileHelper
test_class.var_name: storageFileHelper
test_class.declarator: storagePolicyProcessorJmsMessageListener
test_class.modifier: @Autowired
test_class.original_string: @Autowired
      StoragePolicyProcessorJmsMessageListener storagePolicyProcessorJmsMessageListener;
test_class.type: StoragePolicyProcessorJmsMessageListener
test_class.var_name: storagePolicyProcessorJmsMessageListener
test_class.file: herd-code/herd-service/src/test/java/org/finra/herd/service/helper/StoragePolicyProcessorJmsMessageListenerTest.java
test_class.identifier: StoragePolicyProcessorJmsMessageListenerTest
test_class.superclass: extends AbstractServiceTest
test_case.body: @Test
      public void testProcessMessage() throws Exception
      {
          // Build the expected S3 key prefix for test business object data.
          String s3KeyPrefix =
              getExpectedS3KeyPrefix(BDEF_NAMESPACE, DATA_PROVIDER_NAME, BDEF_NAME, FORMAT_USAGE_CODE, FORMAT_FILE_TYPE_CODE, FORMAT_VERSION, PARTITION_KEY,
                  PARTITION_VALUE, null, null, DATA_VERSION);
  
          // Create S3FileTransferRequestParamsDto to access the source S3 bucket location.
          // Since test S3 key prefix represents a directory, we add a trailing '/' character to it.
          S3FileTransferRequestParamsDto sourceS3FileTransferRequestParamsDto =
              S3FileTransferRequestParamsDto.builder().withS3BucketName(S3_BUCKET_NAME).withS3KeyPrefix(s3KeyPrefix + "/").build();
  
          // Create and persist the relative database entities.
          storagePolicyServiceTestHelper
              .createDatabaseEntitiesForStoragePolicyTesting(STORAGE_POLICY_NAMESPACE_CD, Arrays.asList(STORAGE_POLICY_RULE_TYPE), BDEF_NAMESPACE, BDEF_NAME,
                  Arrays.asList(FORMAT_FILE_TYPE_CODE), Arrays.asList(STORAGE_NAME), Arrays.asList(StoragePolicyTransitionTypeEntity.GLACIER));
  
          // Create a business object data key.
          BusinessObjectDataKey businessObjectDataKey =
              new BusinessObjectDataKey(BDEF_NAMESPACE, BDEF_NAME, FORMAT_USAGE_CODE, FORMAT_FILE_TYPE_CODE, FORMAT_VERSION, PARTITION_VALUE,
                  NO_SUBPARTITION_VALUES, DATA_VERSION);
  
          // Create and persist an ENABLED storage unit in the source storage.
          StorageUnitEntity sourceStorageUnitEntity = storageUnitDaoTestHelper
              .createStorageUnitEntity(STORAGE_NAME, businessObjectDataKey, LATEST_VERSION_FLAG_SET, BusinessObjectDataStatusEntity.VALID,
                  StorageUnitStatusEntity.ENABLED, NO_STORAGE_DIRECTORY_PATH);
  
          // Add storage files to the source storage unit.
          for (String filePath : LOCAL_FILES)
          {
              storageFileDaoTestHelper.createStorageFileEntity(sourceStorageUnitEntity, s3KeyPrefix + "/" + filePath, FILE_SIZE_1_KB, ROW_COUNT_1000);
          }
  
          // Get the source storage files.
          List<StorageFile> sourceStorageFiles = storageFileHelper.createStorageFilesFromEntities(sourceStorageUnitEntity.getStorageFiles());
  
          // Create a storage policy key.
          StoragePolicyKey storagePolicyKey = new StoragePolicyKey(STORAGE_POLICY_NAMESPACE_CD, STORAGE_POLICY_NAME);
  
          // Create and persist a storage policy entity.
          storagePolicyDaoTestHelper
              .createStoragePolicyEntity(storagePolicyKey, STORAGE_POLICY_RULE_TYPE, STORAGE_POLICY_RULE_VALUE, BDEF_NAMESPACE, BDEF_NAME, FORMAT_USAGE_CODE,
                  FORMAT_FILE_TYPE_CODE, STORAGE_NAME, StoragePolicyTransitionTypeEntity.GLACIER, StoragePolicyStatusEntity.ENABLED, INITIAL_VERSION,
                  LATEST_VERSION_FLAG_SET);
  
          // Override configuration to specify some settings required for testing.
          Map<String, Object> overrideMap = new HashMap<>();
          overrideMap.put(ConfigurationValue.S3_ARCHIVE_TO_GLACIER_ROLE_ARN.getKey(), S3_OBJECT_TAGGER_ROLE_ARN);
          overrideMap.put(ConfigurationValue.S3_ARCHIVE_TO_GLACIER_ROLE_SESSION_NAME.getKey(), S3_OBJECT_TAGGER_ROLE_SESSION_NAME);
          modifyPropertySourceInEnvironment(overrideMap);
  
          try
          {
              // Put relative S3 files into the source S3 bucket.
              for (StorageFile storageFile : sourceStorageFiles)
              {
                  s3Operations.putObject(new PutObjectRequest(S3_BUCKET_NAME, storageFile.getFilePath(),
                      new ByteArrayInputStream(new byte[storageFile.getFileSizeBytes().intValue()]), null), null);
              }
  
              // Perform a storage policy transition.
              storagePolicyProcessorJmsMessageListener
                  .processMessage(jsonHelper.objectToJson(new StoragePolicySelection(businessObjectDataKey, storagePolicyKey, INITIAL_VERSION)), null);
  
              // Validate the status of the source storage unit.
              assertEquals(StorageUnitStatusEntity.ARCHIVED, sourceStorageUnitEntity.getStatus().getCode());
          }
          finally
          {
              // Delete test files from S3 storage.
              if (!s3Dao.listDirectory(sourceS3FileTransferRequestParamsDto).isEmpty())
              {
                  s3Dao.deleteDirectory(sourceS3FileTransferRequestParamsDto);
              }
              s3Operations.rollback();
  
              // Restore the property sources so we don't affect other tests.
              restorePropertySourceInEnvironment();
          }
      }
test_case.class_method_signature: StoragePolicyProcessorJmsMessageListenerTest.testProcessMessage()
test_case.full_signature: @Test public void testProcessMessage()
test_case.identifier: testProcessMessage
test_case.invocations: getExpectedS3KeyPrefix build withS3KeyPrefix withS3BucketName builder createDatabaseEntitiesForStoragePolicyTesting asList asList asList asList createStorageUnitEntity createStorageFileEntity createStorageFilesFromEntities getStorageFiles createStoragePolicyEntity put getKey put getKey modifyPropertySourceInEnvironment putObject getFilePath intValue getFileSizeBytes processMessage objectToJson assertEquals getCode getStatus isEmpty listDirectory deleteDirectory rollback restorePropertySourceInEnvironment
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testProcessMessage()
focal_class.declarator: LOGGER = LoggerFactory.getLogger(StoragePolicyProcessorJmsMessageListener.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOGGER = LoggerFactory.getLogger(StoragePolicyProcessorJmsMessageListener.class);
focal_class.type: Logger
focal_class.var_name: LOGGER
focal_class.declarator: configurationHelper
focal_class.modifier: @Autowired
      private
focal_class.original_string: @Autowired
      private ConfigurationHelper configurationHelper;
focal_class.type: ConfigurationHelper
focal_class.var_name: configurationHelper
focal_class.declarator: jsonHelper
focal_class.modifier: @Autowired
      private
focal_class.original_string: @Autowired
      private JsonHelper jsonHelper;
focal_class.type: JsonHelper
focal_class.var_name: jsonHelper
focal_class.declarator: storagePolicyProcessorService
focal_class.modifier: @Autowired
      private
focal_class.original_string: @Autowired
      private StoragePolicyProcessorService storagePolicyProcessorService;
focal_class.type: StoragePolicyProcessorService
focal_class.var_name: storagePolicyProcessorService
focal_class.file: herd-code/herd-service/src/main/java/org/finra/herd/service/helper/StoragePolicyProcessorJmsMessageListener.java
focal_class.identifier: StoragePolicyProcessorJmsMessageListener
focal_class.class_method_signature: StoragePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener()
focal_class.full_signature: @Scheduled(fixedDelay = 60000) public void controlStoragePolicyProcessorJmsMessageListener()
focal_class.identifier: controlStoragePolicyProcessorJmsMessageListener
focal_class.modifiers: @Scheduled(fixedDelay = 60000) public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void controlStoragePolicyProcessorJmsMessageListener()
focal_class.class_method_signature: StoragePolicyProcessorJmsMessageListener.processMessage(String payload, @Headers Map<Object, Object> allHeaders)
focal_class.full_signature: @JmsListener(id = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE, containerFactory = "storagePolicyProcessorJmsListenerContainerFactory", destination = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE) public void processMessage(String payload, @Headers Map<Object, Object> allHeaders)
focal_class.identifier: processMessage
focal_class.modifiers: @JmsListener(id = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE, containerFactory = "storagePolicyProcessorJmsListenerContainerFactory", destination = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE) public
focal_class.parameters: (String payload, @Headers Map<Object, Object> allHeaders)
focal_class.return: void
focal_class.signature: void processMessage(String payload, @Headers Map<Object, Object> allHeaders)
focal_method.body: @JmsListener(id = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE,
          containerFactory = "storagePolicyProcessorJmsListenerContainerFactory",
          destination = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE)
      public void processMessage(String payload, @Headers Map<Object, Object> allHeaders)
      {
          LOGGER.info("Message received from the JMS queue. jmsQueueName=\"{}\" jmsMessageHeaders=\"{}\" jmsMessagePayload={}",
              HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE, allHeaders, payload);
  
          // Process the message as storage policy selection message.
          try
          {
              // Process messages coming from the storage policy selector job.
              StoragePolicySelection storagePolicySelection = jsonHelper.unmarshallJsonToObject(StoragePolicySelection.class, payload);
  
              LOGGER.debug("Received storage policy selection message: businessObjectDataKey={} storagePolicyKey={} storagePolicyVersion={}",
                  jsonHelper.objectToJson(storagePolicySelection.getBusinessObjectDataKey()),
                  jsonHelper.objectToJson(storagePolicySelection.getStoragePolicyKey()), storagePolicySelection.getStoragePolicyVersion());
  
              // Process the storage policy selection message.
              storagePolicyProcessorService.processStoragePolicySelectionMessage(storagePolicySelection);
          }
          catch (RuntimeException | IOException e)
          {
              // Log a warning message if storage unit status is already ARCHIVED. Such error case is typically caused by a duplicate SQS message.
              if (e instanceof IllegalArgumentException &&
                  e.getMessage().startsWith(String.format("Storage unit status is \"%s\"", StorageUnitStatusEntity.ARCHIVED)))
              {
                  LOGGER.warn("Failed to process message from the JMS queue. jmsQueueName=\"{}\" jmsMessagePayload={}",
                      HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE, payload, e);
              }
              // Otherwise, log an error.
              else
              {
                  LOGGER.error("Failed to process message from the JMS queue. jmsQueueName=\"{}\" jmsMessagePayload={}",
                      HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE, payload, e);
              }
          }
      }
focal_method.class_method_signature: StoragePolicyProcessorJmsMessageListener.processMessage(String payload, @Headers Map<Object, Object> allHeaders)
focal_method.full_signature: @JmsListener(id = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE, containerFactory = "storagePolicyProcessorJmsListenerContainerFactory", destination = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE) public void processMessage(String payload, @Headers Map<Object, Object> allHeaders)
focal_method.identifier: processMessage
focal_method.invocations: info unmarshallJsonToObject debug objectToJson getBusinessObjectDataKey objectToJson getStoragePolicyKey getStoragePolicyVersion processStoragePolicySelectionMessage startsWith getMessage format warn error
focal_method.modifiers: @JmsListener(id = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE, containerFactory = "storagePolicyProcessorJmsListenerContainerFactory", destination = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE) public
focal_method.parameters: (String payload, @Headers Map<Object, Object> allHeaders)
focal_method.return: void
focal_method.signature: void processMessage(String payload, @Headers Map<Object, Object> allHeaders)