DTS-50661: Review comment fixes, removed redundant ex try,catch and recommendation jobStrategy fix

Change-log: Batch processing ai-recommendation for performance improvement.

Author: shunaray

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/recommendationcalculation/config/RecommendationCalculationBatchConfig.java

@@ -70,36 +70,21 @@ public class RecommendationCalculationService {
 	public RecommendationsActionPlan calculateRecommendationsForProject(@NonNull ProjectInputDTO projectInput) {
 		Persona persona = recommendationCalculationConfig.getCalculationConfig().getEnabledPersona();
 
-		try {
-			log.info("{} Calculating recommendations for project: {} ({}) - Persona: {}",
-					JobConstants.LOG_PREFIX_RECOMMENDATION, projectInput.name(), projectInput.nodeId(),
-					persona.getDisplayName());
-
-			// Delegate KPI data extraction to specialized service
-			Map<String, Object> kpiData = kpiDataExtractionService.fetchKpiDataForProject(projectInput);
-
-			// Build prompt using PromptService with actual KPI data
-			String prompt = promptService.getKpiRecommendationPrompt(kpiData, persona);
-
-			ChatGenerationRequest request = ChatGenerationRequest.builder().prompt(prompt).build();
-
-			ChatGenerationResponseDTO response = aiGatewayClient.generate(request);
-
-			return buildRecommendationsActionPlan(projectInput, persona, response);
-		} catch (IllegalStateException e) {
-			// Parsing or validation failures - rethrow as-is
-			log.error("{} Validation error for project {} persona {}: {}",
-					JobConstants.LOG_PREFIX_RECOMMENDATION, projectInput.nodeId(), persona.getDisplayName(),
-					e.getMessage(), e);
-			throw e;
-		} catch (RuntimeException e) {
-			// API call failures, network issues - wrap with context
-			log.error("{} Runtime error calculating recommendations for project {} persona {}: {}",
-					JobConstants.LOG_PREFIX_RECOMMENDATION, projectInput.nodeId(), persona.getDisplayName(),
-					e.getMessage(), e);
-			throw new IllegalStateException("Failed to calculate recommendations for project: " + projectInput.nodeId(),
-					e);
-		}
+		log.info("{} Calculating recommendations for project: {} ({}) - Persona: {}",
+				JobConstants.LOG_PREFIX_RECOMMENDATION, projectInput.name(), projectInput.nodeId(),
+				persona.getDisplayName());
+
+		// Delegate KPI data extraction to specialized service
+		Map<String, Object> kpiData = kpiDataExtractionService.fetchKpiDataForProject(projectInput);
+
+		// Build prompt using PromptService with actual KPI data
+		String prompt = promptService.getKpiRecommendationPrompt(kpiData, persona);
+
+		ChatGenerationRequest request = ChatGenerationRequest.builder().prompt(prompt).build();
+
+		ChatGenerationResponseDTO response = aiGatewayClient.generate(request);
+
+		return buildRecommendationsActionPlan(projectInput, persona, response);
 	}
 
 	/**

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/recommendationcalculation/service/RecommendationCalculationService.java

@@ -25,14 +25,24 @@
 import org.springframework.batch.core.job.builder.JobBuilder;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.batch.integration.async.AsyncItemProcessor;
+import org.springframework.batch.integration.async.AsyncItemWriter;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.PlatformTransactionManager;
 
 import com.publicissapient.kpidashboard.common.model.recommendation.batch.RecommendationsActionPlan;
+import com.publicissapient.kpidashboard.common.repository.recommendation.RecommendationRepository;
+import com.publicissapient.kpidashboard.common.service.JobExecutionTraceLogService;
+import com.publicissapient.kpidashboard.common.service.ProcessorExecutionTraceLogService;
 import com.publicissapient.kpidashboard.job.config.base.SchedulingConfig;
-import com.publicissapient.kpidashboard.job.recommendationcalculation.config.RecommendationCalculationBatchConfig;
 import com.publicissapient.kpidashboard.job.recommendationcalculation.config.RecommendationCalculationConfig;
 import com.publicissapient.kpidashboard.job.recommendationcalculation.listener.RecommendationCalculationJobExecutionListener;
+import com.publicissapient.kpidashboard.job.recommendationcalculation.processor.ProjectItemProcessor;
+import com.publicissapient.kpidashboard.job.recommendationcalculation.reader.ProjectItemReader;
+import com.publicissapient.kpidashboard.job.recommendationcalculation.service.RecommendationCalculationService;
+import com.publicissapient.kpidashboard.job.recommendationcalculation.service.RecommendationProjectBatchService;
+import com.publicissapient.kpidashboard.job.recommendationcalculation.writer.ProjectItemWriter;
 import com.publicissapient.kpidashboard.job.shared.dto.ProjectInputDTO;
 import com.publicissapient.kpidashboard.job.strategy.JobStrategy;
 
@@ -46,11 +56,15 @@
 @RequiredArgsConstructor
 public class RecommendationCalculationJobStrategy implements JobStrategy {
 
-	private final RecommendationCalculationConfig recommendationCalculationConfig;
-	private final RecommendationCalculationBatchConfig batchConfig;
-	private final PlatformTransactionManager platformTransactionManager;
 	private final JobRepository jobRepository;
-	private final RecommendationCalculationJobExecutionListener jobExecutionListener;
+	private final TaskExecutor taskExecutor;
+	private final PlatformTransactionManager platformTransactionManager;
+	private final RecommendationCalculationConfig recommendationCalculationConfig;
+	private final RecommendationProjectBatchService projectBatchService;
+	private final RecommendationCalculationService recommendationCalculationService;
+	private final JobExecutionTraceLogService jobExecutionTraceLogService;
+	private final ProcessorExecutionTraceLogService processorExecutionTraceLogService;
+	private final RecommendationRepository recommendationRepository;
 
 	@Override
 	public String getJobName() {
@@ -65,18 +79,33 @@ public Optional<SchedulingConfig> getSchedulingConfig() {
 	@Override
 	public Job getJob() {
 		return new JobBuilder(recommendationCalculationConfig.getName(), jobRepository).start(chunkProcessProjects())
-				.listener(jobExecutionListener).build();
+				.listener(new RecommendationCalculationJobExecutionListener(this.projectBatchService,
+						this.jobExecutionTraceLogService))
+				.build();
 	}
 
 	private Step chunkProcessProjects() {
 		return new StepBuilder(String.format("%s-chunk-process", recommendationCalculationConfig.getName()),
 				jobRepository)
 				.<ProjectInputDTO, Future<RecommendationsActionPlan>>chunk(
 						recommendationCalculationConfig.getBatching().getChunkSize(), platformTransactionManager)
+				.reader(new ProjectItemReader(this.projectBatchService)).processor(asyncProjectProcessor())
+				.writer(asyncItemWriter()).build();
+	}
+
+	private AsyncItemProcessor<ProjectInputDTO, RecommendationsActionPlan> asyncProjectProcessor() {
+		AsyncItemProcessor<ProjectInputDTO, RecommendationsActionPlan> asyncItemProcessor = new AsyncItemProcessor<>();
+		asyncItemProcessor.setDelegate(new ProjectItemProcessor(this.recommendationCalculationService,
+				this.processorExecutionTraceLogService));
+		asyncItemProcessor.setTaskExecutor(taskExecutor);
+		return asyncItemProcessor;
+	}
 
-				.reader(batchConfig.recommendationProjectItemReader())
-				.processor(batchConfig.recommendationAsyncProjectProcessor())
-				.writer(batchConfig.recommendationAsyncItemWriter()).build();
+	private AsyncItemWriter<RecommendationsActionPlan> asyncItemWriter() {
+		AsyncItemWriter<RecommendationsActionPlan> writer = new AsyncItemWriter<>();
+		writer.setDelegate(
+				new ProjectItemWriter(this.recommendationRepository, this.processorExecutionTraceLogService));
+		return writer;
 	}
 
 }