FINERACT-1724: Backward compatibility fix for MariaDB on sequence vs table incrementers + SQL grammar fix for Loan COB catchup

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/service/database/DatabaseSpecificSQLGenerator.java

@@ -160,11 +160,11 @@ public class DatabaseSpecificSQLGenerator {
         }
     }
 
-    public String castBigInt(String sql) {
+    public String castInteger(String sql) {
         if (databaseTypeResolver.isMySQL()) {
-            return format("CAST(%s AS BIGINT)", sql);
+            return format("CAST(%s AS SIGNED INTEGER)", sql);
         } else if (databaseTypeResolver.isPostgreSQL()) {
-            return format("%s::BIGINT", sql);
+            return format("%s::INTEGER", sql);
         } else {
             throw new IllegalStateException("Database type is not supported for casting to bigint " + databaseTypeResolver.databaseType());
         }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/ScheduledJobRunnerConfig.java

@@ -22,6 +22,7 @@ import java.util.List;
 import org.apache.fineract.infrastructure.core.persistence.ExtendedJpaTransactionManager;
 import org.apache.fineract.infrastructure.core.persistence.TransactionLifecycleCallback;
 import org.apache.fineract.infrastructure.core.service.database.RoutingDataSource;
+import org.apache.fineract.infrastructure.jobs.config.FineractDataFieldMaxValueIncrementerFactory;
 import org.springframework.batch.core.configuration.JobRegistry;
 import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
 import org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor;
@@ -31,6 +32,7 @@ import org.springframework.batch.core.launch.support.TaskExecutorJobLauncher;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.dao.Jackson2ExecutionContextStringSerializer;
 import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
+import org.springframework.batch.item.database.support.DataFieldMaxValueIncrementerFactory;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.transaction.TransactionManagerCustomizers;
 import org.springframework.context.annotation.Bean;
@@ -56,6 +58,13 @@ public class ScheduledJobRunnerConfig {
         return new Jackson2ExecutionContextStringSerializer();
     }
 
+    @Bean
+    public DataFieldMaxValueIncrementerFactory incrementerFactory(RoutingDataSource routingDataSource) {
+        // The DefaultDataFieldMaxValueIncrementerFactory has to be overridden because Spring 6 introduced
+        // a new MariaDB incrementer that's incompatible with Spring Batch 4.x
+        return new FineractDataFieldMaxValueIncrementerFactory(routingDataSource);
+    }
+
     @Bean
     public JobRepository jobRepository(RoutingDataSource routingDataSource, PlatformTransactionManager transactionManager)
             throws Exception {
@@ -64,6 +73,7 @@ public class ScheduledJobRunnerConfig {
         factory.setTransactionManager(transactionManager);
         factory.setIsolationLevelForCreate("ISOLATION_READ_COMMITTED");
         factory.setSerializer(executionContextSerializer());
+        factory.setIncrementerFactory(incrementerFactory(routingDataSource));
         factory.afterPropertiesSet();
         return factory.getObject();
     }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/config/FineractBatchConfiguration.java

@@ -1,32 +0,0 @@
-/**
- * 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.config;
-
-import org.springframework.batch.core.configuration.support.DefaultBatchConfiguration;
-import org.springframework.batch.core.repository.ExecutionContextSerializer;
-import org.springframework.batch.core.repository.dao.Jackson2ExecutionContextStringSerializer;
-
-//@Configuration(proxyBeanMethods = false)
-public class FineractBatchConfiguration extends DefaultBatchConfiguration {
-
-    @Override
-    protected ExecutionContextSerializer getExecutionContextSerializer() {
-        return new Jackson2ExecutionContextStringSerializer();
-    }
-}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/config/FineractDataFieldMaxValueIncrementerFactory.java

@@ -0,0 +1,83 @@
+/**
+ * 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.config;
+
+import static org.springframework.batch.support.DatabaseType.MARIADB;
+import static org.springframework.batch.support.DatabaseType.MYSQL;
+import static org.springframework.batch.support.DatabaseType.POSTGRES;
+
+import java.util.List;
+import javax.sql.DataSource;
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.item.database.support.DataFieldMaxValueIncrementerFactory;
+import org.springframework.batch.support.DatabaseType;
+import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
+import org.springframework.jdbc.support.incrementer.MySQLMaxValueIncrementer;
+import org.springframework.jdbc.support.incrementer.PostgresSequenceMaxValueIncrementer;
+
+@RequiredArgsConstructor
+public class FineractDataFieldMaxValueIncrementerFactory implements DataFieldMaxValueIncrementerFactory {
+
+    private static final List<DatabaseType> SUPPORTED_DATABASE_TYPES = List.of(MARIADB, MYSQL, POSTGRES);
+
+    private final DataSource dataSource;
+
+    private String incrementerColumnName = "ID";
+
+    /**
+     * Public setter for the column name (defaults to "ID") in the incrementer. Only used by some platforms (Derby,
+     * HSQL, MySQL, SQL Server and Sybase), and should be fine for use with Spring Batch meta data as long as the
+     * default batch schema hasn't been changed.
+     *
+     * @param incrementerColumnName
+     *            the primary key column name to set
+     */
+    public void setIncrementerColumnName(String incrementerColumnName) {
+        this.incrementerColumnName = incrementerColumnName;
+    }
+
+    @Override
+    public DataFieldMaxValueIncrementer getIncrementer(String incrementerType, String incrementerName) {
+        DatabaseType databaseType = getDatabaseType(incrementerType);
+        if (databaseType == MYSQL || databaseType == MARIADB) {
+            MySQLMaxValueIncrementer mySQLMaxValueIncrementer = new MySQLMaxValueIncrementer(dataSource, incrementerName,
+                    incrementerColumnName);
+            mySQLMaxValueIncrementer.setUseNewConnection(true);
+            return mySQLMaxValueIncrementer;
+        } else if (databaseType == POSTGRES) {
+            return new PostgresSequenceMaxValueIncrementer(dataSource, incrementerName);
+        }
+        throw new IllegalArgumentException("databaseType argument was not on the approved list");
+    }
+
+    @Override
+    public boolean isSupportedIncrementerType(String incrementerType) {
+        DatabaseType databaseType = getDatabaseType(incrementerType);
+        return SUPPORTED_DATABASE_TYPES.contains(databaseType);
+    }
+
+    @Override
+    public String[] getSupportedIncrementerTypes() {
+        return SUPPORTED_DATABASE_TYPES.stream().map(DatabaseType::name).toArray(String[]::new);
+    }
+
+    private DatabaseType getDatabaseType(String incrementerType) {
+        return DatabaseType.valueOf(incrementerType.toUpperCase());
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/domain/JobExecutionRepository.java

@@ -172,7 +172,7 @@ public class JobExecutionRepository implements InitializingBean {
         sqlStatementBuilder.append(
                 "SELECT bje.JOB_EXECUTION_ID FROM BATCH_JOB_INSTANCE bji INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID INNER JOIN BATCH_JOB_EXECUTION_PARAMS bjep ON bje.JOB_EXECUTION_ID = bjep.JOB_EXECUTION_ID"
                         + " WHERE bje.STATUS IN (:statuses) AND bji.JOB_NAME = :jobName AND bjep.PARAMETER_NAME = :jobCustomParamKeyName AND "
-                        + sqlGenerator.castBigInt("bjep.PARAMETER_VALUE") + " IN (" + getSubQueryForCustomJobParameters()
+                        + sqlGenerator.castInteger("bjep.PARAMETER_VALUE") + " IN (" + getSubQueryForCustomJobParameters()
                         + ") AND bje.JOB_INSTANCE_ID NOT IN (SELECT bje.JOB_INSTANCE_ID FROM BATCH_JOB_INSTANCE bji INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID"
                         + " WHERE bje.STATUS = :completedStatus AND bji.JOB_NAME = :jobName)");
         return namedParameterJdbcTemplate.queryForList(