[FINERACT-1937] refactor isCatchUpRunning SQL

fineract-provider/src/main/java/org/apache/fineract/cob/service/LoanCOBCatchUpServiceImpl.java

@@ -20,7 +20,6 @@ package org.apache.fineract.cob.service;
 
 import com.google.gson.Gson;
 import java.time.LocalDate;
-import java.time.format.DateTimeFormatter;
 import java.util.List;
 import lombok.RequiredArgsConstructor;
 import org.apache.fineract.cob.data.IsCatchUpRunningDTO;
@@ -34,7 +33,6 @@ import org.apache.fineract.infrastructure.core.serialization.GoogleGsonSerialize
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
 import org.apache.fineract.infrastructure.jobs.domain.CustomJobParameterRepository;
 import org.apache.fineract.infrastructure.jobs.domain.JobExecutionRepository;
-import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.stereotype.Service;
 
@@ -77,15 +75,9 @@ public class LoanCOBCatchUpServiceImpl implements LoanCOBCatchUpService {
 
     @Override
     public IsCatchUpRunningDTO isCatchUpRunning() {
-        List<Long> runningCatchUpExecutionIds = jobExecutionRepository.getRunningJobsIdsByExecutionParameter(LoanCOBConstant.JOB_NAME,
+        LocalDate runningCatchUpBusinessDate = jobExecutionRepository.getBusinessDateOfRunningJobByExecutionParameter(
-                LoanCOBConstant.LOAN_COB_CUSTOM_JOB_PARAMETER_KEY, LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME, "true");
+                LoanCOBConstant.JOB_NAME, LoanCOBConstant.LOAN_COB_CUSTOM_JOB_PARAMETER_KEY, LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME,
-        return runningCatchUpExecutionIds //
+                "true", LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME);
-                .stream() //
+        return new IsCatchUpRunningDTO(runningCatchUpBusinessDate != null, runningCatchUpBusinessDate);
-                .findFirst() //
-                .map(jobExplorer::getJobExecution) //
-                .map(JobExecution::getExecutionContext) //
-                .map(executionContext -> executionContext.getString(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME)) //
-                .map(result -> new IsCatchUpRunningDTO(true, LocalDate.parse(result, DateTimeFormatter.ISO_DATE))) //
-                .orElse(new IsCatchUpRunningDTO(false, null));
     }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/domain/JobExecutionRepository.java

@@ -24,34 +24,23 @@ import static org.springframework.batch.core.BatchStatus.STARTED;
 import static org.springframework.batch.core.BatchStatus.STARTING;
 import static org.springframework.batch.core.BatchStatus.UNKNOWN;
 
-import com.google.gson.Gson;
+import java.time.LocalDate;
 import java.util.List;
 import java.util.Map;
 import lombok.RequiredArgsConstructor;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
-import org.apache.fineract.infrastructure.core.serialization.GoogleGsonSerializerHelper;
-import org.apache.fineract.infrastructure.core.service.database.DatabaseSpecificSQLGenerator;
 import org.apache.fineract.infrastructure.core.service.database.DatabaseTypeResolver;
-import org.apache.fineract.infrastructure.jobs.data.JobParameterDTO;
+import org.springframework.dao.EmptyResultDataAccessException;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.stereotype.Component;
 
 @Component
 @RequiredArgsConstructor
-public class JobExecutionRepository implements InitializingBean {
+public class JobExecutionRepository {
 
     private final NamedParameterJdbcTemplate namedParameterJdbcTemplate;
     private final FineractProperties fineractProperties;
     private final DatabaseTypeResolver databaseTypeResolver;
-    private final DatabaseSpecificSQLGenerator sqlGenerator;
-    private final GoogleGsonSerializerHelper gsonFactory;
-    private Gson gson;
-
-    @Override
-    public void afterPropertiesSet() throws Exception {
-        this.gson = gsonFactory.createSimpleGson();
-    }
 
     public List<String> getStuckJobNames(NamedParameterJdbcTemplate jdbcTemplate) {
         int threshold = fineractProperties.getJob().getStuckRetryThreshold();
@@ -165,30 +154,79 @@ public class JobExecutionRepository implements InitializingBean {
                 """, Map.of("status", FAILED.name(), "jobExecutionId", stuckJobId));
     }
 
-    public List<Long> getRunningJobsIdsByExecutionParameter(String jobName, String jobCustomParamKeyName, String parameterKeyName,
+    public LocalDate getBusinessDateOfRunningJobByExecutionParameter(String jobName, String jobCustomParamKeyName, String parameterKeyName,
-            String parameterValue) {
+            String parameterValue, String dateParameterName) {
-        final StringBuilder sqlStatementBuilder = new StringBuilder();
+        try {
-        String jsonString = gson.toJson(new JobParameterDTO(parameterKeyName, parameterValue));
+            if (databaseTypeResolver.isPostgreSQL()) {
-        sqlStatementBuilder.append(
+                return namedParameterJdbcTemplate.queryForObject("""
-                "SELECT bje.JOB_EXECUTION_ID FROM BATCH_JOB_INSTANCE bji INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID INNER JOIN BATCH_JOB_EXECUTION_PARAMS bjep ON bje.JOB_EXECUTION_ID = bjep.JOB_EXECUTION_ID"
+                        SELECT
-                        + " WHERE bje.STATUS IN (:statuses) AND bji.JOB_NAME = :jobName AND bjep.PARAMETER_NAME = :jobCustomParamKeyName AND "
+                                J2->>'parameterValue'
-                        + sqlGenerator.castInteger("bjep.PARAMETER_VALUE") + " IN (" + getSubQueryForCustomJobParameters()
+                        FROM
-                        + ") AND bje.JOB_INSTANCE_ID NOT IN (SELECT bje.JOB_INSTANCE_ID FROM BATCH_JOB_INSTANCE bji INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID"
+                            BATCH_JOB_INSTANCE BJI
-                        + " WHERE bje.STATUS = :completedStatus AND bji.JOB_NAME = :jobName)");
+                                INNER JOIN BATCH_JOB_EXECUTION BJE ON BJI.JOB_INSTANCE_ID = BJE.JOB_INSTANCE_ID
-        return namedParameterJdbcTemplate.queryForList(
+                                INNER JOIN BATCH_JOB_EXECUTION_PARAMS BJEP ON BJE.JOB_EXECUTION_ID = BJEP.JOB_EXECUTION_ID
-                sqlStatementBuilder.toString(), Map.of("statuses", List.of(STARTED.name(), STARTING.name()), "jobName", jobName,
+                                inner join batch_custom_job_parameters CJP ON cast(BJEP.parameter_value as bigint) = CJP.id
-                        "completedStatus", COMPLETED.name(), "jobCustomParamKeyName", jobCustomParamKeyName, "jsonString", jsonString),
+                                AND BJEP.parameter_name = :jobCustomParamKeyName
-                Long.class);
+                                CROSS JOIN LATERAL json_array_elements(CJP.parameter_json) J
-    }
+                                CROSS JOIN LATERAL json_array_elements(CJP.parameter_json) J2
-
+                        WHERE
-    private String getSubQueryForCustomJobParameters() {
+                                    J ->> 'parameterName' = :filterParameterName
-        if (databaseTypeResolver.isMySQL()) {
+                          AND J ->> 'parameterValue' = :filterParameterValue
-            return "SELECT cjp.id FROM batch_custom_job_parameters cjp WHERE JSON_CONTAINS(cjp.parameter_json,:jsonString)";
+                          AND J2 ->> 'parameterName' = :dateParameterName
-        } else if (databaseTypeResolver.isPostgreSQL()) {
+                          AND BJE.STATUS IN (:statuses)
-            return "SELECT cjp.id FROM (SELECT id,json_array_elements(parameter_json) AS json_data FROM batch_custom_job_parameters) AS cjp WHERE (cjp.json_data ::jsonb @> :jsonString ::jsonb)";
+                          AND BJI.JOB_NAME = :jobName
-        } else {
+                          AND BJE.JOB_INSTANCE_ID NOT IN (
-            throw new IllegalStateException("Database type is not supported for json query " + databaseTypeResolver.databaseType());
+                            SELECT
+                                IBJE.JOB_INSTANCE_ID
+                            FROM
+                                BATCH_JOB_INSTANCE IBJI
+                                    INNER JOIN BATCH_JOB_EXECUTION IBJE ON IBJI.JOB_INSTANCE_ID = IBJE.JOB_INSTANCE_ID
+                            WHERE
+                                    IBJE.STATUS = :completedStatus
+                              AND IBJI.JOB_NAME = :jobName)
+                        """,
+                        Map.of("jobCustomParamKeyName", jobCustomParamKeyName, "filterParameterName", parameterKeyName,
+                                "filterParameterValue", parameterValue, "statuses", List.of(STARTED.name(), STARTING.name()),
+                                "completedStatus", COMPLETED.name(), "jobName", jobName, "dateParameterName", dateParameterName),
+                        LocalDate.class);
+            } else if (databaseTypeResolver.isMySQL()) {
+                return namedParameterJdbcTemplate.queryForObject(
+                        """
+                                SELECT
+                                     J2.parameter_value
+                                FROM
+                                    BATCH_JOB_INSTANCE BJI
+                                        INNER JOIN BATCH_JOB_EXECUTION BJE ON BJI.JOB_INSTANCE_ID = BJE.JOB_INSTANCE_ID
+                                        INNER JOIN BATCH_JOB_EXECUTION_PARAMS BJEP ON BJE.JOB_EXECUTION_ID = BJEP.JOB_EXECUTION_ID
+                                        inner join batch_custom_job_parameters CJP ON BJEP.parameter_value = CJP.id
+                                        AND BJEP.parameter_name = :jobCustomParamKeyName
+                                        CROSS JOIN json_table(CJP.parameter_json, '$[*]' COLUMNS(parameter_name VARCHAR(100) PATH "$.parameterName", parameter_value VARCHAR(100) PATH "$.parameterValue")) J
+                                        CROSS JOIN json_table(CJP.parameter_json, '$[*]' COLUMNS(parameter_name VARCHAR(100) PATH "$.parameterName", parameter_value VARCHAR(100) PATH "$.parameterValue")) J2
+                                WHERE
+                                        J.parameter_name = :filterParameterName
+                                  AND J.parameter_value = :filterParameterValue
+                                  AND J2.parameter_name = :dateParameterName
+                                  AND BJE.STATUS IN (:statuses)
+                                  AND BJI.JOB_NAME = :jobName
+                                  AND BJE.JOB_INSTANCE_ID NOT IN (
+                                    SELECT
+                                        IBJE.JOB_INSTANCE_ID
+                                    FROM
+                                        BATCH_JOB_INSTANCE IBJI
+                                            INNER JOIN BATCH_JOB_EXECUTION IBJE ON IBJI.JOB_INSTANCE_ID = IBJE.JOB_INSTANCE_ID
+                                    WHERE
+                                            IBJE.STATUS = :completedStatus
+                                      AND IBJI.JOB_NAME = :jobName);
+                                """,
+                        Map.of("jobCustomParamKeyName", jobCustomParamKeyName, "filterParameterName", parameterKeyName,
+                                "filterParameterValue", parameterValue, "dateParameterName", dateParameterName, "statuses",
+                                List.of(STARTED.name(), STARTING.name()), "jobName", jobName, "completedStatus", COMPLETED.name()),
+                        LocalDate.class);
+            } else {
+                throw new IllegalStateException("Database type is not supported for json query " + databaseTypeResolver.databaseType());
+            }
+        } catch (EmptyResultDataAccessException e) {
+            return null;
         }
     }
-
 }