Reprocessor Changes

Signed-off-by: trialblazerseee <84778104+trialblazerseee@users.noreply.github.com>

Author: trialblazerseee

registration-processor/registration-processor-registration-status-service-impl/src/main/java/io/mosip/registration/processor/status/repositary/RegistrationRepositary.java

@@ -65,7 +65,7 @@ public List<String> getProcessedOrProcessingRegIds(@Param("regIds") List<String>
 	@Query("SELECT registration FROM RegistrationStatusEntity registration WHERE registration.regId = :regId AND registration.registrationType = :registrationType AND registration.iteration = :iteration")
 	public List<RegistrationStatusEntity> getByIdAndProcessAndIteration(@Param("regId") String regId, @Param("registrationType") String process, @Param("iteration") int iteration);
 
-	@Query(value ="SELECT * FROM registration r WHERE r.process IN :processes AND r.latest_trn_status_code IN :status AND r.reg_process_retry_count<=:reprocessCount AND r.latest_trn_dtimes <:timeDifference AND r.status_code NOT IN :statusCodes AND r.reg_stage_name NOT IN :excludeStageNames order by r.latest_trn_dtimes LIMIT :fetchSize ", nativeQuery = true)
+	@Query(value ="SELECT * FROM registration r WHERE r.process IN :processList AND r.latest_trn_status_code IN :status AND r.reg_process_retry_count<=:reprocessCount AND r.latest_trn_dtimes <:timeDifference AND r.status_code NOT IN :statusCodes AND r.reg_stage_name NOT IN :excludeStageNames order by r.latest_trn_dtimes LIMIT :fetchSize ", nativeQuery = true)
 	public List<RegistrationStatusEntity> getUnProcessedPackets(@Param("processList") List<String> processes, @Param("status") List<String> status,@Param("reprocessCount") Integer reprocessCount,@Param("timeDifference") LocalDateTime timeDifference,@Param("statusCodes") List<String> statusCodes,@Param("fetchSize") Integer fetchSize,@Param("excludeStageNames") List<String> excludeStageNames);
 
 }

registration-processor/workflow-engine/registration-processor-reprocessor/src/main/java/io/mosip/registration/processor/reprocessor/ReprocessorApplication.java

@@ -21,7 +21,8 @@ public static void main(String[] args) {
 				"io.mosip.registration.processor.reprocessor.config",
 				"io.mosip.registration.processor.status.config",
 				"io.mosip.registration.processor.core.kernel.beans",
-				"io.mosip.registration.processor.packet.storage.config");
+				"io.mosip.registration.processor.packet.storage.config",
+				"io.mosip.registration.processor.reprocessor.service.impl");
 		ctx.refresh();
 		ReprocessorVerticle reprocessorVerticle = ctx.getBean(ReprocessorVerticle.class);
 		reprocessorVerticle.deployVerticle();

registration-processor/workflow-engine/registration-processor-reprocessor/src/main/java/io/mosip/registration/processor/reprocessor/service/ReprocessorVerticalService.java

@@ -1,12 +1,16 @@
 package io.mosip.registration.processor.reprocessor.service;
 
-
 import io.mosip.registration.processor.status.dto.InternalRegistrationStatusDto;
-
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 public interface ReprocessorVerticalService {
 
-    List<InternalRegistrationStatusDto> fetchUnProcessedPackets(List<String> processList, Integer fetchSize, long elapseTime, Integer reprocessCount,
-                                                                List<String> status, List<String> excludeStageNames);
+    CompletableFuture<List<InternalRegistrationStatusDto>> fetchUnProcessedPackets(
+            List<String> processList,
+            Integer fetchSize,
+            long elapseTime,
+            Integer reprocessCount,
+            List<String> status,
+            List<String> excludeStageNames);
 }

registration-processor/workflow-engine/registration-processor-reprocessor/src/main/java/io/mosip/registration/processor/reprocessor/service/impl/ReprocessorVerticalServiceImpl.java

@@ -8,21 +8,21 @@
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
-import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 @Service
 public class ReprocessorVerticalServiceImpl implements ReprocessorVerticalService {
 
     /** The registration status service. */
     @Autowired
-    RegistrationStatusService<String, InternalRegistrationStatusDto, RegistrationStatusDto> registrationStatusService;
-
+    private RegistrationStatusService<String, InternalRegistrationStatusDto, RegistrationStatusDto> registrationStatusService;
 
     @Override
     @Async
-    public List<InternalRegistrationStatusDto> fetchUnProcessedPackets(List<String> processList, Integer fetchSize, long elapseTime, Integer reprocessCount, List<String> status, List<String> excludeStageNames) {
-        return registrationStatusService.getUnProcessedPackets(processList, fetchSize, elapseTime,
+    public CompletableFuture<List<InternalRegistrationStatusDto>> fetchUnProcessedPackets(List<String> processList, Integer fetchSize, long elapseTime, Integer reprocessCount, List<String> status, List<String> excludeStageNames) {
+        List<InternalRegistrationStatusDto> result =  registrationStatusService.getUnProcessedPackets(processList, fetchSize, elapseTime,
                 reprocessCount, status, excludeStageNames);
+        return CompletableFuture.completedFuture(result);
     }
 }

registration-processor/workflow-engine/registration-processor-reprocessor/src/main/java/io/mosip/registration/processor/reprocessor/verticle/ReprocessorVerticle.java

@@ -155,7 +155,6 @@ public class ReprocessorVerticle extends MosipVerticleAPIManager {
 	public void deployVerticle() {
 		mosipEventBus = this.getEventBus(this, clusterManagerUrl);
 		deployScheduler(getVertx());
-
 	}
 
 	/**
@@ -371,7 +370,11 @@ public MessageDTO process(MessageDTO object) {
 											auditLogRequestBuilder.createAuditRequestBuilder(description.getMessage(), eventId, eventName,
 													eventType, moduleId, moduleName, registrationId);
 									}
-								},sendExecutor)).collect(Collectors.toList());
+								},sendExecutor).exceptionally(ex -> {
+									regProcLogger.error(LoggerFileConstant.SESSIONID.toString(),
+											description.getCode() + " -- ",
+											PlatformErrorMessages.RPR_RGS_REGISTRATION_TABLE_NOT_ACCESSIBLE.getMessage(), ex.toString());									return null;
+								})).collect(Collectors.toList());
 
 				CompletableFuture.allOf(sendTasks.toArray(new CompletableFuture[0])).join();
 			}
@@ -384,7 +387,7 @@ public MessageDTO process(MessageDTO object) {
 					description.getCode() + " -- ",
 					PlatformErrorMessages.RPR_RGS_REGISTRATION_TABLE_NOT_ACCESSIBLE.getMessage(), e.toString());
 
-		}catch (Exception ex) {
+		} catch (Exception ex) {
 			isTransactionSuccessful.set(false);
 			description.setMessage(PlatformErrorMessages.REPROCESSOR_VERTICLE_FAILED.getMessage());
 			description.setCode(PlatformErrorMessages.REPROCESSOR_VERTICLE_FAILED.getCode());
@@ -519,7 +522,7 @@ private void fetchPacketsIfBelowThreshold(LinkedHashMap<String, Integer> require
 						Deque<?> cachedPackets = packetCacheMap.get(entry.getKey());
 						return cachedPackets == null ||  cachedPackets.size() < threasholdForFetch;
 				})
-				.map(entry -> CompletableFuture.runAsync(() -> {
+				.map(entry -> {
 					String key = entry.getKey();
 
 					// Parse key into Process & Status
@@ -528,16 +531,23 @@ private void fetchPacketsIfBelowThreshold(LinkedHashMap<String, Integer> require
 					List<String> statusValList = entryPair.getValue();
 
 					// Fetch unprocessed packets
-					List<InternalRegistrationStatusDto> registratiobRegistrationStatusDtos =  reprocessorVerticalService.fetchUnProcessedPackets(processList, recordFetchSize, elapseTime,
-							reprocessCount, (!statusValList.isEmpty() ? statusValList : statusList), reprocessExcludeStageNames);
-
-					// Thread-safe update to cache
-					packetCacheMap.compute(key, (k, existingList) -> {
-						if (existingList == null) return new ConcurrentLinkedDeque<>(registratiobRegistrationStatusDtos);
-						existingList.addAll(new ArrayList<>(registratiobRegistrationStatusDtos));
-						return existingList;
-					});
-				}, fetchExecutor))
+					return reprocessorVerticalService.fetchUnProcessedPackets(processList, recordFetchSize, elapseTime,
+							reprocessCount, (!statusValList.isEmpty() ? statusValList : statusList), reprocessExcludeStageNames)
+							.thenAccept(result -> {
+								// Thread-safe update to cache
+								packetCacheMap.compute(key, (k, existingList) -> {
+									if (existingList == null) return new ConcurrentLinkedDeque<>(result);
+									existingList.addAll(new ArrayList<>(result));
+									return existingList;
+								});
+							})
+							.exceptionally(ex -> {
+								regProcLogger.error(LoggerFileConstant.SESSIONID.toString(),
+										"Error Fetching UnprocessedPackets -- ",
+										" Error Triggered for Process [" + String.join(",", processList) + "] and Status [" + String.join(",", (!statusValList.isEmpty() ? statusValList : statusList)) + "]", ExceptionUtils.getStackTrace(ex));
+								return null;
+							});
+				})
 				.collect(Collectors.toList());
 
 		CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();