FINERACT-1724: Fixed Generate Loss Loan provisioning job + enhanced the stuck job listener to be smarter
This commit is contained in:
committed by
Arnold Gálovics
parent
8c12fbed37
commit
0e96b71f0e
+9
@@ -50,6 +50,8 @@ public class FineractProperties {
|
|||||||
|
|
||||||
private FineractReportProperties report;
|
private FineractReportProperties report;
|
||||||
|
|
||||||
|
private FineractJobProperties job;
|
||||||
|
|
||||||
@Getter
|
@Getter
|
||||||
@Setter
|
@Setter
|
||||||
public static class FineractTenantProperties {
|
public static class FineractTenantProperties {
|
||||||
@@ -225,4 +227,11 @@ public class FineractProperties {
|
|||||||
private String bucketName;
|
private String bucketName;
|
||||||
private Boolean enabled;
|
private Boolean enabled;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Getter
|
||||||
|
@Setter
|
||||||
|
public static class FineractJobProperties {
|
||||||
|
|
||||||
|
private int stuckRetryThreshold;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+128
-46
@@ -22,10 +22,12 @@ import static org.springframework.batch.core.BatchStatus.COMPLETED;
|
|||||||
import static org.springframework.batch.core.BatchStatus.FAILED;
|
import static org.springframework.batch.core.BatchStatus.FAILED;
|
||||||
import static org.springframework.batch.core.BatchStatus.STARTED;
|
import static org.springframework.batch.core.BatchStatus.STARTED;
|
||||||
import static org.springframework.batch.core.BatchStatus.STARTING;
|
import static org.springframework.batch.core.BatchStatus.STARTING;
|
||||||
|
import static org.springframework.batch.core.BatchStatus.UNKNOWN;
|
||||||
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
|
import org.apache.fineract.infrastructure.core.config.FineractProperties;
|
||||||
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
|
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
@@ -34,68 +36,148 @@ import org.springframework.stereotype.Component;
|
|||||||
public class JobExecutionRepository {
|
public class JobExecutionRepository {
|
||||||
|
|
||||||
private final NamedParameterJdbcTemplate namedParameterJdbcTemplate;
|
private final NamedParameterJdbcTemplate namedParameterJdbcTemplate;
|
||||||
|
private final FineractProperties fineractProperties;
|
||||||
|
|
||||||
public List<String> getStuckJobNames(NamedParameterJdbcTemplate jdbcTemplate) {
|
public List<String> getStuckJobNames(NamedParameterJdbcTemplate jdbcTemplate) {
|
||||||
return jdbcTemplate.queryForList(
|
int threshold = fineractProperties.getJob().getStuckRetryThreshold();
|
||||||
"SELECT bji.JOB_NAME as STUCK_JOB_NAME FROM BATCH_JOB_INSTANCE bji "
|
return jdbcTemplate.queryForList("""
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
|
SELECT DISTINCT(bji.JOB_NAME) as STUCK_JOB_NAME
|
||||||
+ "WHERE bje.STATUS IN (:statuses) AND bje.JOB_INSTANCE_ID NOT IN ("
|
FROM BATCH_JOB_INSTANCE bji
|
||||||
+ "SELECT bje.JOB_INSTANCE_ID FROM BATCH_JOB_INSTANCE bji "
|
INNER JOIN BATCH_JOB_EXECUTION bje
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
|
ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID
|
||||||
+ "WHERE bje.STATUS = :completedStatus)",
|
WHERE
|
||||||
Map.of("statuses", List.of(STARTED.name(), FAILED.name()), "completedStatus", COMPLETED.name()), String.class);
|
bje.STATUS IN (:statuses)
|
||||||
|
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
|
||||||
|
WHERE bje.STATUS IN (:completedStatuses)
|
||||||
|
)
|
||||||
|
GROUP BY BJI.JOB_INSTANCE_ID
|
||||||
|
HAVING COUNT(BJI.JOB_INSTANCE_ID) <= :threshold
|
||||||
|
""", Map.of("statuses", List.of(STARTED.name()), "completedStatuses",
|
||||||
|
List.of(COMPLETED.name(), FAILED.name(), UNKNOWN.name()), "threshold", threshold), String.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
public Long getStuckJobCountByJobName(String jobName) {
|
public Long getStuckJobCountByJobName(String jobName) {
|
||||||
return namedParameterJdbcTemplate.queryForObject(
|
int threshold = fineractProperties.getJob().getStuckRetryThreshold();
|
||||||
"SELECT COUNT(*) as STUCK_JOB_COUNT FROM BATCH_JOB_INSTANCE bji "
|
return namedParameterJdbcTemplate.queryForObject("""
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
|
SELECT COUNT(DISTINCT bji.JOB_NAME) as STUCK_JOB_COUNT
|
||||||
+ "WHERE bje.STATUS IN (:statuses) AND bji.JOB_NAME = :jobName AND bje.JOB_INSTANCE_ID NOT IN ("
|
FROM BATCH_JOB_INSTANCE bji
|
||||||
+ "SELECT bje.JOB_INSTANCE_ID FROM BATCH_JOB_INSTANCE bji "
|
INNER JOIN BATCH_JOB_EXECUTION bje
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
|
ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID
|
||||||
+ "WHERE bje.STATUS = :completedStatus AND bji.JOB_NAME = :jobName)",
|
WHERE
|
||||||
Map.of("statuses", List.of(STARTED.name(), FAILED.name()), "jobName", jobName, "completedStatus", COMPLETED.name()),
|
bje.STATUS IN (:statuses)
|
||||||
Long.class);
|
AND
|
||||||
|
bji.JOB_NAME = :jobName
|
||||||
|
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
|
||||||
|
WHERE
|
||||||
|
bje.STATUS IN (:completedStatuses)
|
||||||
|
AND
|
||||||
|
bji.JOB_NAME = :jobName
|
||||||
|
)
|
||||||
|
GROUP BY BJI.JOB_INSTANCE_ID
|
||||||
|
HAVING COUNT(BJI.JOB_INSTANCE_ID) <= :threshold
|
||||||
|
""", Map.of("statuses", List.of(STARTED.name()), "jobName", jobName, "completedStatuses",
|
||||||
|
List.of(COMPLETED.name(), FAILED.name(), UNKNOWN.name()), "threshold", threshold), Long.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
public List<Long> getStuckJobIdsByJobName(String jobName) {
|
public List<Long> getStuckJobIdsByJobName(String jobName) {
|
||||||
return namedParameterJdbcTemplate.queryForList(
|
int threshold = fineractProperties.getJob().getStuckRetryThreshold();
|
||||||
"SELECT bje.JOB_EXECUTION_ID FROM BATCH_JOB_INSTANCE bji "
|
return namedParameterJdbcTemplate.queryForList("""
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
|
SELECT bje.JOB_EXECUTION_ID
|
||||||
+ "WHERE bje.STATUS IN (:statuses) AND bji.JOB_NAME = :jobName AND bje.JOB_INSTANCE_ID NOT IN ("
|
FROM BATCH_JOB_INSTANCE bji
|
||||||
+ "SELECT bje.JOB_INSTANCE_ID FROM BATCH_JOB_INSTANCE bji "
|
INNER JOIN BATCH_JOB_EXECUTION bje
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
|
ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID
|
||||||
+ "WHERE bje.STATUS = :completedStatus AND bji.JOB_NAME = :jobName)",
|
WHERE
|
||||||
Map.of("statuses", List.of(STARTED.name(), FAILED.name()), "jobName", jobName, "completedStatus", COMPLETED.name()),
|
bje.STATUS IN (:statuses)
|
||||||
Long.class);
|
AND
|
||||||
|
bji.JOB_NAME = :jobName
|
||||||
|
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
|
||||||
|
WHERE
|
||||||
|
bje.STATUS IN (:completedStatuses)
|
||||||
|
AND
|
||||||
|
bji.JOB_NAME = :jobName
|
||||||
|
)
|
||||||
|
GROUP BY BJI.JOB_INSTANCE_ID, BJE.JOB_EXECUTION_ID
|
||||||
|
HAVING COUNT(BJI.JOB_INSTANCE_ID) <= :threshold
|
||||||
|
""", Map.of("statuses", List.of(STARTED.name()), "jobName", jobName, "completedStatuses",
|
||||||
|
List.of(COMPLETED.name(), FAILED.name(), UNKNOWN.name()), "threshold", threshold), Long.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
public Long getNotCompletedPartitionsCount(Long jobExecutionId, String partitionerStepName) {
|
public Long getNotCompletedPartitionsCount(Long jobExecutionId, String partitionerStepName) {
|
||||||
return namedParameterJdbcTemplate.queryForObject(
|
return namedParameterJdbcTemplate.queryForObject("""
|
||||||
"SELECT COUNT(bse.STEP_EXECUTION_ID) FROM BATCH_STEP_EXECUTION bse "
|
SELECT COUNT(bse.STEP_EXECUTION_ID)
|
||||||
+ "WHERE bse.JOB_EXECUTION_ID = :jobExecutionId AND bse.STEP_NAME <> :stepName AND bse.status <> :status",
|
FROM BATCH_STEP_EXECUTION bse
|
||||||
Map.of("jobExecutionId", jobExecutionId, "stepName", partitionerStepName, "status", COMPLETED.name()), Long.class);
|
WHERE
|
||||||
|
bse.JOB_EXECUTION_ID = :jobExecutionId
|
||||||
|
AND
|
||||||
|
bse.STEP_NAME <> :stepName
|
||||||
|
AND
|
||||||
|
bse.status <> :status
|
||||||
|
""", Map.of("jobExecutionId", jobExecutionId, "stepName", partitionerStepName, "status", COMPLETED.name()), Long.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
public void updateJobStatusToFailed(Long stuckJobId, String partitionerStepName) {
|
public void updateJobStatusToFailed(Long stuckJobId, String partitionerStepName) {
|
||||||
namedParameterJdbcTemplate.update(
|
namedParameterJdbcTemplate.update("""
|
||||||
"UPDATE BATCH_STEP_EXECUTION SET STATUS = :status WHERE JOB_EXECUTION_ID = :jobExecutionId AND STEP_NAME = :stepName",
|
UPDATE BATCH_STEP_EXECUTION
|
||||||
Map.of("status", FAILED.name(), "jobExecutionId", stuckJobId, "stepName", partitionerStepName));
|
SET STATUS = :status
|
||||||
namedParameterJdbcTemplate.update(
|
WHERE
|
||||||
"UPDATE BATCH_JOB_EXECUTION SET STATUS = :status, START_TIME = null, END_TIME = null WHERE JOB_EXECUTION_ID = :jobExecutionId",
|
JOB_EXECUTION_ID = :jobExecutionId
|
||||||
Map.of("status", FAILED.name(), "jobExecutionId", stuckJobId));
|
AND
|
||||||
|
STEP_NAME = :stepName
|
||||||
|
""", Map.of("status", FAILED.name(), "jobExecutionId", stuckJobId, "stepName", partitionerStepName));
|
||||||
|
namedParameterJdbcTemplate.update("""
|
||||||
|
UPDATE BATCH_JOB_EXECUTION
|
||||||
|
SET
|
||||||
|
STATUS = :status,
|
||||||
|
START_TIME = null,
|
||||||
|
END_TIME = null
|
||||||
|
WHERE
|
||||||
|
JOB_EXECUTION_ID = :jobExecutionId
|
||||||
|
""", Map.of("status", FAILED.name(), "jobExecutionId", stuckJobId));
|
||||||
}
|
}
|
||||||
|
|
||||||
public List<Long> getRunningJobsByExecutionParameter(String jobName, String parameterKeyName, String parameterValue) {
|
public List<Long> getRunningJobsByExecutionParameter(String jobName, String parameterKeyName, String parameterValue) {
|
||||||
return namedParameterJdbcTemplate.queryForList("SELECT bje.JOB_EXECUTION_ID FROM BATCH_JOB_INSTANCE bji "
|
return namedParameterJdbcTemplate.queryForList("""
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
|
SELECT bje.JOB_EXECUTION_ID
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION_PARAMS bjep ON bje.JOB_EXECUTION_ID = bjep.JOB_EXECUTION_ID "
|
FROM BATCH_JOB_INSTANCE bji
|
||||||
+ "WHERE bje.STATUS IN (:statuses) AND bji.JOB_NAME = :jobName AND bjep.KEY_NAME = :parameterKeyName AND bjep.STRING_VAL = :parameterValue AND bje.JOB_INSTANCE_ID NOT IN ("
|
INNER JOIN BATCH_JOB_EXECUTION bje
|
||||||
+ "SELECT bje.JOB_INSTANCE_ID FROM BATCH_JOB_INSTANCE bji "
|
ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID
|
||||||
+ "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
|
INNER JOIN BATCH_JOB_EXECUTION_PARAMS bjep
|
||||||
+ "WHERE bje.STATUS = :completedStatus AND bji.JOB_NAME = :jobName)",
|
ON bje.JOB_EXECUTION_ID = bjep.JOB_EXECUTION_ID
|
||||||
Map.of("statuses", List.of(STARTED.name(), STARTING.name()), "jobName", jobName, "completedStatus", COMPLETED.name(),
|
WHERE
|
||||||
"parameterKeyName", parameterKeyName, "parameterValue", parameterValue),
|
bje.STATUS IN (:statuses)
|
||||||
Long.class);
|
AND
|
||||||
|
bji.JOB_NAME = :jobName
|
||||||
|
AND
|
||||||
|
bjep.KEY_NAME = :parameterKeyName
|
||||||
|
AND
|
||||||
|
bjep.STRING_VAL = :parameterValue
|
||||||
|
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
|
||||||
|
WHERE
|
||||||
|
bje.STATUS = :completedStatus
|
||||||
|
AND
|
||||||
|
bji.JOB_NAME = :jobName
|
||||||
|
)
|
||||||
|
""", Map.of("statuses", List.of(STARTED.name(), STARTING.name()), "jobName", jobName, "completedStatus", COMPLETED.name(),
|
||||||
|
"parameterKeyName", parameterKeyName, "parameterValue", parameterValue), Long.class);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -20,5 +20,5 @@ package org.apache.fineract.infrastructure.jobs.service;
|
|||||||
|
|
||||||
public interface StuckJobExecutorService {
|
public interface StuckJobExecutorService {
|
||||||
|
|
||||||
void executeStuckJob(String jobName, Long jobId);
|
void resumeStuckJob(String jobName);
|
||||||
}
|
}
|
||||||
|
|||||||
+22
-10
@@ -25,7 +25,6 @@ import lombok.RequiredArgsConstructor;
|
|||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.apache.fineract.infrastructure.jobs.data.partitionedjobs.PartitionedJob;
|
import org.apache.fineract.infrastructure.jobs.data.partitionedjobs.PartitionedJob;
|
||||||
import org.apache.fineract.infrastructure.jobs.domain.JobExecutionRepository;
|
import org.apache.fineract.infrastructure.jobs.domain.JobExecutionRepository;
|
||||||
import org.jetbrains.annotations.NotNull;
|
|
||||||
import org.springframework.batch.core.launch.JobOperator;
|
import org.springframework.batch.core.launch.JobOperator;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
import org.springframework.transaction.TransactionStatus;
|
import org.springframework.transaction.TransactionStatus;
|
||||||
@@ -40,20 +39,33 @@ public class StuckJobExecutorServiceImpl implements StuckJobExecutorService {
|
|||||||
private final JobExecutionRepository jobExecutionRepository;
|
private final JobExecutionRepository jobExecutionRepository;
|
||||||
private final TransactionTemplate transactionTemplate;
|
private final TransactionTemplate transactionTemplate;
|
||||||
private final JobOperator jobOperator;
|
private final JobOperator jobOperator;
|
||||||
private final JobRegisterService jobRegisterService;
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void executeStuckJob(String jobName, Long jobId) {
|
public void resumeStuckJob(String jobName) {
|
||||||
|
List<Long> stuckJobIds = getStuckJobIds(jobName);
|
||||||
if (isPartitionedJob(jobName) && areThereStuckJobs(jobName)) {
|
if (isPartitionedJob(jobName) && areThereStuckJobs(jobName)) {
|
||||||
// Restarting stuck Partitioned jobs
|
restartPartitionedJobs(jobName, stuckJobIds);
|
||||||
List<Long> stuckJobIds = getStuckJobIds(jobName);
|
|
||||||
stuckJobIds.forEach(stuckJobId -> handleStuckJob(stuckJobId, getPartitionerStepName(jobName)));
|
|
||||||
} else {
|
} else {
|
||||||
// Executing stuck Tasklet jobs
|
restartTaskletJobs(stuckJobIds);
|
||||||
jobRegisterService.executeJobWithParameters(jobId, null);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void restartTaskletJobs(List<Long> stuckJobIds) {
|
||||||
|
stuckJobIds.forEach(this::handleStuckTaskletJob);
|
||||||
|
}
|
||||||
|
|
||||||
|
private void handleStuckTaskletJob(Long stuckJobId) {
|
||||||
|
try {
|
||||||
|
jobOperator.restart(stuckJobId);
|
||||||
|
} catch (Exception e) {
|
||||||
|
throw new RuntimeException("Exception while handling a stuck job", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void restartPartitionedJobs(String jobName, List<Long> stuckJobIds) {
|
||||||
|
stuckJobIds.forEach(stuckJobId -> handleStuckPartitionedJob(stuckJobId, getPartitionerStepName(jobName)));
|
||||||
|
}
|
||||||
|
|
||||||
private boolean isPartitionedJob(String jobName) {
|
private boolean isPartitionedJob(String jobName) {
|
||||||
return PartitionedJob.existsByJobName(jobName);
|
return PartitionedJob.existsByJobName(jobName);
|
||||||
}
|
}
|
||||||
@@ -71,14 +83,14 @@ public class StuckJobExecutorServiceImpl implements StuckJobExecutorService {
|
|||||||
return jobExecutionRepository.getStuckJobIdsByJobName(jobName);
|
return jobExecutionRepository.getStuckJobIdsByJobName(jobName);
|
||||||
}
|
}
|
||||||
|
|
||||||
private void handleStuckJob(Long stuckJobId, String partitionerStepName) {
|
private void handleStuckPartitionedJob(Long stuckJobId, String partitionerStepName) {
|
||||||
try {
|
try {
|
||||||
waitUntilAllPartitionsFinished(stuckJobId, partitionerStepName);
|
waitUntilAllPartitionsFinished(stuckJobId, partitionerStepName);
|
||||||
transactionTemplate.setPropagationBehavior(PROPAGATION_REQUIRES_NEW);
|
transactionTemplate.setPropagationBehavior(PROPAGATION_REQUIRES_NEW);
|
||||||
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
|
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected void doInTransactionWithoutResult(@NotNull TransactionStatus status) {
|
protected void doInTransactionWithoutResult(TransactionStatus status) {
|
||||||
jobExecutionRepository.updateJobStatusToFailed(stuckJobId, partitionerStepName);
|
jobExecutionRepository.updateJobStatusToFailed(stuckJobId, partitionerStepName);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|||||||
+2
-14
@@ -21,8 +21,6 @@ package org.apache.fineract.infrastructure.jobs.service;
|
|||||||
import java.time.LocalDate;
|
import java.time.LocalDate;
|
||||||
import java.util.HashMap;
|
import java.util.HashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
|
||||||
import java.util.Optional;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
|
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
|
||||||
import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService;
|
import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService;
|
||||||
@@ -31,7 +29,6 @@ import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
|
|||||||
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
|
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
|
||||||
import org.apache.fineract.infrastructure.event.external.service.JdbcTemplateFactory;
|
import org.apache.fineract.infrastructure.event.external.service.JdbcTemplateFactory;
|
||||||
import org.apache.fineract.infrastructure.jobs.domain.JobExecutionRepository;
|
import org.apache.fineract.infrastructure.jobs.domain.JobExecutionRepository;
|
||||||
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameData;
|
|
||||||
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameProvider;
|
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameProvider;
|
||||||
import org.apache.fineract.infrastructure.security.service.TenantDetailsService;
|
import org.apache.fineract.infrastructure.security.service.TenantDetailsService;
|
||||||
import org.springframework.batch.core.configuration.JobRegistry;
|
import org.springframework.batch.core.configuration.JobRegistry;
|
||||||
@@ -67,18 +64,9 @@ public class StuckJobListener implements ApplicationListener<ContextRefreshedEve
|
|||||||
HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates();
|
HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates();
|
||||||
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
|
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
|
||||||
ThreadLocalContextUtil.setBusinessDates(businessDates);
|
ThreadLocalContextUtil.setBusinessDates(businessDates);
|
||||||
stuckJobNames.forEach(stuckJobName -> {
|
stuckJobNames.forEach(stuckJobExecutorService::resumeStuckJob);
|
||||||
String sql = "select id from job where name = :jobName";
|
|
||||||
Optional<JobNameData> jobNameDataOptional = jobNameProvider.provide().stream()
|
|
||||||
.filter(jobNameData -> stuckJobName.equals(jobNameData.getEnumStyleName())).findFirst();
|
|
||||||
JobNameData jobNameData = jobNameDataOptional
|
|
||||||
.orElseThrow(() -> new IllegalArgumentException("Job not found by name: " + stuckJobName));
|
|
||||||
Long stuckJobId = namedParameterJdbcTemplate.queryForObject(sql,
|
|
||||||
Map.of("jobName", jobNameData.getHumanReadableName()), Long.class);
|
|
||||||
stuckJobExecutorService.executeStuckJob(stuckJobName, stuckJobId);
|
|
||||||
});
|
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
throw new RuntimeException(e);
|
throw new RuntimeException("Error while trying to restart stuck jobs", e);
|
||||||
} finally {
|
} finally {
|
||||||
ThreadLocalContextUtil.reset();
|
ThreadLocalContextUtil.reset();
|
||||||
}
|
}
|
||||||
|
|||||||
+7
-9
@@ -22,6 +22,7 @@ import java.time.LocalDate;
|
|||||||
import java.util.Collection;
|
import java.util.Collection;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.apache.commons.collections4.CollectionUtils;
|
||||||
import org.apache.fineract.accounting.provisioning.exception.ProvisioningEntryAlreadyCreatedException;
|
import org.apache.fineract.accounting.provisioning.exception.ProvisioningEntryAlreadyCreatedException;
|
||||||
import org.apache.fineract.accounting.provisioning.service.ProvisioningEntriesWritePlatformService;
|
import org.apache.fineract.accounting.provisioning.service.ProvisioningEntriesWritePlatformService;
|
||||||
import org.apache.fineract.infrastructure.core.service.DateUtils;
|
import org.apache.fineract.infrastructure.core.service.DateUtils;
|
||||||
@@ -31,8 +32,6 @@ import org.springframework.batch.core.StepContribution;
|
|||||||
import org.springframework.batch.core.scope.context.ChunkContext;
|
import org.springframework.batch.core.scope.context.ChunkContext;
|
||||||
import org.springframework.batch.core.step.tasklet.Tasklet;
|
import org.springframework.batch.core.step.tasklet.Tasklet;
|
||||||
import org.springframework.batch.repeat.RepeatStatus;
|
import org.springframework.batch.repeat.RepeatStatus;
|
||||||
import org.springframework.dao.DataIntegrityViolationException;
|
|
||||||
import org.springframework.orm.jpa.JpaSystemException;
|
|
||||||
|
|
||||||
@Slf4j
|
@Slf4j
|
||||||
@RequiredArgsConstructor
|
@RequiredArgsConstructor
|
||||||
@@ -48,14 +47,13 @@ public class GenerateLoanlossProvisioningTasklet implements Tasklet {
|
|||||||
try {
|
try {
|
||||||
Collection<ProvisioningCriteriaData> criteriaCollection = provisioningCriteriaReadPlatformService
|
Collection<ProvisioningCriteriaData> criteriaCollection = provisioningCriteriaReadPlatformService
|
||||||
.retrieveAllProvisioningCriterias();
|
.retrieveAllProvisioningCriterias();
|
||||||
if (criteriaCollection == null || criteriaCollection.size() == 0) {
|
if (CollectionUtils.isNotEmpty(criteriaCollection)) {
|
||||||
throw new NoProvisioningCriteriaDefinitionFoundException();
|
provisioningEntriesWritePlatformService.createProvisioningEntry(currentDate, addJournalEntries);
|
||||||
}
|
}
|
||||||
provisioningEntriesWritePlatformService.createProvisioningEntry(currentDate, addJournalEntries);
|
} catch (ProvisioningEntryAlreadyCreatedException e) {
|
||||||
} catch (ProvisioningEntryAlreadyCreatedException peace) {
|
log.error("Provisioning entry already created", e);
|
||||||
log.error("Provisioning Entry already created", peace);
|
} catch (Exception e) {
|
||||||
} catch (final JpaSystemException | DataIntegrityViolationException dve) {
|
log.error("Problem occurred when generating provisioning entries", e);
|
||||||
log.error("Problem occurred in generateLoanLossProvisioningAmount function", dve);
|
|
||||||
}
|
}
|
||||||
return RepeatStatus.FINISHED;
|
return RepeatStatus.FINISHED;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -43,6 +43,8 @@ fineract.mode.batch-manager-enabled=${FINERACT_MODE_BATCH_MANAGER_ENABLED:true}
|
|||||||
fineract.correlation.enabled=${FINERACT_LOGGING_HTTP_CORRELATION_ID_ENABLED:false}
|
fineract.correlation.enabled=${FINERACT_LOGGING_HTTP_CORRELATION_ID_ENABLED:false}
|
||||||
fineract.correlation.header-name=${FINERACT_LOGGING_HTTP_CORRELATION_ID_HEADER_NAME:X-Correlation-ID}
|
fineract.correlation.header-name=${FINERACT_LOGGING_HTTP_CORRELATION_ID_HEADER_NAME:X-Correlation-ID}
|
||||||
|
|
||||||
|
fineract.job.stuck-retry-threshold=${FINERACT_JOB_STUCK_RETRY_THRESHOLD:5}
|
||||||
|
|
||||||
fineract.partitioned-job.partitioned-job-properties[0].job-name=LOAN_COB
|
fineract.partitioned-job.partitioned-job-properties[0].job-name=LOAN_COB
|
||||||
fineract.partitioned-job.partitioned-job-properties[0].chunk-size=${LOAN_COB_CHUNK_SIZE:100}
|
fineract.partitioned-job.partitioned-job-properties[0].chunk-size=${LOAN_COB_CHUNK_SIZE:100}
|
||||||
fineract.partitioned-job.partitioned-job-properties[0].partition-size=${LOAN_COB_PARTITION_SIZE:100}
|
fineract.partitioned-job.partitioned-job-properties[0].partition-size=${LOAN_COB_PARTITION_SIZE:100}
|
||||||
|
|||||||
+137
@@ -0,0 +1,137 @@
|
|||||||
|
/**
|
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one
|
||||||
|
* or more contributor license agreements. See the NOTICE file
|
||||||
|
* distributed with this work for additional information
|
||||||
|
* regarding copyright ownership. The ASF licenses this file
|
||||||
|
* to you under the Apache License, Version 2.0 (the
|
||||||
|
* "License"); you may not use this file except in compliance
|
||||||
|
* with the License. You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing,
|
||||||
|
* software distributed under the License is distributed on an
|
||||||
|
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||||
|
* KIND, either express or implied. See the License for the
|
||||||
|
* specific language governing permissions and limitations
|
||||||
|
* under the License.
|
||||||
|
*/
|
||||||
|
package org.apache.fineract.portfolio.loanaccount.jobs.generateloanlossprovisioning;
|
||||||
|
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
import static org.mockito.BDDMockito.given;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.verifyNoInteractions;
|
||||||
|
import static org.springframework.batch.repeat.RepeatStatus.FINISHED;
|
||||||
|
|
||||||
|
import java.time.LocalDate;
|
||||||
|
import java.util.Collection;
|
||||||
|
import java.util.HashMap;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
import org.apache.fineract.accounting.provisioning.exception.ProvisioningEntryAlreadyCreatedException;
|
||||||
|
import org.apache.fineract.accounting.provisioning.service.ProvisioningEntriesWritePlatformService;
|
||||||
|
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
|
||||||
|
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
|
||||||
|
import org.apache.fineract.organisation.provisioning.data.ProvisioningCriteriaData;
|
||||||
|
import org.apache.fineract.organisation.provisioning.service.ProvisioningCriteriaReadPlatformService;
|
||||||
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.extension.ExtendWith;
|
||||||
|
import org.mockito.InjectMocks;
|
||||||
|
import org.mockito.Mock;
|
||||||
|
import org.mockito.junit.jupiter.MockitoExtension;
|
||||||
|
import org.mockito.junit.jupiter.MockitoSettings;
|
||||||
|
import org.mockito.quality.Strictness;
|
||||||
|
import org.springframework.batch.core.StepContribution;
|
||||||
|
import org.springframework.batch.core.scope.context.ChunkContext;
|
||||||
|
import org.springframework.batch.repeat.RepeatStatus;
|
||||||
|
|
||||||
|
@ExtendWith(MockitoExtension.class)
|
||||||
|
@MockitoSettings(strictness = Strictness.LENIENT)
|
||||||
|
class GenerateLoanlossProvisioningTaskletTest {
|
||||||
|
|
||||||
|
public static final LocalDate BUSINESS_DATE = LocalDate.of(2022, 6, 12);
|
||||||
|
|
||||||
|
@Mock
|
||||||
|
private StepContribution stepContribution;
|
||||||
|
|
||||||
|
@Mock
|
||||||
|
private ChunkContext chunkContext;
|
||||||
|
|
||||||
|
@Mock
|
||||||
|
private ProvisioningCriteriaReadPlatformService provisioningCriteriaReadPlatformService;
|
||||||
|
@Mock
|
||||||
|
private ProvisioningEntriesWritePlatformService provisioningEntriesWritePlatformService;
|
||||||
|
|
||||||
|
@InjectMocks
|
||||||
|
private GenerateLoanlossProvisioningTasklet underTest;
|
||||||
|
|
||||||
|
@BeforeEach
|
||||||
|
public void setUp() {
|
||||||
|
ThreadLocalContextUtil.setBusinessDates(new HashMap<>(Map.of(BusinessDateType.BUSINESS_DATE, BUSINESS_DATE)));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testExecuteShouldCreateProvisioningEntry() throws Exception {
|
||||||
|
// given
|
||||||
|
Collection<ProvisioningCriteriaData> provisioningCriterias = List.of(new ProvisioningCriteriaData());
|
||||||
|
given(provisioningCriteriaReadPlatformService.retrieveAllProvisioningCriterias()).willReturn(provisioningCriterias);
|
||||||
|
// when
|
||||||
|
RepeatStatus result = underTest.execute(stepContribution, chunkContext);
|
||||||
|
// then
|
||||||
|
verify(provisioningEntriesWritePlatformService).createProvisioningEntry(BUSINESS_DATE, true);
|
||||||
|
assertThat(result).isEqualTo(FINISHED);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testExecuteShouldNotCreateProvisioningEntryWhenNoProvisioningCriteriasArePresent() throws Exception {
|
||||||
|
// given
|
||||||
|
Collection<ProvisioningCriteriaData> provisioningCriterias = List.of();
|
||||||
|
given(provisioningCriteriaReadPlatformService.retrieveAllProvisioningCriterias()).willReturn(provisioningCriterias);
|
||||||
|
// when
|
||||||
|
RepeatStatus result = underTest.execute(stepContribution, chunkContext);
|
||||||
|
// then
|
||||||
|
verifyNoInteractions(provisioningEntriesWritePlatformService);
|
||||||
|
assertThat(result).isEqualTo(FINISHED);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testExecuteShouldNotCreateProvisioningEntryWhenNullProvisioningCriteriasArePresent() throws Exception {
|
||||||
|
// given
|
||||||
|
given(provisioningCriteriaReadPlatformService.retrieveAllProvisioningCriterias()).willReturn(null);
|
||||||
|
// when
|
||||||
|
RepeatStatus result = underTest.execute(stepContribution, chunkContext);
|
||||||
|
// then
|
||||||
|
verifyNoInteractions(provisioningEntriesWritePlatformService);
|
||||||
|
assertThat(result).isEqualTo(FINISHED);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testExecuteShouldNotFailWhenProvisioningEntryIsAlreadyCreated() throws Exception {
|
||||||
|
// given
|
||||||
|
Collection<ProvisioningCriteriaData> provisioningCriterias = List.of(new ProvisioningCriteriaData());
|
||||||
|
given(provisioningCriteriaReadPlatformService.retrieveAllProvisioningCriterias()).willReturn(provisioningCriterias);
|
||||||
|
given(provisioningEntriesWritePlatformService.createProvisioningEntry(BUSINESS_DATE, true))
|
||||||
|
.willThrow(new ProvisioningEntryAlreadyCreatedException(1L, BUSINESS_DATE));
|
||||||
|
// when
|
||||||
|
RepeatStatus result = underTest.execute(stepContribution, chunkContext);
|
||||||
|
// then
|
||||||
|
verify(provisioningEntriesWritePlatformService).createProvisioningEntry(BUSINESS_DATE, true);
|
||||||
|
assertThat(result).isEqualTo(FINISHED);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testExecuteShouldNotFailWhenExceptionIsThrownInProvisioningEntryCreation() throws Exception {
|
||||||
|
// given
|
||||||
|
Collection<ProvisioningCriteriaData> provisioningCriterias = List.of(new ProvisioningCriteriaData());
|
||||||
|
given(provisioningCriteriaReadPlatformService.retrieveAllProvisioningCriterias()).willReturn(provisioningCriterias);
|
||||||
|
given(provisioningEntriesWritePlatformService.createProvisioningEntry(BUSINESS_DATE, true)).willThrow(new RuntimeException("Test"));
|
||||||
|
// when
|
||||||
|
RepeatStatus result = underTest.execute(stepContribution, chunkContext);
|
||||||
|
// then
|
||||||
|
verify(provisioningEntriesWritePlatformService).createProvisioningEntry(BUSINESS_DATE, true);
|
||||||
|
assertThat(result).isEqualTo(FINISHED);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user