RED-6864 - Optimization of Upload and Download for Large Files in Azure Blob...

persistence-service-v1/persistence-service-external-api-impl-v1/src/main/java/com/iqser/red/persistence/service/v1/external/api/impl/controller/DownloadController.java

@@ -9,6 +9,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 

persistence-service-v1/persistence-service-processor-v1/src/main/java/com/iqser/red/service/persistence/management/v1/processor/configuration/WebConfiguration.java

@@ -0,0 +1,27 @@
+package com.iqser.red.service.persistence.management.v1.processor.configuration;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.servlet.AsyncHandlerInterceptor;
+import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
+import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
+
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+
+@Configuration
+public class WebConfiguration implements WebMvcConfigurer {
+
+    private static final int TIMEOUT = 600000; // 10 minutes timeout;
+
+    @Override
+    public  void addInterceptors(InterceptorRegistry registry) {
+
+        registry.addInterceptor(new AsyncHandlerInterceptor() {
+            @Override
+            public void afterConcurrentHandlingStarted(HttpServletRequest request,
+                                                       HttpServletResponse response, Object handler) {
+                request.getAsyncContext().setTimeout(TIMEOUT);
+            }
+        });
+    }
+}

persistence-service-v1/persistence-service-processor-v1/src/main/java/com/iqser/red/service/persistence/management/v1/processor/model/RedactionFileResult.java

@@ -0,0 +1,33 @@
+package com.iqser.red.service.persistence.management.v1.processor.model;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.iqser.red.service.pdftron.redaction.v1.api.model.RedactionResultDetail;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
+@AllArgsConstructor
+@Builder
+@Getter
+@Setter
+public class RedactionFileResult {
+
+    private String fileId;
+
+    private boolean processed;
+
+    private int retryCounter;
+
+    @Builder.Default
+    private List<RedactionResultDetail> redactionResultDetailList = new ArrayList<>();
+
+
+    public void increaseRetryCounter() {
+        this.retryCounter = this.retryCounter + 1;
+    }
+
+}

persistence-service-v1/persistence-service-processor-v1/src/main/java/com/iqser/red/service/persistence/management/v1/processor/service/download/DownloadPreparationService.java

@@ -1,11 +1,12 @@
 package com.iqser.red.service.persistence.management.v1.processor.service.download;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -18,9 +19,12 @@ import com.iqser.red.service.pdftron.redaction.v1.api.model.RedactionResultDetai
 import com.iqser.red.service.pdftron.redaction.v1.api.model.RedactionResultMessage;
 import com.iqser.red.service.pdftron.redaction.v1.api.model.RedactionType;
 import com.iqser.red.service.persistence.management.v1.processor.configuration.MessagingConfiguration;
+import com.iqser.red.service.persistence.management.v1.processor.entity.dossier.DossierEntity;
+import com.iqser.red.service.persistence.management.v1.processor.entity.dossier.DossierTemplateEntity;
 import com.iqser.red.service.persistence.management.v1.processor.entity.dossier.FileEntity;
 import com.iqser.red.service.persistence.management.v1.processor.entity.dossier.ReportTemplateEntity;
 import com.iqser.red.service.persistence.management.v1.processor.entity.download.DownloadStatusEntity;
+import com.iqser.red.service.persistence.management.v1.processor.model.RedactionFileResult;
 import com.iqser.red.service.persistence.management.v1.processor.service.ColorsService;
 import com.iqser.red.service.persistence.management.v1.processor.service.FileManagementStorageService;
 import com.iqser.red.service.persistence.management.v1.processor.service.persistence.DossierTemplatePersistenceService;
@@ -50,7 +54,7 @@ import lombok.extern.slf4j.Slf4j;
 @RequiredArgsConstructor
 @FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
 public class DownloadPreparationService {
-
+    static int MAX_RETRY = 2;
     DownloadStatusPersistenceService downloadStatusPersistenceService;
     FileManagementStorageService fileManagementStorageService;
     ReportTemplatePersistenceService reportTemplatePersistenceService;
@@ -61,12 +65,28 @@ public class DownloadPreparationService {
     ColorsService colorsService;
     FileManagementServiceSettings settings;
     DossierTemplatePersistenceService dossierTemplatePersistenceService;
+    Map<String, List<RedactionFileResult>> redactionFileResultsMap;
 
 
     @Transactional
     public void createDownload(ReportResultMessage reportResultMessage) {
 
-        var downloadStatus = downloadStatusPersistenceService.getStatus(reportResultMessage.getDownloadId());
+        var downloadId = reportResultMessage.getDownloadId();
+        var downloadStatus = downloadStatusPersistenceService.getStatus(downloadId);
+        RedactionMessage.RedactionMessageBuilder messageBuilder = this.generateGeneralRedactionMessage(downloadId, downloadStatus);
+
+        redactionFileResultsMap.put(downloadId, new ArrayList<>());
+
+        downloadStatus.getFiles().forEach(fileEntity -> {
+            RedactionMessage message = messageBuilder.fileId(fileEntity.getId()).unapprovedFile(fileEntity.getWorkflowStatus() != WorkflowStatus.APPROVED).build();
+            redactionFileResultsMap.get(downloadId).add(RedactionFileResult.builder().fileId(fileEntity.getId()).processed(false).retryCounter(0).build());
+            log.info("Sending redaction request for downloadId:{} fileId:{} to pdftron-redaction-queue", downloadId, fileEntity.getId());
+            rabbitTemplate.convertAndSend(MessagingConfiguration.PDFTRON_QUEUE, message);
+        });
+    }
+
+    private RedactionMessage.RedactionMessageBuilder generateGeneralRedactionMessage(String downloadId, DownloadStatusEntity downloadStatus) {
+
         var dossier = downloadStatus.getDossier();
         var storedPreviewColor = downloadStatus.getRedactionPreviewColor();
         var dossierTemplate = dossierTemplatePersistenceService.getDossierTemplate(dossier.getDossierTemplateId());
@@ -79,26 +99,25 @@ public class DownloadPreparationService {
             previewColor = storedPreviewColor;
         }
         String appliedRedactionColor = colorsService.getColors(dossier.getDossierTemplateId()).getAppliedRedactionColor();
+        return buildRedactionMessage(downloadId, downloadStatus, dossier, dossierTemplate, previewColor, appliedRedactionColor);
+    }
 
-        RedactionMessage message = RedactionMessage.builder()
+    private RedactionMessage.RedactionMessageBuilder buildRedactionMessage(String downloadId,
+                                                                           DownloadStatusEntity downloadStatus,
+                                                                           DossierEntity dossier,
+                                                                           DossierTemplateEntity dossierTemplate,
+                                                                           String previewColor,
+                                                                           String appliedRedactionColor) {
+
+        return RedactionMessage.builder()
                 .dossierId(dossier.getId())
-                .downloadId(reportResultMessage.getDownloadId())
+                .downloadId(downloadId)
                 .redactionTypes(toPdfTronRedactionTypes(downloadStatus.getDownloadFileTypes()))
-                .fileIds(downloadStatus.getFiles().stream().map(FileEntity::getId).collect(Collectors.toList()))
-                .unapprovedFileIds(downloadStatus.getFiles()
-                        .stream()
-                        .filter(f -> !WorkflowStatus.APPROVED.equals(f.getWorkflowStatus()))
-                        .map(FileEntity::getId)
-                        .collect(Collectors.toSet()))
                 .redactionPreviewColor(previewColor)
                 .keepImageMetaData(dossierTemplate.isKeepImageMetadata())
                 .keepHiddenText(dossierTemplate.isKeepHiddenText())
                 .keepOverlappingObjects(dossierTemplate.isKeepOverlappingObjects())
-                .appliedRedactionColor(appliedRedactionColor)
+                .appliedRedactionColor(appliedRedactionColor);
-                .build();
-
-        log.info("Sending redaction request for downloadId:{} to pdftron-redaction-queue", message.getDownloadId());
-        rabbitTemplate.convertAndSend(MessagingConfiguration.PDFTRON_QUEUE, message);
     }
 
 
@@ -119,17 +138,55 @@ public class DownloadPreparationService {
         return result;
     }
 
+    public void processingRedactionResultMessage(RedactionResultMessage redactionResultMessage) {
 
-    public void createDownload(RedactionResultMessage reportResultMessage) {
+        String downloadId = redactionResultMessage.getDownloadId();
+        long totalFiles = downloadStatusPersistenceService.getStatus(downloadId).getFiles().size();
+        List<RedactionFileResult> redactionFileResults = redactionFileResultsMap.get(downloadId);
+        Optional<RedactionFileResult> resultOptional = redactionFileResults.stream().filter(r -> r.getFileId().equals(redactionResultMessage.getFileId())).findFirst();
+        resultOptional.ifPresent(redactionFileResult -> {
+            redactionFileResult.setRedactionResultDetailList(redactionResultMessage.getRedactionResultDetails());
+            redactionFileResult.setProcessed(true);
+        });
 
-        DownloadStatusEntity downloadStatus = downloadStatusPersistenceService.getStatus(reportResultMessage.getDownloadId());
+        var processedFilesCount = redactionFileResults.stream().filter(RedactionFileResult::isProcessed).count();
+        log.info("Processed {} files out of total {} files for download {}", processedFilesCount, totalFiles, downloadId);
 
-        var storedFileInformations = getStoredFileInformation(reportResultMessage.getDownloadId());
+        if (processedFilesCount == totalFiles) {
+            createDownload(redactionFileResults, downloadId);
+        }
+    }
+
+    public void checkForRetryProcess(String downloadId, String fileId, boolean unapprovedFile) {
+
+        List<RedactionFileResult> redactionFileResults = redactionFileResultsMap.get(downloadId);
+        Optional<RedactionFileResult> resultOptional = redactionFileResults.stream().filter(r -> r.getFileId().equals(fileId)).findFirst();
+        if (resultOptional.isPresent()) {
+            RedactionFileResult result = resultOptional.get();
+            if (result.getRetryCounter() >= MAX_RETRY) { // update download status
+                log.info("Failed download after max retries for downloadId: {}", downloadId);
+                downloadStatusPersistenceService.updateStatus(downloadId, DownloadStatusValue.FAILED);
+            } else { // retry to send it again
+                result.increaseRetryCounter();
+                var downloadStatus = downloadStatusPersistenceService.getStatus(downloadId);
+                RedactionMessage.RedactionMessageBuilder messageBuilder = this.generateGeneralRedactionMessage(downloadId, downloadStatus);
+                RedactionMessage message = messageBuilder.fileId(fileId).unapprovedFile(unapprovedFile).build();
+                log.info("Resending redaction request for downloadId:{} fileId: {} to pdftron-redaction-queue", downloadId, fileId);
+                rabbitTemplate.convertAndSend(MessagingConfiguration.PDFTRON_QUEUE, message);
+            }
+        }
+    }
+
+    public void createDownload(List<RedactionFileResult> redactionFileResults, String downloadId) {
+
+        DownloadStatusEntity downloadStatus = downloadStatusPersistenceService.getStatus(downloadId);
+
+        var storedFileInformations = getStoredFileInformation(downloadId);
 
         try (FileSystemBackedArchiver fileSystemBackedArchiver = new FileSystemBackedArchiver()) {
 
-            generateAndAddFiles(downloadStatus, reportResultMessage, fileSystemBackedArchiver);
+            generateAndAddFiles(downloadStatus, redactionFileResults, fileSystemBackedArchiver);
-            addReports(reportResultMessage.getDownloadId(), downloadStatus, storedFileInformations, fileSystemBackedArchiver);
+            addReports(downloadId, downloadStatus, storedFileInformations, fileSystemBackedArchiver);
             storeZipFile(downloadStatus, fileSystemBackedArchiver);
 
             updateStatusToReady(downloadStatus, fileSystemBackedArchiver);
@@ -145,10 +202,11 @@ public class DownloadPreparationService {
         }
 
         downloadReportCleanupService.deleteTmpReportFiles(storedFileInformations.stream().map(StoredFileInformation::getStorageId).collect(Collectors.toSet()));
-        downloadReportCleanupService.deleteTmpReportFiles(reportResultMessage.getRedactionResultDetails()
+        redactionFileResults.forEach(redactionFileResult -> downloadReportCleanupService.deleteTmpReportFiles(redactionFileResult.getRedactionResultDetailList()
                 .stream()
                 .map(RedactionResultDetail::getStorageId)
-                .collect(Collectors.toSet()));
+                .collect(Collectors.toSet())));
+
     }
 
 
@@ -158,7 +216,7 @@ public class DownloadPreparationService {
     }
 
 
-    private void generateAndAddFiles(DownloadStatusEntity downloadStatus, RedactionResultMessage reportResultMessage, FileSystemBackedArchiver fileSystemBackedArchiver) {
+    private void generateAndAddFiles(DownloadStatusEntity downloadStatus, List<RedactionFileResult> redactionFileResults, FileSystemBackedArchiver fileSystemBackedArchiver) {
 
         int i = 1;
         long fileGenerationStart = System.currentTimeMillis();
@@ -170,22 +228,28 @@ public class DownloadPreparationService {
             var isFileApproved = WorkflowStatus.APPROVED.equals(file.getWorkflowStatus());
             String filename = isFileApproved ? file.getFilename() : "UNAPPROVED_" + file.getFilename();
             for (DownloadFileType downloadFileType : downloadStatus.getDownloadFileTypes()) {
+                Optional<RedactionFileResult> redactionFileResult = redactionFileResults.stream().filter(rfr -> rfr.getFileId().equals(file.getId())).findFirst();
+
+                if (redactionFileResult.isEmpty()) {
+                    throw new RuntimeException();
+                }
 
                 if (downloadFileType.name().equals(DownloadFileType.ORIGINAL.name())) {
                     fileSystemBackedArchiver.addEntry(new FileSystemBackedArchiver.ArchiveModel("Original", filename, original));
                 }
                 if (downloadFileType.name().equals(DownloadFileType.PREVIEW.name())) {
                     fileSystemBackedArchiver.addEntry(new FileSystemBackedArchiver.ArchiveModel("Preview", addSuffix(filename, "highlighted"), //
-                            getPreview(file.getId(), reportResultMessage.getRedactionResultDetails())));
+                            getPreview(file.getId(), redactionFileResult.get().getRedactionResultDetailList())));
                 }
                 if (downloadFileType.name().equals(DownloadFileType.DELTA_PREVIEW.name())) {
                     fileSystemBackedArchiver.addEntry(new FileSystemBackedArchiver.ArchiveModel("Delta Preview", addSuffix(filename, "delta_highlighted"), //
-                            getDeltaPreview(file.getId(), reportResultMessage.getRedactionResultDetails())));
+                            getDeltaPreview(file.getId(), redactionFileResult.get().getRedactionResultDetailList())));
                 }
                 if (downloadFileType.name().equals(DownloadFileType.REDACTED.name()) && isFileApproved) {
                     fileSystemBackedArchiver.addEntry(new FileSystemBackedArchiver.ArchiveModel("Redacted", addSuffix(file.getFilename(), "redacted"), //
-                            getRedacted(file.getId(), reportResultMessage.getRedactionResultDetails())));
+                            getRedacted(file.getId(), redactionFileResult.get().getRedactionResultDetailList())));
                 }
+
             }
             log.info("Successfully added file {}/{} for downloadId {}, took {}", i, downloadStatus.getFiles().size(), downloadStatus.getStorageId(), System.currentTimeMillis() - start);
             i++;
@@ -202,7 +266,7 @@ public class DownloadPreparationService {
 
     private byte[] getStoredFileBytes(String fileId, List<RedactionResultDetail> redactionResultDetails, RedactionType redactionType) {
 
-        var redactionResultDetail = redactionResultDetails.stream().filter(rrd -> Objects.equals(fileId, rrd.getFileId()) && rrd.getRedactionType() == redactionType).findFirst();
+        var redactionResultDetail = redactionResultDetails.stream().filter(rrd -> rrd.getRedactionType() == redactionType).findFirst();
         if (redactionResultDetail.isPresent()) {
             try (var inputStream = fileManagementStorageService.getObject(TenantContext.getTenantId(), redactionResultDetail.get().getStorageId())) {
                 byte[] storedFileBytes = inputStream.readAllBytes();

persistence-service-v1/persistence-service-processor-v1/src/main/java/com/iqser/red/service/persistence/management/v1/processor/service/download/RedactionDlqMessageReceiver.java

@@ -30,6 +30,7 @@ public class RedactionDlqMessageReceiver {
 
     ObjectMapper objectMapper;
     DownloadStatusPersistenceService downloadStatusPersistenceService;
+    DownloadPreparationService downloadPreparationService;
 
 
     @RabbitHandler
@@ -39,17 +40,25 @@ public class RedactionDlqMessageReceiver {
         // We just assume that the message contains a downloadId.
         JsonNode jsonNode = objectMapper.readTree(message.getBody());
         final String downloadId;
+        final String fileId;
         try {
             downloadId = jsonNode.findValue("downloadId").asText();
+
         } catch (Exception e) {
             log.warn("Received a message in the " + MessagingConfiguration.PDFTRON_DLQ + " that contains no downloadId" + LINE_SEPARATOR + "{}", jsonNode.asText());
             throw new RuntimeException(e);
         }
 
-        log.info("Received a dead message with downloadId:{}, updating the download as failed", downloadId);
+        try {
-
+            fileId = jsonNode.findValue("fileId").asText();
-        downloadStatusPersistenceService.updateStatus(downloadId, DownloadStatusValue.FAILED);
+            var unapproved = jsonNode.findValue("unapprovedFile");
+            log.info("Received a dead message with downloadId: {}, fileId: {} retry", downloadId, fileId);
+            downloadPreparationService.checkForRetryProcess(downloadId, fileId, unapproved.asBoolean());
 
+        } catch (Exception e) {
+            log.info("Received a dead message with downloadId: {}, updating the download as failed", downloadId);
+            downloadStatusPersistenceService.updateStatus(downloadId, DownloadStatusValue.FAILED);
+        }
     }
 
 }

persistence-service-v1/persistence-service-processor-v1/src/main/java/com/iqser/red/service/persistence/management/v1/processor/service/download/RedactionResultMessageReceiver.java

@@ -46,9 +46,9 @@ public class RedactionResultMessageReceiver {
 
     public void receive(RedactionResultMessage redactionResultMessage) {
 
-        log.info("Received redaction results for downloadId:{}", redactionResultMessage.getDownloadId());
+        log.info("Received redaction results for downloadId: {} fileId: {}", redactionResultMessage.getDownloadId(), redactionResultMessage.getFileId());
 
-        downloadPreparationService.createDownload(redactionResultMessage);
+        downloadPreparationService.processingRedactionResultMessage(redactionResultMessage);
     }
 
 }

persistence-service-v1/persistence-service-server-v1/src/test/java/com/iqser/red/service/peristence/v1/server/integration/tests/DownloadPreparationTest.java

@@ -15,6 +15,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.mock.web.MockMultipartFile;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iqser.red.service.pdftron.redaction.v1.api.model.RedactionResultDetail;
 import com.iqser.red.service.pdftron.redaction.v1.api.model.RedactionResultMessage;
 import com.iqser.red.service.peristence.v1.server.integration.client.DossierClient;
 import com.iqser.red.service.peristence.v1.server.integration.client.DownloadClient;
@@ -132,6 +133,7 @@ public class DownloadPreparationTest extends AbstractPersistenceServerServiceTes
         redactionResultMessageReceiver.receive(RedactionResultMessage.builder()
                 .downloadId(downloadId)
                 .dossierId(testData.getDossierId())
+                .fileId(testData.getFileId())
                 .redactionResultDetails(Collections.emptyList())
                 .build());
 

persistence-service-v1/persistence-service-server-v1/src/test/java/com/iqser/red/service/peristence/v1/server/integration/tests/DownloadTest.java

@@ -25,6 +25,7 @@ import com.iqser.red.service.persistence.service.v1.api.shared.model.PrepareDown
 import com.iqser.red.service.persistence.service.v1.api.shared.model.RemoveDownloadRequest;
 import com.iqser.red.service.persistence.service.v1.api.shared.model.dossiertemplate.DownloadFileType;
 import com.iqser.red.service.persistence.service.v1.api.shared.model.dossiertemplate.dossier.file.WorkflowStatus;
+import com.iqser.red.service.redaction.report.v1.api.model.ReportResultMessage;
 import com.knecon.fforesight.tenantcommons.TenantContext;
 
 import feign.FeignException;
@@ -88,7 +89,10 @@ public class DownloadTest extends AbstractPersistenceServerServiceTest {
         var reportInfoId = downloads.getStorageId().substring(0, downloads.getStorageId().length() - 3) + "/REPORT_INFO.json";
         storageService.storeJSONObject(TenantContext.getTenantId(), reportInfoId, new ArrayList<>());
 
-        downloadPreparationService.createDownload(RedactionResultMessage.builder().downloadId(downloads.getStorageId()).build());
+        downloadPreparationService.createDownload(ReportResultMessage.builder().downloadId(downloads.getStorageId()).build());
+
+        downloadPreparationService.processingRedactionResultMessage(RedactionResultMessage.builder().downloadId(downloads.getStorageId()).dossierId(dossier.getId()).fileId(file.getId()).build());
+        downloadPreparationService.processingRedactionResultMessage(RedactionResultMessage.builder().downloadId(downloads.getStorageId()).dossierId(dossier.getId()).fileId(file2.getId()).build());
 
         var statuses = downloadClient.getDownloadStatus();
         assertThat(statuses.getDownloadStatus()).isNotEmpty();

persistence-service-v1/pom.xml

@@ -31,7 +31,7 @@
     <properties>
         <redaction-service.version>4.30.0</redaction-service.version>
         <search-service.version>2.71.0</search-service.version>
-        <pdftron-redaction-service.version>4.18.0</pdftron-redaction-service.version>
+        <pdftron-redaction-service.version>4.29.0</pdftron-redaction-service.version>
         <redaction-report-service.version>4.13.0</redaction-report-service.version>
         <ocr-service.version>3.10.0</ocr-service.version>
     </properties>