FINERACT-2332: Savings COB first business steps

fineract-core/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobName.java

@@ -61,6 +61,7 @@ public enum JobName {
     ADD_PERIODIC_ACCRUAL_ENTRIES_FOR_SAVINGS_WITH_INCOME_POSTED_AS_TRANSACTIONS("Add Accrual Transactions For Savings"), //
     JOURNAL_ENTRY_AGGREGATION("Journal Entry Aggregation"), //
     WORKING_CAPITAL_LOAN_COB_JOB("Working Capital Loan COB"), //
+    SAVINGS_COB("Savings COB"), //
     ; //
 
     private final String name;

fineract-doc/src/docs/en/chapters/features/index.adoc

@@ -31,3 +31,4 @@ include::working-capital-planned-projected-balances-eir.adoc[leveloffset=+1]
 include::working-capital-discount.adoc[leveloffset=+1]
 include::savings-interest-posting.adoc[leveloffset=+1]
 include::working-capital-breach-management.adoc[leveloffset=+1]
+include::savings-cob.adoc[leveloffset=+1]

fineract-provider/src/main/java/org/apache/fineract/cob/listener/ChunkProcessingSavingsItemListener.java

@@ -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.cob.listener;
+
+import static org.springframework.transaction.TransactionDefinition.PROPAGATION_REQUIRES_NEW;
+
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.cob.exceptions.LockedReadException;
+import org.apache.fineract.cob.savings.SavingsLockOwner;
+import org.apache.fineract.cob.savings.SavingsLockingService;
+import org.apache.fineract.infrastructure.core.domain.AbstractPersistableCustom;
+import org.apache.fineract.infrastructure.core.serialization.ThrowableSerialization;
+import org.apache.fineract.portfolio.savings.domain.SavingsAccount;
+import org.springframework.batch.core.annotation.OnProcessError;
+import org.springframework.batch.core.annotation.OnReadError;
+import org.springframework.batch.core.annotation.OnSkipInProcess;
+import org.springframework.batch.core.annotation.OnSkipInRead;
+import org.springframework.batch.core.annotation.OnSkipInWrite;
+import org.springframework.batch.core.annotation.OnWriteError;
+import org.springframework.batch.item.Chunk;
+import org.springframework.lang.NonNull;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallbackWithoutResult;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@Slf4j
+@RequiredArgsConstructor
+public class ChunkProcessingSavingsItemListener {
+
+    private final SavingsLockingService savingsLockingService;
+    private final TransactionTemplate batchJdbcTransactionTemplate;
+
+    private void updateLockWithError(List<Long> savingsIds, String msgTemplate, Throwable e) {
+        batchJdbcTransactionTemplate.setPropagationBehavior(PROPAGATION_REQUIRES_NEW);
+        batchJdbcTransactionTemplate.execute(new TransactionCallbackWithoutResult() {
+
+            @Override
+            protected void doInTransactionWithoutResult(@NonNull TransactionStatus status) {
+                String stacktrace = ThrowableSerialization.serialize(e);
+                for (Long savingsId : savingsIds) {
+                    savingsLockingService.updateLockError(savingsId, SavingsLockOwner.SAVINGS_COB_CHUNK_PROCESSING,
+                            String.format(msgTemplate, savingsId), stacktrace);
+                }
+            }
+        });
+    }
+
+    @OnReadError
+    public void onReadError(Exception e) {
+        if (e instanceof LockedReadException ee) {
+            log.warn("Read error for SavingsAccount (id={}) due to: {}", ee.getId(), ThrowableSerialization.serialize(e));
+            updateLockWithError(List.of(ee.getId()), "SavingsAccount (id: %d) reading is failed", e);
+        } else {
+            log.error("Could not handle read error", e);
+        }
+    }
+
+    @OnProcessError
+    public void onProcessError(@NonNull SavingsAccount item, Exception e) {
+        log.warn("Process error for SavingsAccount (id={}) due to: {}", item.getId(), ThrowableSerialization.serialize(e));
+        updateLockWithError(List.of(item.getId()), "SavingsAccount (id: %d) processing is failed", e);
+    }
+
+    @OnWriteError
+    public void onWriteError(Exception e, @NonNull Chunk<? extends SavingsAccount> items) {
+        List<Long> ids = items.getItems().stream().map(AbstractPersistableCustom::getId).toList();
+        log.warn("Write error for SavingsAccounts (ids={}) due to: {}", ids, ThrowableSerialization.serialize(e));
+        updateLockWithError(ids, "SavingsAccount (id: %d) writing is failed", e);
+    }
+
+    @OnSkipInRead
+    public void onSkipInRead(@NonNull Throwable e) {
+        log.warn("Skipping triggered during savings COB read!");
+    }
+
+    @OnSkipInProcess
+    public void onSkipInProcess(@NonNull SavingsAccount item, @NonNull Throwable e) {
+        log.warn("Skipping triggered during processing of SavingsAccount (id={})", item.getId());
+    }
+
+    @OnSkipInWrite
+    public void onSkipInWrite(@NonNull SavingsAccount item, @NonNull Throwable e) {
+        log.warn("Skipping triggered during writing of SavingsAccount (id={})", item.getId());
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/savings/AbstractSavingsItemProcessor.java

@@ -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.cob.savings;
+
+import org.apache.fineract.cob.COBBusinessStepService;
+import org.apache.fineract.cob.processor.AbstractItemProcessor;
+import org.apache.fineract.portfolio.savings.domain.SavingsAccount;
+
+public abstract class AbstractSavingsItemProcessor extends AbstractItemProcessor<SavingsAccount> {
+
+    public AbstractSavingsItemProcessor(COBBusinessStepService cobBusinessStepService) {
+        super(cobBusinessStepService);
+    }
+
+    @Override
+    public void setLastRun(SavingsAccount savingsAccount) {
+        savingsAccount.setLastClosedBusinessDate(getBusinessDate());
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/savings/AbstractSavingsItemReader.java

@@ -0,0 +1,62 @@
+/**
+ * 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.cob.savings;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import lombok.AccessLevel;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.cob.exceptions.LockedReadException;
+import org.apache.fineract.portfolio.savings.domain.SavingsAccount;
+import org.apache.fineract.portfolio.savings.domain.SavingsAccountRepository;
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.annotation.AfterStep;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.lang.NonNull;
+
+@Slf4j
+@RequiredArgsConstructor
+public abstract class AbstractSavingsItemReader implements ItemReader<SavingsAccount> {
+
+    protected final SavingsAccountRepository savingsAccountRepository;
+
+    @Setter(AccessLevel.PROTECTED)
+    private LinkedBlockingQueue<Long> remainingData;
+
+    @Override
+    public SavingsAccount read() throws Exception {
+        final Long savingsId = remainingData.poll();
+        if (savingsId != null) {
+            try {
+                return savingsAccountRepository.findById(savingsId)
+                        .orElseThrow(() -> new RuntimeException("SavingsAccount not found: " + savingsId));
+            } catch (Exception e) {
+                throw new LockedReadException(savingsId, e);
+            }
+        }
+        return null;
+    }
+
+    @AfterStep
+    public ExitStatus afterStep(@NonNull StepExecution stepExecution) {
+        return ExitStatus.COMPLETED;
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/savings/AbstractSavingsItemWriter.java

@@ -0,0 +1,46 @@
+/**
+ * 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.cob.savings;
+
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.domain.AbstractPersistableCustom;
+import org.apache.fineract.portfolio.savings.domain.SavingsAccount;
+import org.springframework.batch.item.Chunk;
+import org.springframework.batch.item.data.RepositoryItemWriter;
+import org.springframework.lang.NonNull;
+
+@Slf4j
+@RequiredArgsConstructor
+public abstract class AbstractSavingsItemWriter extends RepositoryItemWriter<SavingsAccount> {
+
+    private final SavingsLockingService savingsLockingService;
+
+    @Override
+    public void write(@NonNull Chunk<? extends SavingsAccount> items) throws Exception {
+        if (!items.isEmpty()) {
+            super.write(items);
+            List<Long> savingsIds = items.getItems().stream().map(AbstractPersistableCustom::getId).toList();
+            savingsLockingService.deleteBySavingsIdInAndLockOwner(savingsIds, getLockOwner());
+        }
+    }
+
+    protected abstract SavingsLockOwner getLockOwner();
+}

fineract-provider/src/main/java/org/apache/fineract/cob/savings/ApplySavingsLockTasklet.java

@@ -0,0 +1,108 @@
+/**
+ * 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.cob.savings;
+
+import static org.springframework.transaction.TransactionDefinition.PROPAGATION_REQUIRES_NEW;
+
+import com.google.common.collect.Lists;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.cob.converter.COBParameterConverter;
+import org.apache.fineract.cob.data.COBParameter;
+import org.apache.fineract.cob.exceptions.LockCannotBeAppliedException;
+import org.apache.fineract.cob.resolver.CatchUpFlagResolver;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.repeat.RepeatStatus;
+import org.springframework.lang.NonNull;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallbackWithoutResult;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@Slf4j
+@RequiredArgsConstructor
+public class ApplySavingsLockTasklet implements Tasklet {
+
+    private static final long NUMBER_OF_RETRIES = 3;
+    private final FineractProperties fineractProperties;
+    private final SavingsLockingService savingsLockingService;
+    private final RetrieveSavingsIdService retrieveSavingsIdService;
+    private final TransactionTemplate batchJdbcTransactionTemplate;
+
+    @Override
+    @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
+    public RepeatStatus execute(@NonNull StepContribution contribution, @NonNull ChunkContext chunkContext)
+            throws LockCannotBeAppliedException {
+        ExecutionContext executionContext = contribution.getStepExecution().getExecutionContext();
+        long numberOfExecutions = contribution.getStepExecution().getCommitCount();
+        COBParameter savingsCOBParameter = COBParameterConverter.convert(executionContext.get(SavingsCOBConstant.SAVINGS_COB_PARAMETER));
+        boolean isCatchUp = CatchUpFlagResolver.resolve(contribution.getStepExecution());
+
+        List<Long> savingsIds;
+        if (Objects.isNull(savingsCOBParameter)
+                || (Objects.isNull(savingsCOBParameter.getMinAccountId()) && Objects.isNull(savingsCOBParameter.getMaxAccountId()))
+                || (savingsCOBParameter.getMinAccountId().equals(0L) && savingsCOBParameter.getMaxAccountId().equals(0L))) {
+            savingsIds = Collections.emptyList();
+        } else {
+            savingsIds = new ArrayList<>(retrieveSavingsIdService
+                    .retrieveAllNonClosedSavingsByLastClosedBusinessDateAndMinAndMaxSavingsId(savingsCOBParameter, isCatchUp));
+        }
+
+        int limit = fineractProperties.getQuery().getInClauseParameterSizeLimit();
+        List<List<Long>> partitions = Lists.partition(savingsIds, limit);
+        List<Long> alreadyLocked = new ArrayList<>();
+        partitions.forEach(part -> alreadyLocked.addAll(savingsLockingService.findLockIdsBySavingsIdIn(part)));
+
+        List<Long> toBeProcessed = new ArrayList<>(savingsIds);
+        toBeProcessed.removeAll(alreadyLocked);
+
+        try {
+            applyLocks(toBeProcessed);
+        } catch (Exception e) {
+            if (numberOfExecutions > NUMBER_OF_RETRIES) {
+                String message = "There was an error applying lock to savings accounts.";
+                log.error("{}", message, e);
+                throw new LockCannotBeAppliedException(message, e);
+            } else {
+                return RepeatStatus.CONTINUABLE;
+            }
+        }
+        return RepeatStatus.FINISHED;
+    }
+
+    private void applyLocks(List<Long> savingsIds) {
+        batchJdbcTransactionTemplate.setPropagationBehavior(PROPAGATION_REQUIRES_NEW);
+        batchJdbcTransactionTemplate.execute(new TransactionCallbackWithoutResult() {
+
+            @Override
+            protected void doInTransactionWithoutResult(@NonNull TransactionStatus status) {
+                log.info("Applying savings COB locks for {} accounts", savingsIds.size());
+                savingsLockingService.applyLock(savingsIds, SavingsLockOwner.SAVINGS_COB_CHUNK_PROCESSING);
+            }
+        });
+    }
+}