FINERACT-2386: Journal entry aggregation capabilities

This commit is contained in:
Adam Saghy
2025-10-03 11:07:12 +02:00
parent 297fa8e707
commit 89e9cfcd7d
35 changed files with 2531 additions and 3 deletions
@@ -417,6 +417,16 @@ public class FineractProperties {
private int stuckRetryThreshold;
private boolean loanCobEnabled;
private FineractJournalEntryAggregationProperties journalEntryAggregation;
}
@Getter
@Setter
public static class FineractJournalEntryAggregationProperties {
private Integer excludeRecentNDays;
private boolean enabled;
private Integer chunkSize;
}
@Getter
@@ -58,7 +58,9 @@ public enum JobName {
PURGE_EXTERNAL_EVENTS("Purge External Events"), //
PURGE_PROCESSED_COMMANDS("Purge Processed Commands"), //
ACCRUAL_ACTIVITY_POSTING("Accrual Activity Posting"), //
ADD_PERIODIC_ACCRUAL_ENTRIES_FOR_SAVINGS_WITH_INCOME_POSTED_AS_TRANSACTIONS("Add Accrual Transactions For Savings"); //
ADD_PERIODIC_ACCRUAL_ENTRIES_FOR_SAVINGS_WITH_INCOME_POSTED_AS_TRANSACTIONS("Add Accrual Transactions For Savings"), //
JOURNAL_ENTRY_AGGREGATION("Journal Entry Aggregation"), //
; //
private final String name;
@@ -8,4 +8,5 @@ include::approved-amount-modification.adoc[leveloffset=+1]
include::backdated-interest-modification.adoc[leveloffset=+1]
include::interest-rate-change-progressive-loans.adoc[leveloffset=+1]
include::contract-termination.adoc[leveloffset=+1]
include::loan-charges.adoc[leveloffset=+1]
include::loan-charges.adoc[leveloffset=+1]
include::journal-entry-aggregation.adoc[leveloffset=+1]
@@ -0,0 +1,167 @@
= Journal Entry Aggregation
:experimental:
:source-highlighter: highlightjs
:toc: left
:toclevels: 3
:icons: font
:sectlinks:
:sectnums:
== Overview
The Journal Entry Aggregation Job is a Spring Batch-based solution designed to efficiently aggregate journal entries in the Fineract system. This job processes journal entries in configurable chunks, improving performance and resource utilization when dealing with large volumes of financial transactions.
== Key Features
=== Chunk-based Processing
* Processes journal entries in configurable batch sizes
* Reduces memory footprint by working with manageable data subsets
* Improves performance through efficient batch processing
=== Tracking and Deduplication
* Tracks processed date ranges to prevent duplicate aggregations
* Uses `JournalEntryAggregationTracking` to maintain execution history
* Skips already processed date ranges in subsequent runs
=== Configurable Exclude Recent N Days
* Excludes the last N days (from business date) from processing
* Default `Exclude Recent N Days` can be customized via application properties
== How It Works
=== Job Flow
==== Job Initialization
* Determines the date range to process based on last execution
* Sets up execution context with date boundaries
==== Data Reading
* Fetches unaggregated journal entries within the target date range
* Groups entries by GL account, product, office, and other dimensions
==== Processing
* Aggregates debit and credit amounts for each group
* Handles external asset owner mappings
* Processes data in configurable chunk sizes
==== Tracking
* Records successful aggregation runs
* Maintains execution history for future reference
== Configuration
=== Job Parameters
* `aggregatedOnDate`: (Optional) Specific date to process (defaults to business date)
* `chunkSize`: (Optional) Number of records to process in each chunk
=== Application Properties
[source,properties]
----
# Exclude Recent N days from aggregation
fineract.job.journal-entry-aggregation.exclude-recent-N-days=1
# Chunk size for batch processing
fineract.job.journal-entry-aggregation.chunk-size=1000
----
== Usage
=== Manual Execution
Trigger the job manually through the Fineract API:
[source,http]
----
POST /jobs/short-name/JRNL_AGG
Content-Type: application/json
{
}
----
=== Scheduled Execution
Configure the job to run on a schedule by adding to your scheduler configuration.
=== Monitoring
Monitor job execution through:
* Job execution logs
* `JOURNAL_ENTRY_AGGREGATION_TRACKING` table
* Spring Batch job execution tables
== Best Practices
=== Chunk Size Tuning
* Larger chunks improve throughput but increase memory usage
* Monitor memory usage and adjust chunk size accordingly
=== Scheduling
* Schedule during off-peak hours for large datasets
* Consider running more frequently with smaller `Exclude Recent N Days` values
=== Error Handling
* Failed jobs can be restarted from the last successful chunk
* Review job execution logs for any processing issues
== Performance Considerations
* *Indexing*: Ensure proper indexes exist on `aggregated_on_date`, `office_id`, and other filtering columns
* *Partitioning*: Consider partitioning large journal entry tables by date for better performance
* *Batch Window*: Allocate sufficient time for the job to complete during maintenance windows
== Database Schema
=== m_journal_entry_aggregation_summary Table
This table stores the aggregated journal entry amounts, grouped by various dimensions for efficient reporting and analysis.
[cols="1,2,2,2", options="header"]
|===
| Column | Type | Nullable | Description
| id | BIGINT | No | Primary key
| gl_account_id | BIGINT | No | Reference to `acc_gl_account`
| product_id | BIGINT | Yes | Reference to the product (if applicable)
| office_id | BIGINT | No | Reference to `m_office`
| entity_type_enum | SMALLINT | No | Type of entity (e.g., loan, savings)
| submitted_on_date | DATE | No | The date of the business date when entry was submitted
| aggregated_on_date | DATE | No | The date when aggregation was performed
| debit_amount | DECIMAL(19,6) | No | Sum of debit amounts
| credit_amount | DECIMAL(19,6) | No | Sum of credit amounts
| external_owner_id | BIGINT | Yes | Reference to external owner (if applicable)
| job_execution_id | BIGINT | No | Reference to batch job execution
| created_date | TIMESTAMP | No | Record creation timestamp
| last_modified_date | TIMESTAMP | Yes | Last modification timestamp
|===
The table is designed to support efficient querying of aggregated financial data by:
* Date ranges (using `submitted_on_date` and `aggregated_on_date`)
* Organizational structure (using `office_id`)
* Financial dimensions (using `gl_account_id` and `product_id`)
* Entity types (using `entity_type_enum`)
=== m_journal_entry_aggregation_tracking Table
This table maintains a history of aggregation job executions, tracking which date ranges have been processed to prevent duplicate aggregations.
[cols="1,2,2,2", options="header"]
|===
| Column | Type | Nullable | Description
| id | BIGINT | No | Primary key
| job_execution_id | BIGINT | No | Reference to Spring Batch job execution
| aggregated_on_date_from | DATE | No | Start date of the aggregation period
| aggregated_on_date_to | DATE | No | End date of the aggregation period
| submitted_on_date | DATE | No | The date of the business date when entry was submitted
| status | VARCHAR(20) | No | Status of the aggregation (e.g., COMPLETED, FAILED)
| error_message | TEXT | Yes | Error details if the job failed
| start_time | TIMESTAMP | No | When the aggregation started
| end_time | TIMESTAMP | Yes | When the aggregation completed
| records_processed | INT | Yes | Number of records processed
| created_date | TIMESTAMP | No | Record creation timestamp
| last_modified_date | TIMESTAMP | Yes | Last modification timestamp
|===
Key aspects of the tracking table:
* Tracks the exact date ranges processed in each job execution
* Maintains job status and error information for debugging
* Records performance metrics (processing time, record counts)
* Used by the job to determine which date ranges need processing in subsequent runs
Indexes are created on frequently queried columns to ensure optimal performance for reporting and analysis.
This aggregation job provides a robust, scalable solution for processing journal entries while maintaining data integrity and providing clear audit trails of all aggregation activities.
@@ -42,4 +42,5 @@
<include relativeToChangelogFile="true" file="parts/0018_add_external_asset_owner_transfer_outstanding_interest_strategy.xml"/>
<include relativeToChangelogFile="true" file="parts/0019_add_configurable_allowed_loan_statuses.xml"/>
<include relativeToChangelogFile="true" file="parts/0020_add_previous_owner_reference.xml"/>
<include relativeToChangelogFile="true" file="parts/0021_external_owner_reference_in_journal_entry_aggregation.xml"/>
</databaseChangeLog>
@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.1.xsd">
<changeSet author="fineract" id="1">
<preConditions onFail="MARK_RAN">
<tableExists tableName="m_journal_entry_aggregation_summary"/>
</preConditions>
<addForeignKeyConstraint baseColumnNames="external_owner_id" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_EXTERNAL_OWNER_ID" referencedColumnNames="id" referencedTableName="m_external_asset_owner"/>
</changeSet>
</databaseChangeLog>
@@ -266,7 +266,7 @@ public class JobRegisterServiceImpl implements JobRegisterService, ApplicationLi
scheduledJobDetails.setNextRunTime(null);
final String stackTrace = getStackTraceAsString(throwable);
scheduledJobDetails.setErrorLog(stackTrace);
log.error("Could not schedule job: {}", scheduledJobDetails.getJobName(), throwable);
log.warn("Could not schedule job: {}", scheduledJobDetails.getJobName(), throwable);
}
scheduledJobDetails.setCurrentlyRunning(false);
}
@@ -0,0 +1,91 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
import static org.apache.fineract.infrastructure.jobs.service.aggregationjob.JournalEntryAggregationJobConstant.JOB_SUMMARY_STEP_NAME;
import static org.apache.fineract.infrastructure.jobs.service.aggregationjob.JournalEntryAggregationJobConstant.JOB_TRACKING_STEP_NAME;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.apache.fineract.infrastructure.core.service.migration.TenantDataSourceFactory;
import org.apache.fineract.infrastructure.jobs.service.JobName;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationSummaryData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.listener.JournalEntryAggregationJobListener;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.tasklet.JournalEntryAggregationTrackingTasklet;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;
@Configuration
@ConditionalOnProperty(value = "fineract.job.journal-entry-aggregation.enabled", havingValue = "true")
public class JournalEntryAggregationJobConfiguration {
@Autowired
private JournalEntryAggregationJobListener journalEntryAggregationJobListener;
@Autowired
private JobRepository jobRepository;
@Autowired
private JournalEntryAggregationJobExecutionDecider journalEntryAggregationJobExecutionDecider;
@Autowired
private JournalEntryAggregationJobWriter aggregationItemWriter;
@Autowired
private PlatformTransactionManager transactionManager;
@Autowired
private FineractProperties fineractProperties;
@Autowired
private JournalEntryAggregationTrackingTasklet journalEntryAggregationTrackingTasklet;
@Autowired
private TenantDataSourceFactory tenantDataSourceFactory;
@Bean
public Step journalEntryAggregationSummaryStep() {
return new StepBuilder(JOB_SUMMARY_STEP_NAME, jobRepository)
.<JournalEntryAggregationSummaryData, JournalEntryAggregationSummaryData>chunk(
fineractProperties.getJob().getJournalEntryAggregation().getChunkSize(), transactionManager)
.reader(journalEntryAggregationJobReader()).writer(aggregationItemWriter).allowStartIfComplete(true).build();
}
@Bean
public JournalEntryAggregationJobReader journalEntryAggregationJobReader() {
return new JournalEntryAggregationJobReader(tenantDataSourceFactory);
}
@Bean
protected Step journalEntryAggregationTrackingStep() {
return new StepBuilder(JOB_TRACKING_STEP_NAME, jobRepository).tasklet(journalEntryAggregationTrackingTasklet, transactionManager)
.build();
}
@Bean(name = "journalEntryAggregation")
public Job journalEntryAggregation() {
return new JobBuilder(JobName.JOURNAL_ENTRY_AGGREGATION.name(), jobRepository).listener(journalEntryAggregationJobListener)
.start(journalEntryAggregationJobExecutionDecider).on(JournalEntryAggregationJobConstant.NO_OP_EXECUTION).end()
.from(journalEntryAggregationJobExecutionDecider).on(JournalEntryAggregationJobConstant.CONTINUE_JOB_EXECUTION)
.to(journalEntryAggregationSummaryStep()).next(journalEntryAggregationTrackingStep()).end()
.incrementer(new RunIdIncrementer()).build();
}
}
@@ -0,0 +1,34 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
public final class JournalEntryAggregationJobConstant {
public static final String CONTINUE_JOB_EXECUTION = "CONTINUE_JOB_EXECUTION";
public static final String NO_OP_EXECUTION = "NO_OP_EXECUTION";
public static final String JOURNAL_ENTRY_AGGREGATION_JOB_NAME = "JOURNAL_ENTRY_AGGREGATION";
public static final String JOB_SUMMARY_STEP_NAME = "JournalEntryAggregation Summary Insert - Step";
public static final String JOB_TRACKING_STEP_NAME = "JournalEntryAggregation Tracking Insert - Step";
public static final String AGGREGATED_ON_DATE = "aggregatedOnDate";
public static final String AGGREGATED_ON_DATE_FROM = "aggregatedOnDateFrom";
public static final String AGGREGATED_ON_DATE_TO = "aggregatedOnDateTo";
public static final String LAST_AGGREGATED_ON_DATE = "lastAggregatedOnDate";
private JournalEntryAggregationJobConstant() {}
}
@@ -0,0 +1,64 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
import java.time.LocalDate;
import java.util.Objects;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class JournalEntryAggregationJobExecutionDecider implements JobExecutionDecider {
@Override
@NonNull
public FlowExecutionStatus decide(final @NonNull JobExecution jobExecution, final StepExecution stepExecution) {
final LocalDate aggregatedOnDate = (LocalDate) jobExecution.getExecutionContext()
.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE);
final LocalDate lastAggregatedOnDate = (LocalDate) jobExecution.getExecutionContext()
.get(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE);
// check if aggregation for given date range already exist.
if (aggregationAlreadyExist(lastAggregatedOnDate, aggregatedOnDate)) {
log.info(
"Journal entry aggregation for given aggregatedOnDate already exist hence skipping jobName={} execution for aggregatedOnDate={}",
JournalEntryAggregationJobConstant.JOURNAL_ENTRY_AGGREGATION_JOB_NAME, aggregatedOnDate);
jobExecution.setExitStatus(ExitStatus.NOOP);
return new FlowExecutionStatus(JournalEntryAggregationJobConstant.NO_OP_EXECUTION);
}
log.info("Continue executing journal entry aggregation job jobName={} execution for aggregatedOnDate={}",
JournalEntryAggregationJobConstant.JOURNAL_ENTRY_AGGREGATION_JOB_NAME, aggregatedOnDate);
return new FlowExecutionStatus(JournalEntryAggregationJobConstant.CONTINUE_JOB_EXECUTION);
}
private boolean aggregationAlreadyExist(final LocalDate lastAggregatedOnDate, final LocalDate aggregatedOnDate) {
if (Objects.isNull(lastAggregatedOnDate)) {
log.info("Journal Entry aggregation job being executed for the first time");
return false;
}
return aggregatedOnDate.isBefore(lastAggregatedOnDate) || aggregatedOnDate.isEqual(lastAggregatedOnDate);
}
}
@@ -0,0 +1,103 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.LocalDate;
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
import org.apache.fineract.infrastructure.core.domain.JdbcSupport;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.core.service.migration.TenantDataSourceFactory;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationSummaryData;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.stereotype.Component;
@Component
@StepScope
public class JournalEntryAggregationJobReader extends JdbcCursorItemReader<JournalEntryAggregationSummaryData> {
private LocalDate aggregatedOnDateFrom;
private LocalDate aggregatedOnDateTo;
public JournalEntryAggregationJobReader(TenantDataSourceFactory tenantDataSourceFactory) {
FineractPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
setDataSource(tenantDataSourceFactory.create(tenant));
setSql(buildAggregationQuery());
setRowMapper(this::mapRow);
}
@BeforeStep
public void beforeStep(StepExecution stepExecution) {
ExecutionContext ctx = stepExecution.getJobExecution().getExecutionContext();
this.aggregatedOnDateFrom = (LocalDate) ctx.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_FROM);
this.aggregatedOnDateTo = (LocalDate) ctx.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_TO);
setPreparedStatementSetter(ps -> {
ps.setObject(1, aggregatedOnDateFrom);
ps.setObject(2, aggregatedOnDateTo);
});
}
private JournalEntryAggregationSummaryData mapRow(ResultSet rs, int rowNum) throws SQLException {
return JournalEntryAggregationSummaryData.builder() //
.glAccountId(rs.getLong("glAccountId")) //
.productId(rs.getLong("productId")) //
.office(rs.getLong("officeId")) //
.entityTypeEnum(rs.getLong("entityTypeEnum")) //
.currencyCode(rs.getString("currencyCode")) //
.submittedOnDate(ThreadLocalContextUtil.getBusinessDate()) //
.aggregatedOnDate(JdbcSupport.getLocalDate(rs, "aggregatedOnDate")) //
.externalOwnerId(JdbcSupport.getLong(rs, "externalOwner")) //
.debitAmount(rs.getBigDecimal("debitAmount")) //
.creditAmount(rs.getBigDecimal("creditAmount")) //
.manualEntry(false) //
.build(); //
}
private String buildAggregationQuery() {
return """
SELECT lp.id AS productId,
acc_gl_account.id AS glAccountId,
acc_gl_journal_entry.entity_type_enum AS entityTypeEnum,
acc_gl_journal_entry.office_id AS officeId,
aw.owner_id AS externalOwner,
SUM(CASE WHEN acc_gl_journal_entry.type_enum = 2 THEN amount ELSE 0 END) AS debitAmount,
SUM(CASE WHEN acc_gl_journal_entry.type_enum = 1 THEN amount ELSE 0 END) AS creditAmount,
acc_gl_journal_entry.submitted_on_date as aggregatedOnDate,
acc_gl_journal_entry.currency_code as currencyCode
FROM acc_gl_account
JOIN acc_gl_journal_entry ON acc_gl_account.id = acc_gl_journal_entry.account_id
JOIN m_loan m ON m.id = acc_gl_journal_entry.entity_id
JOIN m_product_loan lp ON lp.id = m.product_id
LEFT JOIN m_external_asset_owner_journal_entry_mapping aw
ON aw.journal_entry_id = acc_gl_journal_entry.id
WHERE acc_gl_journal_entry.entity_type_enum = 1
AND acc_gl_journal_entry.submitted_on_date > ?
AND acc_gl_journal_entry.submitted_on_date <= ?
GROUP BY productId, glAccountId, externalOwner, aggregatedOnDate, currencyCode, entityTypeEnum, officeId
""";
}
}
@@ -0,0 +1,56 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationSummaryData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.services.JournalEntryAggregationWriterService;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemWriter;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;
@Component
@Slf4j
@RequiredArgsConstructor
public class JournalEntryAggregationJobWriter implements ItemWriter<JournalEntryAggregationSummaryData>, StepExecutionListener {
private final JournalEntryAggregationWriterService journalEntryAggregationWriterService;
private StepExecution stepExecution;
@Override
public void beforeStep(@NonNull StepExecution stepExecution) {
this.stepExecution = stepExecution;
}
@Override
public void write(@NonNull Chunk<? extends JournalEntryAggregationSummaryData> journalEntrySummaries) {
final Long jobExecutionId = stepExecution.getJobExecution().getId();
List<JournalEntryAggregationSummaryData> summariesList = journalEntrySummaries.getItems().stream().map(item -> {
item.setJobExecutionId(jobExecutionId);
return (JournalEntryAggregationSummaryData) item;
}).toList();
journalEntryAggregationWriterService.insertJournalEntrySummaryBatch(summariesList);
}
}
@@ -0,0 +1,44 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.data;
import java.math.BigDecimal;
import java.time.LocalDate;
import lombok.Builder;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
@Builder(toBuilder = true)
public class JournalEntryAggregationSummaryData {
private Long productId;
private Long glAccountId;
private Long office;
private Long entityTypeEnum;
private LocalDate submittedOnDate;
private LocalDate aggregatedOnDate;
private Long externalOwnerId;
private BigDecimal debitAmount;
private BigDecimal creditAmount;
private Boolean manualEntry;
private String currencyCode;
private Long jobExecutionId;
}
@@ -0,0 +1,35 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.data;
import java.time.LocalDate;
import lombok.Builder;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
@Builder(toBuilder = true)
public class JournalEntryAggregationTrackingData {
private LocalDate aggregatedOnDateFrom;
private LocalDate aggregatedOnDateTo;
private LocalDate submittedOnDate;
private Long jobExecutionId;
}
@@ -0,0 +1,47 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.domain;
import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.Table;
import java.time.LocalDate;
import lombok.Getter;
import lombok.Setter;
import org.apache.fineract.infrastructure.core.domain.AbstractAuditableWithUTCDateTimeCustom;
@Entity
@Table(name = "m_journal_entry_aggregation_tracking")
@Getter
@Setter
public class JournalEntryAggregationTracking extends AbstractAuditableWithUTCDateTimeCustom<Long> {
@Column(name = "aggregated_on_date_from", nullable = false)
private LocalDate aggregatedOnDateFrom;
@Column(name = "aggregated_on_date_to", nullable = false)
private LocalDate aggregatedOnDateTo;
@Column(name = "submitted_on_date", nullable = false)
private LocalDate submittedOnDate;
@Column(name = "job_execution_id", nullable = false)
private Long jobExecutionId;
}
@@ -0,0 +1,35 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.domain;
import java.time.LocalDate;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
public interface JournalEntryAggregationTrackingRepository
extends JpaRepository<JournalEntryAggregationTracking, Long>, JpaSpecificationExecutor<JournalEntryAggregationTracking> {
@Query("select max(jeat.aggregatedOnDateTo) from JournalEntryAggregationTracking jeat")
LocalDate findLatestAggregatedOnDate();
@Modifying
void deleteByJobExecutionId(Long jobExecutionId);
}
@@ -0,0 +1,69 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.domain;
import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.Table;
import java.math.BigDecimal;
import java.time.LocalDate;
import lombok.Getter;
import lombok.Setter;
import org.apache.fineract.infrastructure.core.domain.AbstractAuditableWithUTCDateTimeCustom;
@Getter
@Setter
@Entity
@Table(name = "m_journal_entry_aggregation_summary")
public class JournalEntrySummary extends AbstractAuditableWithUTCDateTimeCustom<Long> {
@Column(name = "product_id", nullable = false)
private Long product;
@Column(name = "gl_account_id", nullable = false)
private Long glAccountId;
@Column(name = "office_id", nullable = false)
private Long office;
@Column(name = "entity_type_enum", nullable = false)
private Long entityTypeEnum;
@Column(name = "aggregated_on_date", nullable = false)
private LocalDate aggregatedOnDate;
@Column(name = "submitted_on_date", nullable = false)
private LocalDate submittedOnDate;
@Column(name = "external_owner_id", nullable = false)
private Long externalOwnerId;
@Column(name = "debit_amount")
private BigDecimal debitAmount;
@Column(name = "credit_amount")
private BigDecimal creditAmount;
@Column(name = "manual_entry", nullable = false)
private Boolean manualEntry = false;
@Column(name = "job_execution_id", nullable = false)
private Long jobExecutionId;
}
@@ -0,0 +1,28 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.domain;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Modifying;
public interface JournalEntrySummaryRepository extends JpaRepository<JournalEntrySummary, Long> {
@Modifying
void deleteByJobExecutionId(Long jobExecutionId);
}
@@ -0,0 +1,122 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.listener;
import static org.apache.fineract.infrastructure.jobs.service.aggregationjob.JournalEntryAggregationJobConstant.JOB_SUMMARY_STEP_NAME;
import static org.apache.fineract.infrastructure.jobs.service.aggregationjob.JournalEntryAggregationJobConstant.JOURNAL_ENTRY_AGGREGATION_JOB_NAME;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.JournalEntryAggregationJobConstant;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntryAggregationTrackingRepository;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.services.JournalEntryAggregationWriterService;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.StepExecution;
import org.springframework.stereotype.Component;
@Component
@Slf4j
@RequiredArgsConstructor
public class JournalEntryAggregationJobListener implements JobExecutionListener {
private static final List<String> successJobStatus = List.of(ExitStatus.COMPLETED.getExitCode(), ExitStatus.NOOP.getExitCode());
private final FineractProperties fineractProperties;
private final JournalEntryAggregationTrackingRepository journalEntryAggregationTrackingRepository;
private final JournalEntryAggregationWriterService journalEntryAggregationWriterService;
@Override
public void beforeJob(final JobExecution jobExecution) {
log.info("Journal Entry Aggregation Job Started jobName={}, jobExecutionId={}", JOURNAL_ENTRY_AGGREGATION_JOB_NAME,
jobExecution.getId());
LocalDate providedAggregatedOnDate = (LocalDate) jobExecution.getExecutionContext()
.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE);
// if aggregatedOnDate not provided in the parameter it will be defaulted to businessDate.
final LocalDate aggregatedOnDate = providedAggregatedOnDate != null //
? providedAggregatedOnDate.minusDays(fineractProperties.getJob().getJournalEntryAggregation().getExcludeRecentNDays()) //
: ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.BUSINESS_DATE)
.minusDays(fineractProperties.getJob().getJournalEntryAggregation().getExcludeRecentNDays()); //
// get last or most recent aggregatedOnDate from tacking table.
final LocalDate lastAggregatedOnDate = journalEntryAggregationTrackingRepository.findLatestAggregatedOnDate();
initializeDates(jobExecution, aggregatedOnDate, lastAggregatedOnDate);
}
private void initializeDates(final JobExecution jobExecution, final LocalDate aggregatedOnDate, final LocalDate lastAggregatedOnDate) {
jobExecution.getExecutionContext().put(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE, aggregatedOnDate);
jobExecution.getExecutionContext().put(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_TO, aggregatedOnDate);
if (lastAggregatedOnDate == null) {
// Job is running for the first time
jobExecution.getExecutionContext().put(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE, null);
jobExecution.getExecutionContext().put(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_FROM, LocalDate.of(1970, 1, 1));
} else {
// Job is running for the subsequent time.
jobExecution.getExecutionContext().put(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE, lastAggregatedOnDate);
jobExecution.getExecutionContext().put(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_FROM, lastAggregatedOnDate);
}
}
@Override
public void afterJob(final JobExecution jobExecution) {
// if job not completed successfully, rollback any insertion made.
if (!successJobStatus.contains(jobExecution.getExitStatus().getExitCode())) {
journalEntryAggregationWriterService.rollbackJournalEntrySummary(jobExecution.getId());
journalEntryAggregationWriterService.rollbackJournalEntryTracking(jobExecution.getId());
}
// if job completed successfully, log the summary.
logJobExecutionSummary(jobExecution);
}
private void logJobExecutionSummary(final JobExecution jobExecution) {
final LocalDate aggregatedOnDateFrom = (LocalDate) jobExecution.getExecutionContext()
.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_FROM);
final LocalDate aggregatedOnDateTo = (LocalDate) jobExecution.getExecutionContext()
.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_TO);
final Long jobExecutionId = jobExecution.getId();
final Long recordProcessCount = jobExecution.getStepExecutions().stream()
.filter(stepExecution -> stepExecution.getStepName().equals(JOB_SUMMARY_STEP_NAME)).mapToLong(StepExecution::getWriteCount)
.sum();
final Instant startDateTime = jobExecution.getStartTime() != null ? jobExecution.getStartTime().toInstant(ZoneOffset.UTC) : null;
final Instant endDateTime = jobExecution.getEndTime() != null ? jobExecution.getEndTime().toInstant(ZoneOffset.UTC) : null;
long jobDuration = 0L;
Long startDateTimeMilliSecond = null;
Long endDateTimeMilliSecond = null;
if (startDateTime != null && endDateTime != null) {
startDateTimeMilliSecond = startDateTime.toEpochMilli();
endDateTimeMilliSecond = endDateTime.toEpochMilli();
jobDuration = startDateTime.until(endDateTime, ChronoUnit.MINUTES);
}
log.info(
"Execution Summary for jobName={}, aggregatedDateFrom={}, aggregatedDateTo={}, totalRecordProcessCount={}, startTime={}, endTime={}, startTime_ms={}, endTime_ms={}, "
+ "jobExecutionId={}, jobExecutionDurationInMinutes={}, tenantId={}",
JOURNAL_ENTRY_AGGREGATION_JOB_NAME, aggregatedOnDateFrom, aggregatedOnDateTo, recordProcessCount, startDateTime,
endDateTime, startDateTimeMilliSecond, endDateTimeMilliSecond, jobExecutionId, jobDuration,
ThreadLocalContextUtil.getTenant().getTenantIdentifier());
}
}
@@ -0,0 +1,34 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.services;
import java.util.List;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationSummaryData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationTrackingData;
public interface JournalEntryAggregationWriterService {
void insertJournalEntrySummaryBatch(List<JournalEntryAggregationSummaryData> journalEntrySummaries);
void insertJournalEntryTracking(JournalEntryAggregationTrackingData journalEntrySummaryTrackingDTO) throws Exception;
void rollbackJournalEntrySummary(Long jobExecutionId);
void rollbackJournalEntryTracking(Long jobExecutionId);
}
@@ -0,0 +1,90 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.services;
import java.util.List;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationSummaryData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationTrackingData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntryAggregationTracking;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntryAggregationTrackingRepository;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntrySummary;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntrySummaryRepository;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Component;
@Component
@AllArgsConstructor
@Slf4j
public class JournalEntryAggregationWriterServiceImpl implements JournalEntryAggregationWriterService {
private JournalEntrySummaryRepository journalSummaryRepository;
private JournalEntryAggregationTrackingRepository journalEntryAggregationTrackingRepository;
private NamedParameterJdbcTemplate namedParameterJdbcTemplate;
@Override
public void insertJournalEntrySummaryBatch(final List<JournalEntryAggregationSummaryData> journalEntrySummaries) {
List<JournalEntrySummary> entities = journalEntrySummaries.stream().map(this::convertToJournalEntrySummary).toList();
journalSummaryRepository.saveAll(entities);
}
@Override
public void insertJournalEntryTracking(JournalEntryAggregationTrackingData journalEntrySummaryTrackingDTO) throws Exception {
log.info("Inserting journal entry tracking");
saveJournalEntryTracking(journalEntrySummaryTrackingDTO);
}
@Override
public void rollbackJournalEntrySummary(final Long jobExecutionId) {
log.info("Rolling back journal entry summary for jobExecutionId={}", jobExecutionId);
journalSummaryRepository.deleteByJobExecutionId(jobExecutionId);
}
@Override
public void rollbackJournalEntryTracking(final Long jobExecutionId) {
log.info("Rolling back journal entry summary tracking for jobExecutionId={}", jobExecutionId);
journalEntryAggregationTrackingRepository.deleteByJobExecutionId(jobExecutionId);
}
private void saveJournalEntryTracking(final JournalEntryAggregationTrackingData journalEntrySummaryTrackingDTO) {
JournalEntryAggregationTracking journalEntryAggregationTracking = new JournalEntryAggregationTracking();
journalEntryAggregationTracking.setSubmittedOnDate(journalEntrySummaryTrackingDTO.getSubmittedOnDate());
journalEntryAggregationTracking.setAggregatedOnDateFrom(journalEntrySummaryTrackingDTO.getAggregatedOnDateFrom());
journalEntryAggregationTracking.setAggregatedOnDateTo(journalEntrySummaryTrackingDTO.getAggregatedOnDateTo());
journalEntryAggregationTracking.setJobExecutionId(journalEntrySummaryTrackingDTO.getJobExecutionId());
journalEntryAggregationTrackingRepository.save(journalEntryAggregationTracking);
}
private JournalEntrySummary convertToJournalEntrySummary(final JournalEntryAggregationSummaryData summaryDTO) {
JournalEntrySummary entrySummary = new JournalEntrySummary();
entrySummary.setProduct(summaryDTO.getProductId());
entrySummary.setGlAccountId(summaryDTO.getGlAccountId());
entrySummary.setOffice(summaryDTO.getOffice());
entrySummary.setEntityTypeEnum(summaryDTO.getEntityTypeEnum());
entrySummary.setSubmittedOnDate(summaryDTO.getSubmittedOnDate());
entrySummary.setDebitAmount(summaryDTO.getDebitAmount());
entrySummary.setCreditAmount(summaryDTO.getCreditAmount());
entrySummary.setExternalOwnerId(summaryDTO.getExternalOwnerId());
entrySummary.setAggregatedOnDate(summaryDTO.getAggregatedOnDate());
entrySummary.setJobExecutionId(summaryDTO.getJobExecutionId());
return entrySummary;
}
}
@@ -0,0 +1,68 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.tasklet;
import java.time.LocalDate;
import java.util.Optional;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.JournalEntryAggregationJobConstant;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationTrackingData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.services.JournalEntryAggregationWriterService;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;
@RequiredArgsConstructor
@Component
@Slf4j
public class JournalEntryAggregationTrackingTasklet implements Tasklet {
private final JournalEntryAggregationWriterService journalEntryAggregationWriterService;
@Override
public RepeatStatus execute(StepContribution contribution, @NonNull ChunkContext chunkContext) throws Exception {
// If no record is persisted in summary table then skip persisting data in tracking as well
Optional<StepExecution> jobSummaryStepExecution = contribution.getStepExecution().getJobExecution().getStepExecutions().stream()
.filter(stepExecution -> stepExecution.getStepName().equals(JournalEntryAggregationJobConstant.JOB_SUMMARY_STEP_NAME))
.findFirst();
long writeCount = jobSummaryStepExecution.map(StepExecution::getWriteCount).orElse(0L);
log.info("Starting journal entry aggregation tasklet to insert into tracking table writeCount={}", writeCount);
if (writeCount > 0) {
final JobExecution jobExecutionContext = chunkContext.getStepContext().getStepExecution().getJobExecution();
final LocalDate aggregatedOnDateFrom = (LocalDate) jobExecutionContext.getExecutionContext()
.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_FROM);
final LocalDate aggregatedOnDateTo = (LocalDate) jobExecutionContext.getExecutionContext()
.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_TO);
JournalEntryAggregationTrackingData journalEntrySummaryTrackingDTO = JournalEntryAggregationTrackingData.builder()
.submittedOnDate(ThreadLocalContextUtil.getBusinessDate()).aggregatedOnDateFrom(aggregatedOnDateFrom)
.aggregatedOnDateTo(aggregatedOnDateTo).jobExecutionId(contribution.getStepExecution().getJobExecution().getId())
.build();
journalEntryAggregationWriterService.insertJournalEntryTracking(journalEntrySummaryTrackingDTO);
}
return RepeatStatus.FINISHED;
}
}
@@ -71,6 +71,11 @@ fineract.correlation.header-name=${FINERACT_LOGGING_HTTP_CORRELATION_ID_HEADER_N
fineract.job.stuck-retry-threshold=${FINERACT_JOB_STUCK_RETRY_THRESHOLD:5}
fineract.ip-tracking.enabled=${FINERACT_CLIENT_IP_TRACKING_ENABLED:false}
fineract.job.loan-cob-enabled=${FINERACT_JOB_LOAN_COB_ENABLED:true}
# Aggregation job configuration
fineract.job.journal-entry-aggregation.exclude-recent-N-days=${FINERACT_JOB_JOURNAL_ENTRY_AGGREGATION_EXCLUDE_RECENT_N_DAYS:1}
fineract.job.journal-entry-aggregation.enabled=${FINERACT_JOB_JOURNAL_ENTRY_AGGREGATION_ENABLED:true}
#this property if enabled, will create aggregated entry for all data on first run, instead of one entry per submitted_on_date
fineract.job.journal-entry-aggregation.chunk-size=${FINERACT_JOB_JOURNAL_ENTRY_AGGREGATION_CHUNK_SIZE:2000}
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}
@@ -218,4 +218,7 @@
<include file="parts/0197_add_deleted_and_closed_to_capitalized_income_balance.xml" relativeToChangelogFile="true" />
<include file="parts/0198_add_classification_id_to_acc_product_mapping.xml" relativeToChangelogFile="true" />
<include file="parts/0199_write_off_reason_mapping_loan.xml" relativeToChangelogFile="true" />
<include file="parts/0200_add_journal_entry_aggregation_tables.xml" relativeToChangelogFile="true"/>
<include file="parts/0201_add_journal_entry_aggregation_job.xml" relativeToChangelogFile="true"/>
<include file="parts/0202_trial_balance_summary_with_asset_owner_journal_entry_aggregation.xml" relativeToChangelogFile="true" />
</databaseChangeLog>
@@ -0,0 +1,246 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.1.xsd">
<changeSet id="1" author="fineract" context="postgresql">
<preConditions onFail="MARK_RAN">
<not>
<tableExists tableName="m_journal_entry_aggregation_summary"/>
</not>
</preConditions>
<createTable tableName="m_journal_entry_aggregation_summary">
<column autoIncrement="true" name="id" type="BIGINT">
<constraints nullable="false" primaryKey="true" primaryKeyName="pk_journal_entry_aggregation_summary"/>
</column>
<column name="gl_account_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="product_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="office_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="entity_type_enum" type="SMALLINT">
<constraints nullable="false"/>
</column>
<column name="aggregated_on_date" type="DATE">
<constraints nullable="false"/>
</column>
<column name="submitted_on_date" type="DATE">
<constraints nullable="false"/>
</column>
<column name="external_owner_id" type="BIGINT"/>
<column name="debit_amount" type="DECIMAL(19, 6)"/>
<column name="credit_amount" type="DECIMAL(19, 6)"/>
<column defaultValueBoolean="false" name="manual_entry" type="boolean">
<constraints nullable="false"/>
</column>
<column name="job_execution_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="created_by" type="BIGINT"/>
<column name="created_on_utc" type="timestamp with time zone"/>
<column name="last_modified_by" type="BIGINT"/>
<column name="last_modified_on_utc" type="timestamp with time zone"/>
</createTable>
<addForeignKeyConstraint baseColumnNames="created_by" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_CREATED_BY" referencedColumnNames="id" referencedTableName="m_appuser"/>
<addForeignKeyConstraint baseColumnNames="last_modified_by" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_LAST_MODIFIED_BY" referencedColumnNames="id" referencedTableName="m_appuser"/>
<addForeignKeyConstraint baseColumnNames="office_id" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_OFFICE_ID" referencedColumnNames="id" referencedTableName="m_office"/>
<addForeignKeyConstraint baseColumnNames="gl_account_id" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_GL_ACCOUNT_ID" referencedColumnNames="id" referencedTableName="acc_gl_account"/>
<addForeignKeyConstraint baseColumnNames="job_execution_id" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_JOB_EXECUTION_ID" referencedColumnNames="job_execution_id" referencedTableName="batch_job_execution"/>
</changeSet>
<changeSet id="2" author="fineract" runInTransaction="false" context="postgresql">
<preConditions onFail="MARK_RAN">
<not>
<indexExists tableName="m_journal_entry_aggregation_summary" columnNames="submitted_on_date,gl_account_id"/>
</not>
</preConditions>
<sql>
create index concurrently idx_journal_entry_aggregation_summary on m_journal_entry_aggregation_summary(submitted_on_date,gl_account_id);
</sql>
</changeSet>
<changeSet id="3" author="fineract" context="postgresql">
<preConditions onFail="MARK_RAN">
<not>
<tableExists tableName="m_journal_entry_aggregation_tracking"/>
</not>
</preConditions>
<createTable tableName="m_journal_entry_aggregation_tracking">
<column autoIncrement="true" name="id" type="BIGINT">
<constraints nullable="false" primaryKey="true" primaryKeyName="pk_journal_entry_aggregation_tracking"/>
</column>
<column name="aggregated_on_date_from" type="DATE">
<constraints nullable="false"/>
</column>
<column name="aggregated_on_date_to" type="DATE">
<constraints nullable="false"/>
</column>
<column name="submitted_on_date" type="DATE">
<constraints nullable="false"/>
</column>
<column name="job_execution_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="created_by" type="BIGINT"/>
<column name="created_on_utc" type="timestamp with time zone"/>
<column name="last_modified_by" type="BIGINT"/>
<column name="last_modified_on_utc" type="timestamp with time zone"/>
</createTable>
<addForeignKeyConstraint baseColumnNames="job_execution_id" baseTableName="m_journal_entry_aggregation_tracking" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_TRACKING_ON_JOB_EXECUTION_ID" referencedColumnNames="job_execution_id" referencedTableName="batch_job_execution"/>
<addForeignKeyConstraint baseColumnNames="created_by" baseTableName="m_journal_entry_aggregation_tracking" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_TRACKING_ON_CREATED_BY" referencedColumnNames="id" referencedTableName="m_appuser"/>
<addForeignKeyConstraint baseColumnNames="last_modified_by" baseTableName="m_journal_entry_aggregation_tracking" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_TRACKING_ON_LAST_MODIFIED_BY" referencedColumnNames="id" referencedTableName="m_appuser"/>
</changeSet>
<changeSet id="4" author="fineract" runInTransaction="false" context="postgresql">
<preConditions onFail="MARK_RAN">
<not>
<indexExists tableName="m_journal_entry_aggregation_tracking" columnNames="submitted_on_date"/>
</not>
</preConditions>
<sql>
create index concurrently idx_m_journal_entry_aggregation_tracking on m_journal_entry_aggregation_tracking(submitted_on_date);
</sql>
</changeSet>
<changeSet id="5" author="fineract" runInTransaction="false" context="postgresql">
<preConditions onFail="MARK_RAN">
<not>
<indexExists tableName="m_journal_entry_aggregation_tracking" columnNames="aggregated_on_date_to"/>
</not>
</preConditions>
<sql>
create unique index concurrently idx2_m_journal_entry_aggregation_tracking on m_journal_entry_aggregation_tracking(aggregated_on_date_to);
</sql>
</changeSet>
<changeSet id="6" author="fineract" context="mysql">
<preConditions onFail="MARK_RAN">
<not>
<tableExists tableName="m_journal_entry_aggregation_summary"/>
</not>
</preConditions>
<createTable tableName="m_journal_entry_aggregation_summary">
<column autoIncrement="true" name="id" type="BIGINT">
<constraints nullable="false" primaryKey="true" primaryKeyName="pk_journal_entry_aggregation_summary"/>
</column>
<column name="gl_account_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="product_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="office_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="entity_type_enum" type="SMALLINT">
<constraints nullable="false"/>
</column>
<column name="aggregated_on_date" type="DATE">
<constraints nullable="false"/>
</column>
<column name="submitted_on_date" type="DATE">
<constraints nullable="false"/>
</column>
<column name="external_owner_id" type="BIGINT"/>
<column name="debit_amount" type="DECIMAL(19, 6)"/>
<column name="credit_amount" type="DECIMAL(19, 6)"/>
<column defaultValueBoolean="false" name="manual_entry" type="boolean">
<constraints nullable="false"/>
</column>
<column name="job_execution_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="created_by" type="BIGINT"/>
<column name="created_on_utc" type="DATETIME"/>
<column name="last_modified_by" type="BIGINT"/>
<column name="last_modified_on_utc" type="DATETIME"/>
</createTable>
<addForeignKeyConstraint baseColumnNames="created_by" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_CREATED_BY" referencedColumnNames="id" referencedTableName="m_appuser"/>
<addForeignKeyConstraint baseColumnNames="last_modified_by" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_LAST_MODIFIED_BY" referencedColumnNames="id" referencedTableName="m_appuser"/>
<addForeignKeyConstraint baseColumnNames="office_id" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_OFFICE_ID" referencedColumnNames="id" referencedTableName="m_office"/>
<addForeignKeyConstraint baseColumnNames="gl_account_id" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_GL_ACCOUNT_ID" referencedColumnNames="id" referencedTableName="acc_gl_account"/>
<addForeignKeyConstraint baseColumnNames="job_execution_id" baseTableName="m_journal_entry_aggregation_summary" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_SUMMARY_ON_JOB_EXECUTION_ID" referencedColumnNames="JOB_EXECUTION_ID" referencedTableName="BATCH_JOB_EXECUTION"/>
</changeSet>
<changeSet id="7" author="fineract" runInTransaction="false" context="mysql">
<preConditions onFail="MARK_RAN">
<not>
<indexExists tableName="m_journal_entry_aggregation_summary" columnNames="submitted_on_date,gl_account_id"/>
</not>
</preConditions>
<sql>
create index idx_journal_entry_aggregation_summary on m_journal_entry_aggregation_summary(submitted_on_date,gl_account_id);
</sql>
</changeSet>
<changeSet id="8" author="fineract" context="mysql">
<preConditions onFail="MARK_RAN">
<not>
<tableExists tableName="m_journal_entry_aggregation_tracking"/>
</not>
</preConditions>
<createTable tableName="m_journal_entry_aggregation_tracking">
<column autoIncrement="true" name="id" type="BIGINT">
<constraints nullable="false" primaryKey="true" primaryKeyName="pk_journal_entry_aggregation_tracking"/>
</column>
<column name="aggregated_on_date_from" type="DATE">
<constraints nullable="false"/>
</column>
<column name="aggregated_on_date_to" type="DATE">
<constraints nullable="false"/>
</column>
<column name="submitted_on_date" type="DATE">
<constraints nullable="false"/>
</column>
<column name="job_execution_id" type="BIGINT">
<constraints nullable="false"/>
</column>
<column name="created_by" type="BIGINT"/>
<column name="created_on_utc" type="DATETIME"/>
<column name="last_modified_by" type="BIGINT"/>
<column name="last_modified_on_utc" type="DATETIME"/>
</createTable>
<addForeignKeyConstraint baseColumnNames="job_execution_id" baseTableName="m_journal_entry_aggregation_tracking" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_TRACKING_ON_JOB_EXECUTION_ID" referencedColumnNames="JOB_EXECUTION_ID" referencedTableName="BATCH_JOB_EXECUTION"/>
<addForeignKeyConstraint baseColumnNames="created_by" baseTableName="m_journal_entry_aggregation_tracking" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_TRACKING_ON_CREATED_BY" referencedColumnNames="id" referencedTableName="m_appuser"/>
<addForeignKeyConstraint baseColumnNames="last_modified_by" baseTableName="m_journal_entry_aggregation_tracking" constraintName="FK_GL_JOURNAL_ENTRY_AGGREGATION_TRACKING_ON_LAST_MODIFIED_BY" referencedColumnNames="id" referencedTableName="m_appuser"/>
</changeSet>
<changeSet id="9" author="fineract" runInTransaction="false" context="mysql">
<preConditions onFail="MARK_RAN">
<not>
<indexExists tableName="m_journal_entry_aggregation_tracking" columnNames="submitted_on_date"/>
</not>
</preConditions>
<sql>
create index idx_m_journal_entry_aggregation_tracking on m_journal_entry_aggregation_tracking(submitted_on_date);
</sql>
</changeSet>
<changeSet id="10" author="fineract" runInTransaction="false" context="mysql">
<preConditions onFail="MARK_RAN">
<not>
<indexExists tableName="m_journal_entry_aggregation_tracking" columnNames="aggregated_on_date_to"/>
</not>
</preConditions>
<sql>
create unique index idx2_m_journal_entry_aggregation_tracking on m_journal_entry_aggregation_tracking(aggregated_on_date_to);
</sql>
</changeSet>
</databaseChangeLog>
@@ -0,0 +1,50 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.3.xsd">
<property name="current_datetime" value="NOW()"/>
<changeSet id="1" author="fineract">
<preConditions onFail="MARK_RAN">
<sqlCheck expectedResult="0">select count(1) from job where short_name = 'JRNL_ENTRY_AGG'</sqlCheck>
</preConditions>
<insert tableName="job">
<column name="name" value="Journal Entry Aggregation"/>
<column name="display_name" value="Journal Entry Aggregation"/>
<column name="cron_expression" value="0 0 6 * * ?"/>
<column name="create_time" valueDate="${current_datetime}"/>
<column name="task_priority" valueNumeric="5"/>
<column name="group_name"/>
<column name="previous_run_start_time"/>
<column name="job_key" value="Journal Entry Aggregation _ DEFAULT"/>
<column name="initializing_errorlog"/>
<column name="is_active" valueBoolean="false"/>
<column name="currently_running" valueBoolean="false"/>
<column name="updates_allowed" valueBoolean="true"/>
<column name="scheduler_group" valueNumeric="0"/>
<column name="is_misfired" valueBoolean="false"/>
<column name="node_id" valueNumeric="1"/>
<column name="is_mismatched_job" valueBoolean="true"/>
<column name="short_name" value="JRNL_AGG"/>
</insert>
</changeSet>
</databaseChangeLog>
@@ -0,0 +1,180 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.3.xsd">
<changeSet author="fineract" id="trial-balance-summary-with-asset-owner-update-1">
<update tableName="stretchy_report">
<column name="report_sql" value="WITH retained_earning AS (
SELECT DISTINCT '${endDate}' AS postingdate,
lp.name AS product,
gl_code AS glacct,
COALESCE((SELECT name FROM acc_gl_account WHERE gl_code = e.gl_code), '') AS description,
COALESCE(e.owner_external_id, 'self') AS assetowner,
SUM(opening_balance_amount) AS beginningbalance,
0 AS debitmovement,
0 AS creditmovement,
SUM(opening_balance_amount) AS endingbalance
FROM acc_gl_journal_entry_annual_summary e, m_product_loan lp
WHERE e.office_id = ${officeId}
AND lp.id = product_id
AND date_part('year', e.year_end_date::date) &lt; date_part('year', '${endDate}'::date)
GROUP BY gl_code, lp.name, office_id, owner_external_id
),
aggregated_date AS (
SELECT MAX(aggregated_on_date_to) as latest
FROM m_journal_entry_aggregation_tracking
),
summary_snapshot_baseline_data AS (
--using m_journal_entry_aggregation_summary to fetch aggregated journal entry data
SELECT lp.NAME AS productname,
acc_gl_account.gl_code AS glcode,
acc_gl_account.NAME AS glname,
CASE WHEN ags.external_owner_id IS NULL THEN 0 ELSE ags.external_owner_id END AS assetowner,
SUM(ags.debit_amount) AS debitamount,
SUM(ags.credit_amount) AS creditamount
FROM acc_gl_account
JOIN m_journal_entry_aggregation_summary ags ON acc_gl_account.id = ags.gl_account_id
JOIN m_product_loan lp ON lp.id = ags.product_id
WHERE ags.entity_type_enum = 1
AND ags.manual_entry = false
AND ags.submitted_on_date &lt;= (select latest from aggregated_date)
AND (ags.office_id = ${officeId})
GROUP BY productname, glcode, glname, assetowner
),
post_snapshot_delta_data AS (
-- still using acc_gl_journal_entry for dates after last aggregated data and before the cob date
SELECT lp.NAME AS productname,
acc_gl_account.gl_code AS glcode,
acc_gl_account.NAME AS glname,
CASE WHEN aw.owner_id IS NULL THEN 0 ELSE aw.owner_id END AS assetowner,
SUM(CASE WHEN acc_gl_journal_entry.type_enum = 2 THEN amount ELSE 0 END) AS debitamount,
SUM(CASE WHEN acc_gl_journal_entry.type_enum = 1 THEN amount ELSE 0 END) AS creditamount
FROM acc_gl_account
JOIN acc_gl_journal_entry ON acc_gl_account.id = acc_gl_journal_entry.account_id
JOIN m_loan m ON m.id = acc_gl_journal_entry.entity_id
JOIN m_product_loan lp ON lp.id = m.product_id
LEFT JOIN m_external_asset_owner_journal_entry_mapping aw ON aw.journal_entry_id = acc_gl_journal_entry.id
WHERE acc_gl_journal_entry.entity_type_enum = 1
AND acc_gl_journal_entry.manual_entry = false
AND acc_gl_journal_entry.submitted_on_date &gt; (select latest from aggregated_date)
AND acc_gl_journal_entry.submitted_on_date &lt; '${endDate}'
AND (acc_gl_journal_entry.office_id = ${officeId})
GROUP BY productname, glcode, glname, assetowner
),
-- fetch all data before cob date
merged_historical_data AS (
SELECT summary_snapshot_baseline_data.productname,
summary_snapshot_baseline_data.glcode,
summary_snapshot_baseline_data.glname,
summary_snapshot_baseline_data.assetowner,
COALESCE(summary_snapshot_baseline_data.debitamount, 0) + COALESCE(post_snapshot_delta_data.debitamount, 0) AS debitamount,
COALESCE(summary_snapshot_baseline_data.creditamount, 0) + COALESCE(post_snapshot_delta_data.creditamount, 0) AS creditamount
FROM summary_snapshot_baseline_data
FULL OUTER JOIN post_snapshot_delta_data ON summary_snapshot_baseline_data.glcode = post_snapshot_delta_data.glcode
AND summary_snapshot_baseline_data.productname = post_snapshot_delta_data.productname
AND summary_snapshot_baseline_data.assetowner = post_snapshot_delta_data.assetowner
),
-- fetch data for cob date
current_cob_data AS (
SELECT lp.name AS productname,
account_id,
acc_gl_account.gl_code AS glcode,
acc_gl_account.name AS glname,
CASE WHEN aw.owner_id IS NULL THEN 0 ELSE aw.owner_id END AS assetowner,
SUM(CASE WHEN acc_gl_journal_entry.type_enum = 2 THEN amount ELSE 0 END) AS debitamount,
SUM(CASE WHEN acc_gl_journal_entry.type_enum = 1 THEN amount ELSE 0 END) AS creditamount
FROM acc_gl_journal_entry
JOIN acc_gl_account ON acc_gl_account.id = acc_gl_journal_entry.account_id
JOIN m_loan m ON m.id = acc_gl_journal_entry.entity_id
JOIN m_product_loan lp ON lp.id = m.product_id
LEFT JOIN m_external_asset_owner_journal_entry_mapping aw ON aw.journal_entry_id = acc_gl_journal_entry.id
WHERE acc_gl_journal_entry.entity_type_enum = 1
AND acc_gl_journal_entry.manual_entry = false
AND acc_gl_journal_entry.submitted_on_date = '${endDate}'
AND (acc_gl_journal_entry.office_id = ${officeId})
GROUP BY productname, account_id, glcode, glname, assetowner
)
SELECT *
FROM (
SELECT *
FROM retained_earning
WHERE glacct = (SELECT gl_code FROM acc_gl_account WHERE name = 'Retained Earnings Prior Year')
UNION
SELECT txnreport.postingdate,
txnreport.product,
txnreport.glacct,
txnreport.description,
txnreport.assetowner,
(COALESCE(txnreport.beginningbalance, 0) + COALESCE(summary.beginningbalance, 0)) AS beginningbalance,
txnreport.debitmovement AS debitmovement,
txnreport.creditmovement AS creditmovement,
(COALESCE(txnreport.endingbalance, 0) + COALESCE(summary.beginningbalance, 0)) AS endingbalance
FROM (
SELECT *
FROM (
SELECT DISTINCT '${endDate}' AS postingdate,
loan.pname AS product,
loan.gl_code AS glacct,
loan.glname AS description,
COALESCE((SELECT external_id FROM m_external_asset_owner WHERE id = loan.assetowner), 'self') AS
assetowner,
loan.openingbalance AS beginningbalance,
(loan.debitamount * 1) AS debitmovement,
(loan.creditamount * -1) AS creditmovement,
(loan.openingbalance + loan.debitamount - loan.creditamount) AS endingbalance
FROM (
SELECT DISTINCT COALESCE(g.pname, current_cob_data.productname) pname,
COALESCE(g.gl_code, current_cob_data.glcode) gl_code,
COALESCE(g.glname, current_cob_data.glname) glname,
COALESCE(merged_historical_data.assetowner, current_cob_data.assetowner) assetowner,
COALESCE(merged_historical_data.debitamount, 0) - COALESCE(merged_historical_data.creditamount, 0) AS openingbalance,
COALESCE(current_cob_data.debitamount, 0) AS debitamount,
COALESCE(current_cob_data.creditamount, 0) AS creditamount
FROM (
SELECT DISTINCT ag.gl_code, ag.id, pl.NAME AS pname, ag.NAME AS glname
FROM acc_gl_account ag
JOIN acc_product_mapping am ON am.gl_account_id = ag.id AND am.product_type = 1
JOIN m_product_loan pl ON pl.id = am.product_id
) g
LEFT JOIN merged_historical_data ON g.gl_code = merged_historical_data.glcode AND merged_historical_data.productname = g.pname
FULL OUTER JOIN current_cob_data ON g.gl_code = current_cob_data.glcode
AND current_cob_data.productname = g.pname
AND merged_historical_data.assetowner = current_cob_data.assetowner
) loan
) a
) AS txnreport
LEFT JOIN retained_earning summary ON txnreport.glacct = summary.glacct
AND txnreport.assetowner = summary.assetowner
AND summary.product = txnreport.product
) report
WHERE report.endingbalance != 0
OR report.debitmovement != 0
OR report.creditmovement != 0
ORDER BY glacct"/>
<where>report_name='Trial Balance Summary Report with Asset Owner'</where>
</update>
</changeSet>
</databaseChangeLog>
@@ -0,0 +1,92 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.mockito.BDDMockito.given;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.apache.fineract.infrastructure.core.service.migration.TenantDataSourceFactory;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.listener.JournalEntryAggregationJobListener;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.tasklet.JournalEntryAggregationTrackingTasklet;
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.springframework.batch.core.JobExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.scope.context.JobSynchronizationManager;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.transaction.PlatformTransactionManager;
@EnableConfigurationProperties({ FineractProperties.class })
@ExtendWith(MockitoExtension.class)
class JournalEntryAggregationJobConfigurationTest {
@InjectMocks
private JournalEntryAggregationJobConfiguration configuration;
@Mock
JournalEntryAggregationJobListener journalEntryAggregationJobListener;
@Mock
private JobRepository jobRepository;
@Mock
private JournalEntryAggregationJobExecutionDecider journalEntryAggregationJobExecutionDecider;
@Mock
private JournalEntryAggregationJobWriter aggregationItemWriter;
@Mock
private PlatformTransactionManager transactionManager;
@Mock
private JournalEntryAggregationTrackingTasklet journalEntryAggregationTrackingTasklet;
@Mock
private TenantDataSourceFactory tenantDataSourceFactory;
@Mock
private FineractProperties fineractProperties;
@Mock
private FineractProperties.FineractJobProperties fineractJobProperties;
@Mock
private FineractProperties.FineractJournalEntryAggregationProperties journalEntryAggregationProperties;
@Mock
private JobExecution jobExecution;
@Mock
private ExecutionContext executionContext;
/**
* Test method for {@link JournalEntryAggregationJobConfiguration#journalEntryAggregation()}.
*/
@Test
public void testJournalEntryDailyAggregationJob() {
given(fineractProperties.getJob()).willReturn(fineractJobProperties);
given(fineractJobProperties.getJournalEntryAggregation()).willReturn(journalEntryAggregationProperties);
given(fineractJobProperties.getJournalEntryAggregation().getChunkSize()).willReturn(5);
JobSynchronizationManager.register(jobExecution);
assertNotNull(configuration.journalEntryAggregation(), "The journalEntryDailyAggregationJob bean should not be null");
}
}
@@ -0,0 +1,133 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.when;
import java.time.Clock;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.stream.Stream;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.item.ExecutionContext;
@ExtendWith(MockitoExtension.class)
public class JournalEntryAggregationJobExecutionDeciderTest {
final LocalDate aggregatedOnDate = LocalDate.now(ZoneId.of("UTC"));
@InjectMocks
private JournalEntryAggregationJobExecutionDecider decider;
@Mock
private JobExecution jobExecution;
@Mock
private StepExecution stepExecution;
@Mock
private ExecutionContext executionContext;
@Mock
private FineractProperties.FineractJobProperties fineractJobProperties;
@BeforeEach
public void setUp() {
when(jobExecution.getExecutionContext()).thenReturn(executionContext);
when(jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE)).thenReturn(aggregatedOnDate);
}
@Test
public void testDecideContinueExecution() {
// Arrange
LocalDate lastAggregatedOnDate = aggregatedOnDate.minusDays(1);
when(jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE))
.thenReturn(lastAggregatedOnDate);
// Act
FlowExecutionStatus status = decider.decide(jobExecution, stepExecution);
// Assert
assertEquals(JournalEntryAggregationJobConstant.CONTINUE_JOB_EXECUTION, status.getName(),
"Should continue execution when lastAggregatedOnDate is before aggregatedOnDate");
}
@ParameterizedTest
@MethodSource("provideNoOpExecutionScenarios")
public void testDecideNoOpExecution(LocalDate lastAggregatedOnDate, String scenarioDescription) {
// Arrange
when(jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE))
.thenReturn(lastAggregatedOnDate);
when(jobExecution.getExitStatus()).thenReturn(ExitStatus.NOOP);
// Act
FlowExecutionStatus status = decider.decide(jobExecution, stepExecution);
// Assert
assertEquals(JournalEntryAggregationJobConstant.NO_OP_EXECUTION, status.getName(),
"Should perform no-op execution when " + scenarioDescription);
assertEquals(ExitStatus.NOOP, jobExecution.getExitStatus(),
"Exit status should remain NOOP for scenario: " + scenarioDescription);
}
private static Stream<Arguments> provideNoOpExecutionScenarios() {
LocalDate aggregatedOnDate = LocalDate.now(Clock.systemUTC());
return Stream.of(Arguments.of(aggregatedOnDate, "lastAggregatedOnDate equals aggregatedOnDate"),
Arguments.of(aggregatedOnDate.plusDays(1), "lastAggregatedOnDate is after aggregatedOnDate"));
}
@Test
public void testDecideFirstTimeExecution() {
// Arrange
when(jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE))
.thenReturn(null);
// Act
FlowExecutionStatus status = decider.decide(jobExecution, stepExecution);
// Assert
assertEquals(JournalEntryAggregationJobConstant.CONTINUE_JOB_EXECUTION, status.getName(),
"Should continue execution when no previous execution date exists (first time execution)");
}
@Test
public void testDecideContinueExecutionWithMultipleDaysDifference() {
// Arrange
LocalDate lastAggregatedOnDate = aggregatedOnDate.minusDays(5);
when(jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE))
.thenReturn(lastAggregatedOnDate);
// Act
FlowExecutionStatus status = decider.decide(jobExecution, stepExecution);
// Assert
assertEquals(JournalEntryAggregationJobConstant.CONTINUE_JOB_EXECUTION, status.getName(),
"Should continue execution when lastAggregatedOnDate is multiple days before aggregatedOnDate");
}
}
@@ -0,0 +1,151 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.mockito.Mockito.when;
import com.zaxxer.hikari.HikariDataSource;
import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.sql.Date;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.HashMap;
import java.util.Map;
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenantConnection;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.core.service.migration.TenantDataSourceFactory;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationSummaryData;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.scope.context.JobSynchronizationManager;
import org.springframework.batch.item.ExecutionContext;
@ExtendWith(MockitoExtension.class)
public class JournalEntryAggregationJobReaderTest {
@Mock
private FineractPlatformTenantConnection fineractPlatformTenantConnection;
@Mock
private TenantDataSourceFactory tenantDataSourceFactory;
@Mock
private JobExecution jobExecution;
@Mock
private ExecutionContext executionContext;
@Mock
private HikariDataSource dataSource;
@Mock
private ResultSet resultSet;
private JournalEntryAggregationJobReader reader;
private FineractPlatformTenant tenant;
@BeforeEach
public void setUp() {
tenant = new FineractPlatformTenant(1L, "default", "Default Tenant", "default", fineractPlatformTenantConnection);
ThreadLocalContextUtil.setTenant(tenant);
ThreadLocalContextUtil
.setBusinessDates(new HashMap<>(Map.of(BusinessDateType.BUSINESS_DATE, LocalDate.now(ZoneId.systemDefault()))));
}
@AfterEach
public void tearDown() {
ThreadLocalContextUtil.reset();
if (JobSynchronizationManager.getContext() != null) {
JobSynchronizationManager.close();
}
}
@Test
public void testRowMapping() throws Exception {
// Arrange
setupResultSetMocks();
// Act
reader = new JournalEntryAggregationJobReader(tenantDataSourceFactory);
JournalEntryAggregationSummaryData result = invokeMapRowMethod(resultSet, 1);
// Assert
assertNotNull(result, "Mapped result should not be null");
assertEquals(Long.valueOf(1001L), result.getGlAccountId(), "GL Account ID should match");
assertEquals(100L, result.getProductId(), "Product ID should match");
assertEquals(Long.valueOf(1L), result.getOffice(), "Office should be 1L");
assertEquals(1L, result.getEntityTypeEnum(), "Entity type should be 1");
assertEquals("USD", result.getCurrencyCode(), "Currency code should match");
assertEquals(LocalDate.of(2023, 6, 15), result.getAggregatedOnDate(), "Aggregated date should match");
assertEquals(Long.valueOf(500L), result.getExternalOwnerId(), "Asset owner should match");
assertEquals(new BigDecimal("1000.00"), result.getDebitAmount(), "Debit amount should match");
assertEquals(Boolean.FALSE, result.getManualEntry(), "Manual entry should be false");
assertEquals(ThreadLocalContextUtil.getBusinessDate(), result.getSubmittedOnDate(), "Submitted on date should match business date");
}
@Test
public void testRowMappingWithNullAssetOwner() throws Exception {
// Arrange
setupResultSetMocksWithNullOwner();
// Act
reader = new JournalEntryAggregationJobReader(tenantDataSourceFactory);
JournalEntryAggregationSummaryData result = invokeMapRowMethod(resultSet, 1);
// Assert
assertEquals(Long.valueOf(0L), result.getExternalOwnerId(), "Asset owner should be 0 when null in resultset");
}
private void setupResultSetMocks() throws SQLException {
when(resultSet.getLong("glAccountId")).thenReturn(1001L);
when(resultSet.getLong("productId")).thenReturn(100L);
when(resultSet.getString("currencyCode")).thenReturn("USD");
when(resultSet.getDate("aggregatedOnDate")).thenReturn(Date.valueOf(LocalDate.of(2023,6,15)));
when(resultSet.findColumn("externalOwner")).thenReturn(5);
when(resultSet.getLong(5)).thenReturn(500L);
when(resultSet.getLong("officeId")).thenReturn(1L);
when(resultSet.getLong("entityTypeEnum")).thenReturn(1L);
when(resultSet.getBigDecimal("debitAmount")).thenReturn(new BigDecimal("1000.00"));
when(resultSet.getBigDecimal("creditAmount")).thenReturn(new BigDecimal("800.00"));
}
private void setupResultSetMocksWithNullOwner() throws SQLException {
when(resultSet.getLong("glAccountId")).thenReturn(1001L);
when(resultSet.getLong("productId")).thenReturn(100L);
when(resultSet.getString("currencyCode")).thenReturn("USD");
when(resultSet.getDate("aggregatedOnDate")).thenReturn(Date.valueOf(LocalDate.of(2023,6,15)));
when(resultSet.getLong("officeId")).thenReturn(1L);
when(resultSet.getLong("entityTypeEnum")).thenReturn(1L);
when(resultSet.getBigDecimal("debitAmount")).thenReturn(new BigDecimal("1000.00"));
when(resultSet.getBigDecimal("creditAmount")).thenReturn(new BigDecimal("800.00"));
}
private JournalEntryAggregationSummaryData invokeMapRowMethod(ResultSet rs, int rowNum) throws Exception {
Method mapRowMethod = JournalEntryAggregationJobReader.class.getDeclaredMethod("mapRow", ResultSet.class, int.class);
mapRowMethod.setAccessible(true);
return (JournalEntryAggregationSummaryData) mapRowMethod.invoke(reader, rs, rowNum);
}
}
@@ -0,0 +1,110 @@
/**
* 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.infrastructure.jobs.service.aggregationjob;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.List;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationSummaryData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.services.JournalEntryAggregationWriterService;
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.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.item.Chunk;
@ExtendWith(MockitoExtension.class)
public class JournalEntryAggregationJobWriterTest {
@InjectMocks
private JournalEntryAggregationJobWriter writer;
@Mock
private JournalEntryAggregationWriterService writerService;
@Test
public void testWrite() {
// Arrange
JournalEntryAggregationSummaryData summaryData1 = mock(JournalEntryAggregationSummaryData.class);
JournalEntryAggregationSummaryData summaryData2 = mock(JournalEntryAggregationSummaryData.class);
Chunk<JournalEntryAggregationSummaryData> chunk = new Chunk<>(List.of(summaryData1, summaryData2));
StepExecution stepExecution = mock(StepExecution.class);
JobExecution jobExecution = mock(JobExecution.class);
when(stepExecution.getJobExecution()).thenReturn(jobExecution);
when(jobExecution.getId()).thenReturn(123L);
writer.beforeStep(stepExecution);
// Act
writer.write(chunk);
// Assert
verify(writerService, times(1)).insertJournalEntrySummaryBatch(anyList());
verify(summaryData1, times(1)).setJobExecutionId(123L);
verify(summaryData2, times(1)).setJobExecutionId(123L);
}
@Test
public void testWriteEmptyChunk() {
// Arrange
Chunk<JournalEntryAggregationSummaryData> emptyChunk = new Chunk<>();
StepExecution stepExecution = mock(StepExecution.class);
JobExecution jobExecution = mock(JobExecution.class);
when(stepExecution.getJobExecution()).thenReturn(jobExecution);
when(jobExecution.getId()).thenReturn(123L);
writer.beforeStep(stepExecution);
// Act
writer.write(emptyChunk);
// Assert
verify(writerService, times(1)).insertJournalEntrySummaryBatch(anyList());
}
@Test
public void testWriteSingleItem() {
// Arrange
JournalEntryAggregationSummaryData summaryData = mock(JournalEntryAggregationSummaryData.class);
Chunk<JournalEntryAggregationSummaryData> chunk = new Chunk<>(List.of(summaryData));
StepExecution stepExecution = mock(StepExecution.class);
JobExecution jobExecution = mock(JobExecution.class);
when(stepExecution.getJobExecution()).thenReturn(jobExecution);
when(jobExecution.getId()).thenReturn(456L);
writer.beforeStep(stepExecution);
// Act
writer.write(chunk);
// Assert
verify(writerService, times(1)).insertJournalEntrySummaryBatch(anyList());
verify(summaryData, times(1)).setJobExecutionId(456L);
}
}
@@ -0,0 +1,131 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.listener;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.mockito.Mockito.when;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.HashMap;
import java.util.Map;
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenantConnection;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.JournalEntryAggregationJobConstant;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntryAggregationTrackingRepository;
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.springframework.batch.core.JobExecution;
import org.springframework.batch.core.scope.context.JobContext;
import org.springframework.batch.item.ExecutionContext;
@ExtendWith(MockitoExtension.class)
public class JournalEntryAggregationJobListenerTest {
private final LocalDate businessDate = LocalDate.now(ZoneId.of("UTC"));
@Mock
FineractPlatformTenantConnection fineractPlatformTenantConnection;
@Mock
private FineractProperties fineractProperties;
@Mock
private FineractProperties.FineractJobProperties fineractJobProperties;
@Mock
private FineractProperties.FineractJournalEntryAggregationProperties fineractJournalEntryAggregationProperties;
@Mock
private JournalEntryAggregationTrackingRepository trackingRepository;
@Mock
private JobExecution jobExecution;
@Mock
private JobContext jobContext;
@Mock
private ExecutionContext executionContext;
@InjectMocks
private JournalEntryAggregationJobListener listener;
@BeforeEach
public void setUp() {
ThreadLocalContextUtil
.setTenant(new FineractPlatformTenant(1L, "default", "Default Tenant", "default", fineractPlatformTenantConnection));
ThreadLocalContextUtil.setBusinessDates(new HashMap<>(Map.of(BusinessDateType.COB_DATE, businessDate.minusDays(1))));
ThreadLocalContextUtil.setBusinessDates(new HashMap<>(Map.of(BusinessDateType.BUSINESS_DATE, businessDate)));
}
@Test
public void testBeforeJobWhenLastAggregatedOnDateNull() {
// Arrange
int lookBackDays = 2;
when(fineractProperties.getJob()).thenReturn(fineractJobProperties);
when(fineractJobProperties.getJournalEntryAggregation()).thenReturn(fineractJournalEntryAggregationProperties);
when(fineractJobProperties.getJournalEntryAggregation().getExcludeRecentNDays()).thenReturn(lookBackDays);
when(trackingRepository.findLatestAggregatedOnDate()).thenReturn(null);
ExecutionContext executionContext = new ExecutionContext();
when(jobExecution.getExecutionContext()).thenReturn(executionContext);
// Act
listener.beforeJob(jobExecution);
// Assert
assertEquals(businessDate.minusDays(lookBackDays), executionContext.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE));
assertEquals(LocalDate.of(1970, 1, 1), executionContext.get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_FROM));
assertNull(executionContext.get(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE));
}
@Test
public void testBeforeJobWithLastAggregatedOnDate() {
// Arrange
LocalDate businessDate = LocalDate.now(ZoneId.of("UTC"));
LocalDate lastAggregatedOnDate = businessDate.minusDays(5);
int lookBackDays = 2;
when(fineractProperties.getJob()).thenReturn(fineractJobProperties);
when(fineractJobProperties.getJournalEntryAggregation()).thenReturn(fineractJournalEntryAggregationProperties);
when(fineractJobProperties.getJournalEntryAggregation().getExcludeRecentNDays()).thenReturn(lookBackDays);
when(trackingRepository.findLatestAggregatedOnDate()).thenReturn(lastAggregatedOnDate);
JobExecution jobExecution = new JobExecution(1L);
// Act
listener.beforeJob(jobExecution);
// Assert
assertEquals(businessDate.minusDays(lookBackDays),
jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE));
assertEquals(businessDate.minusDays(lookBackDays),
jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_TO));
assertEquals(lastAggregatedOnDate,
jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.LAST_AGGREGATED_ON_DATE));
assertEquals(lastAggregatedOnDate,
jobExecution.getExecutionContext().get(JournalEntryAggregationJobConstant.AGGREGATED_ON_DATE_FROM));
}
}
@@ -0,0 +1,153 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.services;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.math.BigDecimal;
import java.time.Clock;
import java.time.LocalDate;
import java.util.Arrays;
import java.util.List;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationSummaryData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationTrackingData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntryAggregationTrackingRepository;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntrySummary;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.domain.JournalEntrySummaryRepository;
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.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
@ExtendWith(MockitoExtension.class)
public class JournalEntryAggregationWriterServiceImplTest {
@InjectMocks
private JournalEntryAggregationWriterServiceImpl writerService;
@Mock
private JournalEntrySummaryRepository journalSummaryRepository;
@Mock
private JournalEntryAggregationTrackingRepository journalEntryAggregationTrackingRepository;
@Mock
private NamedParameterJdbcTemplate namedParameterJdbcTemplate;
@Test
public void testInsertJournalEntryTracking() throws Exception {
// Arrange
JournalEntryAggregationTrackingData trackingData = mock(JournalEntryAggregationTrackingData.class);
// Act
writerService.insertJournalEntryTracking(trackingData);
// Assert
verify(journalEntryAggregationTrackingRepository, times(1)).save(any());
}
@Test
public void testRollbackJournalEntrySummary() {
// Arrange
Long jobExecutionId = 123L;
// Act
writerService.rollbackJournalEntrySummary(jobExecutionId);
// Assert
verify(journalSummaryRepository, times(1)).deleteByJobExecutionId(jobExecutionId);
}
@Test
public void testRollbackJournalEntryTracking() {
// Arrange
final Long jobExecutionId = 123L;
// Act
writerService.rollbackJournalEntryTracking(jobExecutionId);
// Assert
verify(journalEntryAggregationTrackingRepository, times(1)).deleteByJobExecutionId(jobExecutionId);
}
@Test
public void testInsertJournalEntrySummaryBatch() {
// Arrange
JournalEntryAggregationSummaryData summaryData1 = mock(JournalEntryAggregationSummaryData.class);
JournalEntryAggregationSummaryData summaryData2 = mock(JournalEntryAggregationSummaryData.class);
when(summaryData1.getAggregatedOnDate()).thenReturn(LocalDate.now(Clock.systemUTC()));
when(summaryData2.getAggregatedOnDate()).thenReturn(LocalDate.now(Clock.systemUTC()));
when(summaryData1.getJobExecutionId()).thenReturn(123L);
when(summaryData2.getJobExecutionId()).thenReturn(123L);
List<JournalEntryAggregationSummaryData> summariesList = Arrays.asList(summaryData1, summaryData2);
// Act
writerService.insertJournalEntrySummaryBatch(summariesList);
// Assert
verify(journalSummaryRepository, times(1)).saveAll(anyList());
}
@Test
public void testInsertJournalEntrySummaryBatchEmpty() {
// Arrange
List<JournalEntryAggregationSummaryData> emptySummariesList = List.of();
// Act
writerService.insertJournalEntrySummaryBatch(emptySummariesList);
// Assert
verify(journalSummaryRepository, times(1)).saveAll(anyList());
}
@Test
public void testInsertJournalEntrySummaryBatchWithAmounts() {
// Arrange
JournalEntryAggregationSummaryData summaryData = mock(JournalEntryAggregationSummaryData.class);
BigDecimal debitAmount = BigDecimal.valueOf(100.50);
BigDecimal creditAmount = BigDecimal.valueOf(200.75);
when(summaryData.getAggregatedOnDate()).thenReturn(LocalDate.now(Clock.systemUTC()));
when(summaryData.getDebitAmount()).thenReturn(debitAmount);
when(summaryData.getCreditAmount()).thenReturn(creditAmount);
when(summaryData.getJobExecutionId()).thenReturn(123L);
List<JournalEntryAggregationSummaryData> summariesList = List.of(summaryData);
// Act
writerService.insertJournalEntrySummaryBatch(summariesList);
// Assert
verify(journalSummaryRepository, times(1)).saveAll(argThat(entities -> {
if (!(entities instanceof List) || ((List<?>) entities).isEmpty()) {
return false;
}
JournalEntrySummary entity = (JournalEntrySummary) ((List<?>) entities).getFirst();
return entity.getDebitAmount().compareTo(debitAmount) == 0 && entity.getCreditAmount().compareTo(creditAmount) == 0;
}));
verify(summaryData, times(1)).getDebitAmount();
verify(summaryData, times(1)).getCreditAmount();
}
}
@@ -0,0 +1,136 @@
/**
* 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.infrastructure.jobs.service.aggregationjob.tasklet;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.time.Clock;
import java.time.LocalDate;
import java.util.Collections;
import java.util.HashMap;
import java.util.stream.Stream;
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
import org.apache.fineract.infrastructure.core.domain.ActionContext;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.JournalEntryAggregationJobConstant;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.data.JournalEntryAggregationTrackingData;
import org.apache.fineract.infrastructure.jobs.service.aggregationjob.services.JournalEntryAggregationWriterService;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
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.JobExecution;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.RepeatStatus;
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
class JournalEntryAggregationTrackingTaskletTest {
@Mock
private StepContribution stepContribution;
@Mock
private ChunkContext chunkContext;
@Mock
private StepContext stepContext;
@Mock
private StepExecution stepExecution;
@Mock
private JobExecution jobExecutionContext;
@Mock
private ExecutionContext executionContext;
@Mock
private JournalEntryAggregationWriterService journalEntryAggregationWriterService;
@InjectMocks
private JournalEntryAggregationTrackingTasklet underTest;
/**
* DataProvider for the testHappyPath method.
*
* @return the argument list
*/
private static Stream<Arguments> recordWriteCount() {
// spotless:off
return Stream.of(Arguments.of(1L), Arguments.of(0L));
// spotless:on
}
@ParameterizedTest
@MethodSource("recordWriteCount")
public void testHappyPath(final Long recordWriteCount) throws Exception {
given(stepContribution.getStepExecution()).willReturn(stepExecution);
given(stepExecution.getWriteCount()).willReturn(recordWriteCount);
doNothing().when(journalEntryAggregationWriterService).insertJournalEntryTracking(any(JournalEntryAggregationTrackingData.class));
given(chunkContext.getStepContext()).willReturn(stepContext);
given(stepContext.getStepExecution()).willReturn(stepExecution);
given(stepExecution.getJobExecution()).willReturn(jobExecutionContext);
given(jobExecutionContext.getExecutionContext()).willReturn(executionContext);
given(jobExecutionContext.getStepExecutions()).willReturn(Collections.singletonList(stepExecution));
given(stepExecution.getStepName()).willReturn(JournalEntryAggregationJobConstant.JOB_SUMMARY_STEP_NAME);
given(executionContext.get(any())).willReturn(LocalDate.now(Clock.systemUTC()));
final HashMap<BusinessDateType, LocalDate> businessDates = new HashMap<>();
final LocalDate cobDate = LocalDate.now(Clock.systemUTC());
businessDates.put(BusinessDateType.COB_DATE, cobDate);
businessDates.put(BusinessDateType.BUSINESS_DATE, cobDate.plusDays(1));
ThreadLocalContextUtil.setActionContext(ActionContext.COB);
ThreadLocalContextUtil.setBusinessDates(businessDates);
RepeatStatus repeatStatus = underTest.execute(stepContribution, chunkContext);
if (recordWriteCount > 0) {
verify(chunkContext, times(1)).getStepContext();
verify(stepContext, times(1)).getStepExecution();
verify(stepExecution, times(3)).getJobExecution();
verify(jobExecutionContext, times(2)).getExecutionContext();
verify(executionContext, times(2)).get(any());
verify(journalEntryAggregationWriterService, times(1))
.insertJournalEntryTracking(any(JournalEntryAggregationTrackingData.class));
} else {
verify(chunkContext, times(0)).getStepContext();
verify(stepContext, times(0)).getStepExecution();
verify(stepExecution, times(1)).getJobExecution();
verify(jobExecutionContext, times(0)).getExecutionContext();
verify(executionContext, times(0)).get(any());
}
assertEquals(RepeatStatus.FINISHED, repeatStatus);
}
}
@@ -104,6 +104,11 @@ fineract.jpa.statementLoggingEnabled=${FINERACT_STATEMENT_LOGGING_ENABLED:false}
fineract.database.defaultMasterPassword=${FINERACT_DEFAULT_MASTER_PASSWORD:fineract}
fineract.job.loan-cob-enabled=${FINERACT_JOB_LOAN_COB_ENABLED:true}
# Aggregation job configuration
fineract.job.journal-entry-aggregation.exclude-recent-N-days=${FINERACT_JOB_JOURNAL_ENTRY_AGGREGATION_EXCLUDE_RECENT_N_DAYS:1}
fineract.job.journal-entry-aggregation.enabled=${FINERACT_JOB_JOURNAL_ENTRY_AGGREGATION_ENABLED:true}
#this property if enabled, will create aggregated entry for all data on first run, instead of one entry per submitted_on_date
fineract.job.journal-entry-aggregation.chunk-size=${FINERACT_JOB_JOURNAL_ENTRY_AGGREGATION_CHUNK_SIZE:2000}
fineract.sampling.enabled=false
fineract.sampling.sampledClasses=