VED-1059 Made adjustment to batch queueing logic (#1189)

Author: dlzhry2nhs

infrastructure/instance/kinesis_config.tf

@@ -1,11 +1,12 @@
-# Define the Kinesis Data Stream resource with 15 shards
 resource "aws_kinesis_stream" "processor_data_streams" {
   name = "${local.short_prefix}-processingdata-stream"
   stream_mode_details {
     stream_mode = "ON_DEMAND"
   }
-  encryption_type = "KMS"
-  kms_key_id      = data.aws_kms_key.existing_kinesis_encryption_key.arn
+
+  retention_period = var.environment == "prod" ? 24 * 7 : 24
+  encryption_type  = "KMS"
+  kms_key_id       = data.aws_kms_key.existing_kinesis_encryption_key.arn
 }
 
 locals {

lambdas/batch_processor_filter/src/batch_audit_repository.py

@@ -17,7 +17,10 @@ class BatchAuditRepository:
         | Key(AuditTableKeys.STATUS).eq(FileStatus.PREPROCESSED)
         | Key(AuditTableKeys.STATUS).eq(FileStatus.PROCESSING)
     )
-    _PROCESSING_AND_FAILED_STATUSES = {FileStatus.PROCESSING, FileStatus.FAILED}
+    # VED-1059: Record Forwarder Lambda consumes from the Kinesis stream at a slower rate than events come in. So there
+    # are negligible gains in starting the next file once preprocessing is complete. Hence, why we also include
+    # FileStatus.PREPROCESSED in this list and wait until a file is fully processed before picking up the next.
+    _PROCESSING_AND_FAILED_STATUSES = {FileStatus.PROCESSING, FileStatus.PREPROCESSED, FileStatus.FAILED}
 
     def __init__(self):
         self._batch_audit_table = get_dynamodb_table(AUDIT_TABLE_NAME)

lambdas/batch_processor_filter/tests/test_lambda_handler.py

@@ -203,6 +203,10 @@ def test_lambda_handler_raises_error_when_event_already_processing_for_supplier_
                 "Event is already being processed for supplier + vacc type queue",
                 FileStatus.PROCESSING,
             ),
+            (
+                "Event is already processing (preprocessed status) for supplier + vacc type queue",
+                FileStatus.PREPROCESSED,
+            ),
             (
                 "There is a failed event to be checked in supplier + vacc type queue",
                 FileStatus.FAILED,