Develop

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/client/customapi/KnowHOWClient.java

@@ -74,6 +74,22 @@ public List<KpiElement> getKpiIntegrationValues(List<KpiRequest> kpiRequests) {
 		}).flatMapIterable(list -> list).collectList().block();
 	}
 
+	public List<KpiElement> getKpiIntegrationValuesKanban(List<KpiRequest> kpiRequests) {
+		return Flux.fromIterable(kpiRequests).publishOn(Schedulers.boundedElastic()).flatMap(kpiRequest -> {
+			try {
+				semaphore.acquire();
+				return this.knowHOWWebClient.post()
+						.uri(this.knowHOWApiClientConfig.getKpiIntegrationValuesKanbanEndpointConfig().getPath())
+						.bodyValue(kpiRequest).retrieve().bodyToFlux(KpiElement.class).retryWhen(retrySpec())
+						.collectList().doFinally(signalType -> semaphore.release());
+			} catch (InterruptedException e) {
+				log.error("Could not get kpi integration values kanban for kpiRequest {}", kpiRequest);
+				Thread.currentThread().interrupt();
+				return Flux.error(e);
+			}
+		}).flatMapIterable(list -> list).collectList().block();
+	}
+
 	private RetryBackoffSpec retrySpec() {
 		return Retry
 				.backoff(knowHOWApiClientConfig.getRetryPolicy().getMaxAttempts(),

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/client/customapi/config/KnowHOWApiClientConfig.java

@@ -50,4 +50,8 @@ public static class EndpointConfig {
     public EndpointConfig getKpiIntegrationValuesEndpointConfig() {
         return this.endpoints.get("kpi-integration-values");
     }
+
+    public EndpointConfig getKpiIntegrationValuesKanbanEndpointConfig() {
+        return this.endpoints.get("kpi-integration-values-kanban");
+    }
 }

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/client/shareddataservice/SharedDataServiceClient.java

@@ -17,15 +17,20 @@
 package com.publicissapient.kpidashboard.client.shareddataservice;
 
 import com.publicissapient.kpidashboard.client.shareddataservice.config.SharedDataServiceConfig;
-import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.PagedAIUsagePerOrgLevel;
+import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.AIUsagePerOrgLevel;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.stereotype.Component;
 import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClientResponseException;
 import reactor.util.retry.Retry;
+import reactor.util.retry.RetryBackoffSpec;
 
+import java.net.ConnectException;
 import java.time.Duration;
 
+@Slf4j
 @Component
 public class SharedDataServiceClient {
     private final WebClient webClient;
@@ -47,9 +52,15 @@ public SharedDataServiceClient(SharedDataServiceConfig sharedDataServiceConfig)
                 .build();
     }
 
-    public PagedAIUsagePerOrgLevel getAIUsageStatsAsync(String levelName) {
-        int maxAttempts = sharedDataServiceConfig.getRetryPolicy().getMaxAttempts();
-        int minBackoffDuration = sharedDataServiceConfig.getRetryPolicy().getMinBackoffDuration();
+    public AIUsagePerOrgLevel getAIUsageStatsAsync(String levelName) {
+        RetryBackoffSpec retrySpec = Retry.backoff(
+                sharedDataServiceConfig.getRetryPolicy().getMaxAttempts(),
+                Duration.of(sharedDataServiceConfig.getRetryPolicy().getMinBackoffDuration(),
+                        sharedDataServiceConfig.getRetryPolicy().getMinBackoffTimeUnit().toChronoUnit()))
+                .filter(SharedDataServiceClient::shouldRetry)
+                .doBeforeRetry(retrySignal ->
+                        log.info("Retry #{} due to {}", retrySignal.totalRetries(), retrySignal.failure().toString()));
+
         String path = sharedDataServiceConfig.getAiUsageStatisticsEndpoint().getPath();
 
         return webClient.get()
@@ -58,8 +69,15 @@ public PagedAIUsagePerOrgLevel getAIUsageStatsAsync(String levelName) {
                         .queryParam(LEVEL_NAME_PARAM, levelName)
                         .build())
                 .retrieve()
-                .bodyToMono(PagedAIUsagePerOrgLevel.class)
-                .retryWhen(Retry.backoff(maxAttempts, Duration.ofSeconds(minBackoffDuration)).jitter(0.5))
+                .bodyToMono(AIUsagePerOrgLevel.class)
+                .retryWhen(retrySpec)
                 .block();
     }
+
+    private static boolean shouldRetry(Throwable throwable) {
+        if (throwable instanceof WebClientResponseException ex) {
+            return ex.getStatusCode().is5xxServerError();
+        }
+        return throwable instanceof ConnectException;
+    }
 }

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/aiusagestatisticscollector/dto/PagedAIUsagePerOrgLevel.java → ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/aiusagestatisticscollector/dto/AIUsagePerOrgLevel.java

@@ -19,13 +19,13 @@
 import java.time.Instant;
 import java.util.List;
 
-public record PagedAIUsagePerOrgLevel(String levelType,
-                                      String levelName,
-                                      Instant statsDate,
-                                      AIUsageSummary usageSummary,
-                                      List<AIUsagePerUser> users,
-                                      int currentPage,
-                                      int totalPages,
-                                      long totalElements,
-                                      int pageSize) {
+public record AIUsagePerOrgLevel(String levelType,
+                                 String levelName,
+                                 Instant statsDate,
+                                 AIUsageSummary usageSummary,
+                                 List<AIUsagePerUser> users,
+                                 int currentPage,
+                                 int totalPages,
+                                 long totalElements,
+                                 int pageSize) {
 }

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/aiusagestatisticscollector/dto/mapper/AIUsageStatisticsMapper.java

@@ -16,7 +16,7 @@
 
 package com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.mapper;
 
-import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.PagedAIUsagePerOrgLevel;
+import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.AIUsagePerOrgLevel;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.model.AIUsageStatistics;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
@@ -28,5 +28,5 @@ public interface AIUsageStatisticsMapper {
 
     @Mapping(target = "users", ignore = true)
     @Mapping(target = "ingestTimestamp", expression = "java(java.time.Instant.now())")
-    AIUsageStatistics toEntity(PagedAIUsagePerOrgLevel pagedAIUsagePerOrgLevel);
+    AIUsageStatistics toEntity(AIUsagePerOrgLevel aIUsagePerOrgLevel);
 }

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/aiusagestatisticscollector/processor/AccountItemProcessor.java

@@ -18,7 +18,7 @@
 
 import org.springframework.batch.item.ItemProcessor;
 
-import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.PagedAIUsagePerOrgLevel;
+import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.AIUsagePerOrgLevel;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.model.AIUsageStatistics;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.service.AIUsageStatisticsService;
 
@@ -28,12 +28,17 @@
 
 @Slf4j
 @AllArgsConstructor
-public class AccountItemProcessor implements ItemProcessor<PagedAIUsagePerOrgLevel, AIUsageStatistics> {
+public class AccountItemProcessor implements ItemProcessor<AIUsagePerOrgLevel, AIUsageStatistics> {
     private final AIUsageStatisticsService aiUsageStatisticsService;
 
     @Override
-    public AIUsageStatistics process(@Nonnull PagedAIUsagePerOrgLevel item) {
+    public AIUsageStatistics process(@Nonnull AIUsagePerOrgLevel item) {
         log.debug("[ai-usage-statistics-collector job] Fetching AI usage statistics for level name: {}", item.levelName());
-        return aiUsageStatisticsService.fetchAIUsageStatistics(item.levelName());
+        try {
+            return aiUsageStatisticsService.fetchAIUsageStatistics(item.levelName());
+        } catch (Exception ex) {
+            log.error("Failed fetching AI stats for {} – skipping", item.levelName());
+            throw ex;
+        }
     }
 }

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/aiusagestatisticscollector/reader/AccountItemReader.java

@@ -16,23 +16,27 @@
 
 package com.publicissapient.kpidashboard.job.aiusagestatisticscollector.reader;
 
+import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.AIUsagePerOrgLevel;
 import org.springframework.batch.item.ItemReader;
 
-import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.PagedAIUsagePerOrgLevel;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.service.AccountBatchService;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 @RequiredArgsConstructor
-public class AccountItemReader implements ItemReader<PagedAIUsagePerOrgLevel> {
+public class AccountItemReader implements ItemReader<AIUsagePerOrgLevel> {
 
     private final AccountBatchService accountBatchService;
 
     @Override
-    public PagedAIUsagePerOrgLevel read() {
-        PagedAIUsagePerOrgLevel aiUsageStatistics = accountBatchService.getNextAccountPage();
+    public AIUsagePerOrgLevel read() {
+        AIUsagePerOrgLevel aiUsageStatistics = accountBatchService.getNextAccount();
+        if (aiUsageStatistics == null) {
+            log.info("No more accounts.");
+            return null;
+        }
         log.info("[ai-usage-statistics-collector job] Reader fetched level name: {}", aiUsageStatistics.levelName());
         return aiUsageStatistics;
     }

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/aiusagestatisticscollector/service/AIUsageStatisticsService.java

@@ -20,7 +20,7 @@
 
 import com.publicissapient.kpidashboard.client.shareddataservice.SharedDataServiceClient;
 import com.publicissapient.kpidashboard.exception.InternalServerErrorException;
-import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.PagedAIUsagePerOrgLevel;
+import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.AIUsagePerOrgLevel;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.model.AIUsageStatistics;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.repository.AIUsageStatisticsRepository;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.mapper.AIUsageStatisticsMapper;
@@ -40,7 +40,7 @@ public class AIUsageStatisticsService {
 
     public AIUsageStatistics fetchAIUsageStatistics(String levelName) {
         try {
-            PagedAIUsagePerOrgLevel aiUsageStatistics = sharedDataServiceClient.getAIUsageStatsAsync(levelName);
+            AIUsagePerOrgLevel aiUsageStatistics = sharedDataServiceClient.getAIUsageStatsAsync(levelName);
             return aiUsageStatisticsMapper.toEntity(aiUsageStatistics);
         } catch (Exception ex) {
             log.error("Failed to fetch AI usage stats for {}: {}", levelName, ex.getMessage());
@@ -51,6 +51,6 @@ public AIUsageStatistics fetchAIUsageStatistics(String levelName) {
     @Transactional
     public void saveAll(List<AIUsageStatistics> aiUsageStatisticsList) {
         aiUsageStatisticsRepository.saveAll(aiUsageStatisticsList);
-        log.info("Successfully fetched and saved {} AI usage statistics", aiUsageStatisticsList.size());
+        log.info("Successfully fetched and saved {} AI usage statistics for account: {}", aiUsageStatisticsList.size(), aiUsageStatisticsList.get(0).getLevelName());
     }
 }

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/aiusagestatisticscollector/service/AccountBatchService.java

@@ -18,7 +18,7 @@
 
 import com.publicissapient.kpidashboard.common.model.application.AccountHierarchy;
 import com.publicissapient.kpidashboard.common.repository.application.AccountHierarchyRepository;
-import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.PagedAIUsagePerOrgLevel;
+import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.AIUsagePerOrgLevel;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -62,10 +62,10 @@ public void initializeBatchProcessingParametersForTheNextProcess() {
     }
 
     /**
-     * Return next account to process as a PagedAIUsagePerOrgLevel.
+     * Return next account to process as a AIUsagePerOrgLevel.
      * Each page contains exactly 1 account because the endpoint supports only 1 account per request.
      */
-    public PagedAIUsagePerOrgLevel getNextAccountPage() {
+    public AIUsagePerOrgLevel getNextAccount() {
         if (!initialized) {
             initializeBatchProcessingParametersForTheNextProcess();
         }
@@ -76,7 +76,7 @@ public PagedAIUsagePerOrgLevel getNextAccountPage() {
 
         AccountHierarchy account = allAccounts.get(currentIndex);
 
-        PagedAIUsagePerOrgLevel page = new PagedAIUsagePerOrgLevel(
+        AIUsagePerOrgLevel page = new AIUsagePerOrgLevel(
                 "account",
                 account.getNodeName(),
                 Instant.now(),

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/aiusagestatisticscollector/strategy/AIUsageStatisticsJobStrategy.java

@@ -19,6 +19,7 @@
 import java.util.Optional;
 import java.util.concurrent.Future;
 
+import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.AIUsagePerOrgLevel;
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.job.builder.JobBuilder;
@@ -32,7 +33,6 @@
 
 import com.publicissapient.kpidashboard.common.service.ProcessorExecutionTraceLogServiceImpl;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.config.AIUsageStatisticsCollectorJobConfig;
-import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.dto.PagedAIUsagePerOrgLevel;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.listener.AIUsageStatisticsJobCompletionListener;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.model.AIUsageStatistics;
 import com.publicissapient.kpidashboard.job.aiusagestatisticscollector.processor.AccountItemProcessor;
@@ -85,18 +85,21 @@ public Optional<SchedulingConfig> getSchedulingConfig() {
     }
 
     private Step chunkProcessAIUsageStatisticsForAccounts() {
-
         return new StepBuilder("process-ai-usage-statistics", jobRepository)
-                .<PagedAIUsagePerOrgLevel, Future<AIUsageStatistics>>chunk(
+                .<AIUsagePerOrgLevel, Future<AIUsageStatistics>>chunk(
                         aiUsageStatisticsCollectorJobConfig.getBatching().getChunkSize(), transactionManager)
+                .faultTolerant()
+                .skip(Exception.class)
+                .skipLimit(1000)
+                .noRetry(Exception.class)
                 .reader(new AccountItemReader(accountBatchService))
                 .processor(asyncAccountProcessor())
                 .writer(asyncItemWriter())
                 .build();
     }
 
-    private AsyncItemProcessor<PagedAIUsagePerOrgLevel, AIUsageStatistics> asyncAccountProcessor() {
-        AsyncItemProcessor<PagedAIUsagePerOrgLevel, AIUsageStatistics> asyncItemProcessor = new AsyncItemProcessor<>();
+    private AsyncItemProcessor<AIUsagePerOrgLevel, AIUsageStatistics> asyncAccountProcessor() {
+        AsyncItemProcessor<AIUsagePerOrgLevel, AIUsageStatistics> asyncItemProcessor = new AsyncItemProcessor<>();
         asyncItemProcessor.setDelegate(new AccountItemProcessor(this.aiUsageStatisticsService));
         asyncItemProcessor.setTaskExecutor(taskExecutor);
         return asyncItemProcessor;

ai-data-processor/src/main/java/com/publicissapient/kpidashboard/job/kpimaturitycalculation/processor/ProjectItemProcessor.java

@@ -34,7 +34,8 @@ public class ProjectItemProcessor implements ItemProcessor<ProjectInputDTO, KpiM
 
     @Override
     public KpiMaturity process(@Nonnull ProjectInputDTO item) {
-        log.info("[kpi-maturity-calculation job] Starting kpi metrics calculation for project with nodeId: {}", item.nodeId());
+        log.info("[kpi-maturity-calculation job] Starting kpi metrics calculation for project with nodeId: {} and deliveryMethodology: {}", item.nodeId(), item
+                .deliveryMethodology());
 
         return kpiMaturityCalculationService.calculateKpiMaturityForProject(item);
     }