FINERACT-2081: async liquibase

marta-jankovics · Nov 25, 2024 · dcb18d8 · dcb18d8
1 parent 4e3bcbf
commit dcb18d8
Show file tree

Hide file tree

Showing 5 changed files with 44 additions and 2 deletions.
diff --git a/...core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java b/...core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java
@@ -282,6 +282,9 @@ public static class FineractTaskExecutor {
 
         private int defaultTaskExecutorCorePoolSize;
         private int defaultTaskExecutorMaxPoolSize;
+        private int tenantUpgradeTaskExecutorCorePoolSize;
+        private int tenantUpgradeTaskExecutorMaxPoolSize;
+        private int tenantUpgradeTaskExecutorQueueCapacity;
     }
 
     @Getter

diff --git a/...g/apache/fineract/infrastructure/core/service/migration/TenantDatabaseUpgradeService.java b/...g/apache/fineract/infrastructure/core/service/migration/TenantDatabaseUpgradeService.java
@@ -21,10 +21,15 @@
 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.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 void afterPropertiesSet() throws Exception {
             }
         }
         try {
+            Scope.setScopeManager(new ThreadLocalScopeManager());
             upgradeTenantStore();
             upgradeIndividualTenants();
         } catch (LiquibaseException e) {
@@ -121,17 +128,41 @@ private void logTenantStoreDetails() {
 
     }
 
-    private void upgradeIndividualTenants() throws LiquibaseException {
+    private void upgradeIndividualTenants() {
         log.info("Upgrading all tenants");
         List<FineractPlatformTenant> tenants = tenantDetailsService.findAllTenants();
+        List<Future<String>> 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<String> 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
@@ -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/...vider/src/test/java/org/apache/fineract/infrastructure/core/LiquibaseStepDefinitions.java b/...vider/src/test/java/org/apache/fineract/infrastructure/core/LiquibaseStepDefinitions.java
@@ -104,6 +104,8 @@ public LiquibaseStepDefinitions() {
                 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
@@ -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