original patch

kjozsa · Aug 21, 2024 · fa3493e · fa3493e
1 parent 7304af5
commit fa3493e
Show file tree

Hide file tree

Showing 7 changed files with 100 additions and 5 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
@@ -279,6 +279,8 @@ public static class FineractTaskExecutor {
 
         private int defaultTaskExecutorCorePoolSize;
         private int defaultTaskExecutorMaxPoolSize;
+        private int tenantUpgradeTaskExecutorCorePoolSize;
+        private int tenantUpgradeTaskExecutorMaxPoolSize;
     }
 
     @Getter

diff --git a/...pache/fineract/infrastructure/core/service/migration/TenantUpgradeTaskExecutorConfig.java b/...pache/fineract/infrastructure/core/service/migration/TenantUpgradeTaskExecutorConfig.java
@@ -0,0 +1,42 @@
+/**
+ * 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.core.service.migration;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+@Slf4j
+@Configuration
+public class TenantUpgradeTaskExecutorConfig {
+
+    @Autowired
+    private FineractProperties fineractProperties;
+
+    @Bean("tenantUpgradeThreadPoolTaskExecutor")
+    public ThreadPoolTaskExecutor tenantUpgradeThreadPoolTaskExecutor() {
+        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
+        threadPoolTaskExecutor.setCorePoolSize(fineractProperties.getTaskExecutor().getTenantUpgradeTaskExecutorCorePoolSize());
+        threadPoolTaskExecutor.setMaxPoolSize(fineractProperties.getTaskExecutor().getTenantUpgradeTaskExecutorMaxPoolSize());
+        return threadPoolTaskExecutor;
+    }
+}
diff --git a/...ider/src/main/java/org/apache/fineract/infrastructure/core/config/TaskExecutorConfig.java b/...ider/src/main/java/org/apache/fineract/infrastructure/core/config/TaskExecutorConfig.java
@@ -47,4 +47,5 @@ public ThreadPoolTaskExecutor fineractConfigurableThreadPoolTaskExecutor() {
         threadPoolTaskExecutor.setMaxPoolSize(fineractProperties.getTaskExecutor().getDefaultTaskExecutorMaxPoolSize());
         return threadPoolTaskExecutor;
     }
+
 }
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;
 
 /**
@@ -60,6 +66,8 @@ public class TenantDatabaseUpgradeService implements InitializingBean {
     private final ExtendedSpringLiquibaseFactory liquibaseFactory;
     private final TenantDataSourceFactory tenantDataSourceFactory;
     private final Environment environment;
+    @Qualifier("tenantUpgradeThreadPoolTaskExecutor")
+    private final ThreadPoolTaskExecutor tenantUpgradeTaskExecutor;
 
     // DO NOT REMOVE! Required for liquibase custom task initialization
     private final List<CustomTaskChange> customTaskChangesForDependencyInjection;
@@ -76,6 +84,7 @@ public void afterPropertiesSet() throws Exception {
             }
         }
         try {
+            Scope.setScopeManager(new ThreadLocalScopeManager());
             upgradeTenantStore();
             upgradeIndividualTenants();
         } catch (LiquibaseException e) {
@@ -121,14 +130,28 @@ 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<>();
         if (isNotEmpty(tenants)) {
             for (FineractPlatformTenant tenant : tenants) {
-                upgradeIndividualTenant(tenant);
+                futures.add(tenantUpgradeTaskExecutor.submit(() -> {
+                    upgradeIndividualTenant(tenant);
+                    return tenant.getName();
+                }));
             }
         }
+
+        try {
+            for (Future<String> future : futures) {
+                future.get();
+            }
+        } catch (InterruptedException | ExecutionException exception) {
+            throw new RuntimeException(exception);
+        } finally {
+            tenantUpgradeTaskExecutor.shutdown();
+        }
         log.info("Tenant upgrades have finished");
     }
 

diff --git a/fineract-provider/src/main/resources/application.properties b/fineract-provider/src/main/resources/application.properties
@@ -128,6 +128,8 @@ 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.idempotency-key-header-name=${FINERACT_IDEMPOTENCY_KEY_HEADER_NAME:Idempotency-Key}
 

diff --git a/fineract-provider/src/test/java/org/apache/fineract/TestConfiguration.java b/fineract-provider/src/test/java/org/apache/fineract/TestConfiguration.java
@@ -68,6 +68,7 @@
 import org.springframework.context.annotation.Primary;
 import org.springframework.core.env.Environment;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
 
@@ -147,6 +148,14 @@ public TenantDatabaseStateVerifier tenantDatabaseStateVerifier(DatabaseIndepende
         return new TenantDatabaseStateVerifier(liquibaseProperties, databaseIndependentQueryService, databaseTypeResolver);
     }
 
+    @Bean
+    public ThreadPoolTaskExecutor tenantUpgradeThreadPoolTaskExecutor() {
+        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
+        threadPoolTaskExecutor.setCorePoolSize(1);
+        threadPoolTaskExecutor.setMaxPoolSize(1);
+        return threadPoolTaskExecutor;
+    }
+
     /**
      * Override TenantDatabaseUpgradeService binding, because the real one has a @PostConstruct upgradeAllTenants()
      * which accesses the database on start-up.
@@ -155,10 +164,11 @@ public TenantDatabaseStateVerifier tenantDatabaseStateVerifier(DatabaseIndepende
     public TenantDatabaseUpgradeService tenantDatabaseUpgradeService(TenantDetailsService tenantDetailsService,
             HikariDataSource tenantDataSource, TenantDatabaseStateVerifier tenantDatabaseStateVerifier,
             ExtendedSpringLiquibaseFactory liquibaseFactory, TenantDataSourceFactory tenantDataSourceFactory,
-            FineractProperties fineractProperties, Environment environment,
+            FineractProperties fineractProperties, Environment environment, ThreadPoolTaskExecutor tenantUpgradeThreadPoolTaskExecutor,
             List<CustomTaskChange> customTaskChangesForDependencyInjection) {
         return new TenantDatabaseUpgradeService(tenantDetailsService, tenantDataSource, fineractProperties, tenantDatabaseStateVerifier,
-                liquibaseFactory, tenantDataSourceFactory, environment, customTaskChangesForDependencyInjection);
+                liquibaseFactory, tenantDataSourceFactory, environment, tenantUpgradeThreadPoolTaskExecutor,
+                customTaskChangesForDependencyInjection);
     }
 
     /**

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
@@ -30,6 +30,8 @@
 import io.cucumber.java8.En;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
 import javax.sql.DataSource;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
@@ -41,9 +43,11 @@
 import org.apache.fineract.infrastructure.core.service.migration.TenantDatabaseUpgradeService;
 import org.apache.fineract.infrastructure.core.service.migration.TenantPasswordEncryptionTask;
 import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsService;
+import org.mockito.ArgumentCaptor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.env.Environment;
 import org.springframework.mock.env.MockEnvironment;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 public class LiquibaseStepDefinitions implements En {
 
@@ -65,6 +69,7 @@ public class LiquibaseStepDefinitions implements En {
     private SchemaUpgradeNeededException executionException;
     private DataSource defaultTenantDataSource;
     private Environment environment;
+    private ThreadPoolTaskExecutor tenantUpgradeThreadPoolTaskExecutor;
 
     public LiquibaseStepDefinitions() {
         Given("Liquibase is disabled with a default tenant", () -> {
@@ -171,6 +176,7 @@ private void initializeLiquibase(boolean liquibaseEnabled) {
         tenantStoreLiquibase = mock(ExtendedSpringLiquibase.class);
 
         defaultTenantDataSource = mock(DataSource.class);
+        tenantUpgradeThreadPoolTaskExecutor = mock(ThreadPoolTaskExecutor.class);
 
         TenantPasswordEncryptionTask tenantPasswordEncryptor = mock(TenantPasswordEncryptionTask.class);
 
@@ -185,8 +191,17 @@ private void initializeLiquibase(boolean liquibaseEnabled) {
         given(liquibaseFactory.create(defaultTenantDataSource, "tenant_db", "defaultTenant")).willReturn(tenantLiquibase);
         given(liquibaseFactory.create(defaultTenantDataSource, "tenant_db", "custom_changelog", "defaultTenant"))
                 .willReturn(customChangeLogLiquibase);
+        ArgumentCaptor<Callable> callableArgumentCaptor = ArgumentCaptor.forClass(Callable.class);
+        given(tenantUpgradeThreadPoolTaskExecutor.submit(callableArgumentCaptor.capture())).willAnswer((invocation) -> {
+            Callable callable = callableArgumentCaptor.getValue();
+            callable.call();
+            Future future = mock(Future.class);
+            given(future.get()).willReturn(null);
+            return future;
+        });
 
         tenantDatabaseUpgradeService = new TenantDatabaseUpgradeService(tenantDetailsService, tenantStoreDataSource, fineractProperties,
-                databaseStateVerifier, liquibaseFactory, tenantDataSourceFactory, environment, Arrays.asList(tenantPasswordEncryptor));
+                databaseStateVerifier, liquibaseFactory, tenantDataSourceFactory, environment, tenantUpgradeThreadPoolTaskExecutor,
+                Arrays.asList(tenantPasswordEncryptor));
     }
 }