From dcb18d8f5e530ad4a77e26420fe7652ead12c4e5 Mon Sep 17 00:00:00 2001 From: "adam.magyari" Date: Wed, 20 Nov 2024 15:55:04 +0100 Subject: [PATCH] FINERACT-2081: async liquibase --- .../core/config/FineractProperties.java | 3 ++ .../TenantDatabaseUpgradeService.java | 35 +++++++++++++++++-- .../src/main/resources/application.properties | 3 ++ .../core/LiquibaseStepDefinitions.java | 2 ++ .../resources/application-test.properties | 3 ++ 5 files changed, 44 insertions(+), 2 deletions(-) diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java index 39f75109f..961d5a582 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java @@ -282,6 +282,9 @@ public class FineractProperties { private int defaultTaskExecutorCorePoolSize; private int defaultTaskExecutorMaxPoolSize; + private int tenantUpgradeTaskExecutorCorePoolSize; + private int tenantUpgradeTaskExecutorMaxPoolSize; + private int tenantUpgradeTaskExecutorQueueCapacity; } @Getter diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/service/migration/TenantDatabaseUpgradeService.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/service/migration/TenantDatabaseUpgradeService.java index 75a1f8daf..102c12264 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/service/migration/TenantDatabaseUpgradeService.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/service/migration/TenantDatabaseUpgradeService.java @@ -21,10 +21,15 @@ package org.apache.fineract.infrastructure.core.service.migration; import static org.apache.commons.collections4.CollectionUtils.isNotEmpty; import static org.apache.commons.lang3.StringUtils.isNotBlank; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.function.Function; import javax.sql.DataSource; +import liquibase.Scope; +import liquibase.ThreadLocalScopeManager; import liquibase.change.custom.CustomTaskChange; import liquibase.exception.LiquibaseException; import liquibase.integration.spring.SpringLiquibase; @@ -37,6 +42,7 @@ import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsServi import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.core.env.Environment; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; /** @@ -76,6 +82,7 @@ public class TenantDatabaseUpgradeService implements InitializingBean { } } try { + Scope.setScopeManager(new ThreadLocalScopeManager()); upgradeTenantStore(); upgradeIndividualTenants(); } catch (LiquibaseException e) { @@ -121,17 +128,41 @@ public class TenantDatabaseUpgradeService implements InitializingBean { } - private void upgradeIndividualTenants() throws LiquibaseException { + private void upgradeIndividualTenants() { log.info("Upgrading all tenants"); List tenants = tenantDetailsService.findAllTenants(); + List> futures = new ArrayList<>(); + final ThreadPoolTaskExecutor tenantUpgradeThreadPoolTaskExecutor = createTenantUpgradeThreadPoolTaskExecutor(); if (isNotEmpty(tenants)) { for (FineractPlatformTenant tenant : tenants) { - upgradeIndividualTenant(tenant); + futures.add(tenantUpgradeThreadPoolTaskExecutor.submit(() -> { + upgradeIndividualTenant(tenant); + return tenant.getName(); + })); } } + + try { + for (Future future : futures) { + future.get(); + } + } catch (InterruptedException | ExecutionException exception) { + throw new RuntimeException(exception); + } finally { + tenantUpgradeThreadPoolTaskExecutor.shutdown(); + } log.info("Tenant upgrades have finished"); } + private ThreadPoolTaskExecutor createTenantUpgradeThreadPoolTaskExecutor() { + ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor(); + threadPoolTaskExecutor.setCorePoolSize(fineractProperties.getTaskExecutor().getTenantUpgradeTaskExecutorCorePoolSize()); + threadPoolTaskExecutor.setMaxPoolSize(fineractProperties.getTaskExecutor().getTenantUpgradeTaskExecutorMaxPoolSize()); + threadPoolTaskExecutor.setQueueCapacity(fineractProperties.getTaskExecutor().getTenantUpgradeTaskExecutorQueueCapacity()); + threadPoolTaskExecutor.initialize(); + return threadPoolTaskExecutor; + } + /** * Upgrade each tenant's database * diff --git a/fineract-provider/src/main/resources/application.properties b/fineract-provider/src/main/resources/application.properties index 2d5cbbafd..a9899f281 100644 --- a/fineract-provider/src/main/resources/application.properties +++ b/fineract-provider/src/main/resources/application.properties @@ -131,6 +131,9 @@ fineract.events.external.producer.kafka.admin.extra-properties=${FINERACT_EXTERN fineract.task-executor.default-task-executor-core-pool-size=${FINERACT_DEFAULT_TASK_EXECUTOR_CORE_POOL_SIZE:10} fineract.task-executor.default-task-executor-max-pool-size=${FINERACT_DEFAULT_TASK_EXECUTOR_MAX_POOL_SIZE:100} +fineract.task-executor.tenant-upgrade-task-executor-core-pool-size=${FINERACT_TENANT_UPGRADE_TASK_EXECUTOR_CORE_POOL_SIZE:1} +fineract.task-executor.tenant-upgrade-task-executor-max-pool-size=${FINERACT_TENANT_UPGRADE_TASK_EXECUTOR_MAX_POOL_SIZE:1} +fineract.task-executor.tenant-upgrade-task-executor-queue-capacity=${FINERACT_TENANT_UPGRADE_TASK_EXECUTOR_QUEUE_CAPACITY:100} fineract.idempotency-key-header-name=${FINERACT_IDEMPOTENCY_KEY_HEADER_NAME:Idempotency-Key} diff --git a/fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/LiquibaseStepDefinitions.java b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/LiquibaseStepDefinitions.java index e706a9004..238497497 100644 --- a/fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/LiquibaseStepDefinitions.java +++ b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/LiquibaseStepDefinitions.java @@ -104,6 +104,8 @@ public class LiquibaseStepDefinitions implements En { tenantDatabaseUpgradeService.afterPropertiesSet(); } catch (SchemaUpgradeNeededException e) { executionException = e; + } catch (RuntimeException e) { + executionException = (SchemaUpgradeNeededException) e.getCause().getCause(); } }); diff --git a/fineract-provider/src/test/resources/application-test.properties b/fineract-provider/src/test/resources/application-test.properties index 243d4d36c..3f9cba12b 100644 --- a/fineract-provider/src/test/resources/application-test.properties +++ b/fineract-provider/src/test/resources/application-test.properties @@ -65,6 +65,9 @@ fineract.events.external.producer.jms.thread-pool-task-executor-max-pool-size=${ fineract.task-executor.default-task-executor-core-pool-size=${FINERACT_DEFAULT_TASK_EXECUTOR_CORE_POOL_SIZE:10} fineract.task-executor.default-task-executor-max-pool-size=${FINERACT_DEFAULT_TASK_EXECUTOR_MAX_POOL_SIZE:100} +fineract.task-executor.tenant-upgrade-task-executor-core-pool-size=${FINERACT_TENANT_UPGRADE_TASK_EXECUTOR_CORE_POOL_SIZE:1} +fineract.task-executor.tenant-upgrade-task-executor-max-pool-size=${FINERACT_TENANT_UPGRADE_TASK_EXECUTOR_MAX_POOL_SIZE:1} +fineract.task-executor.tenant-upgrade-task-executor-queue-capacity=${FINERACT_TENANT_UPGRADE_TASK_EXECUTOR_QUEUE_CAPACITY:100} fineract.loan.transactionprocessor.creocore.enabled=true fineract.loan.transactionprocessor.early-repayment.enabled=true