FINERACT-2081: Support read-only transactions via Batch API

This commit is contained in:
Adam Saghy
2024-12-17 14:49:18 +01:00
parent 8716751c88
commit d03973165d
5 changed files with 125 additions and 2 deletions
@@ -56,6 +56,7 @@ import org.apache.fineract.infrastructure.core.exception.ErrorHandler;
import org.apache.fineract.infrastructure.core.filters.BatchCallHandler;
import org.apache.fineract.infrastructure.core.filters.BatchFilter;
import org.apache.fineract.infrastructure.core.filters.BatchRequestPreprocessor;
import org.apache.fineract.infrastructure.core.persistence.ExtendedJpaTransactionManager;
import org.jetbrains.annotations.NotNull;
import org.springframework.dao.ConcurrencyFailureException;
import org.springframework.dao.NonTransientDataAccessException;
@@ -141,6 +142,9 @@ public class BatchApiServiceImpl implements BatchApiService {
List<BatchResponse> responseList = new ArrayList<>();
try {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
if (transactionManager instanceof ExtendedJpaTransactionManager extendedJpaTransactionManager) {
transactionTemplate.setReadOnly(extendedJpaTransactionManager.isReadOnlyConnection());
}
transactionConfigurator.accept(transactionTemplate);
return transactionTemplate.execute(status -> {
BatchRequestContextHolder.setEnclosingTransaction(status);
@@ -240,7 +244,8 @@ public class BatchApiServiceImpl implements BatchApiService {
BatchRequestContextHolder.setRequestAttributes(new HashMap<>(Optional.ofNullable(request.getHeaders())
.map(list -> list.stream().collect(Collectors.toMap(Header::getName, Header::getValue)))
.orElse(Collections.emptyMap())));
if (BatchRequestContextHolder.isEnclosingTransaction()) {
if (BatchRequestContextHolder.isEnclosingTransaction()
&& BatchRequestContextHolder.getEnclosingTransaction().stream().anyMatch(ts -> !ts.isReadOnly())) {
entityManager.flush();
}
BatchCallHandler callHandler = new BatchCallHandler(this.batchFilters, commandStrategy::execute);
@@ -0,0 +1,49 @@
/**
* 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.persistence;
import java.util.Map;
import org.apache.fineract.infrastructure.core.service.database.DatabaseType;
import org.apache.fineract.infrastructure.core.service.database.DatabaseTypeResolver;
import org.eclipse.persistence.config.PersistenceUnitProperties;
import org.eclipse.persistence.config.TargetDatabase;
import org.springframework.orm.jpa.persistenceunit.MutablePersistenceUnitInfo;
import org.springframework.orm.jpa.persistenceunit.PersistenceUnitPostProcessor;
public class DatabaseSelectingPersistenceUnitPostProcessor implements PersistenceUnitPostProcessor {
private static final Map<DatabaseType, String> TARGET_DATABASE_MAP = Map.of(DatabaseType.MYSQL, TargetDatabase.MySQL,
DatabaseType.POSTGRESQL, TargetDatabase.PostgreSQL);
private final DatabaseTypeResolver databaseTypeResolver;
public DatabaseSelectingPersistenceUnitPostProcessor(DatabaseTypeResolver databaseTypeResolver) {
this.databaseTypeResolver = databaseTypeResolver;
}
@Override
public void postProcessPersistenceUnitInfo(MutablePersistenceUnitInfo pui) {
DatabaseType databaseType = databaseTypeResolver.databaseType();
String targetDatabase = TARGET_DATABASE_MAP.get(databaseType);
if (targetDatabase == null) {
throw new IllegalStateException("Unsupported database: " + databaseType);
}
pui.addProperty(PersistenceUnitProperties.TARGET_DATABASE, targetDatabase);
}
}
@@ -0,0 +1,101 @@
/**
* 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.persistence;
import jakarta.persistence.EntityManager;
import jakarta.persistence.FlushModeType;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import org.springframework.jdbc.datasource.JdbcTransactionObjectSupport;
import org.springframework.orm.jpa.EntityManagerHolder;
import org.springframework.orm.jpa.JpaTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.DefaultTransactionStatus;
import org.springframework.transaction.support.TransactionSynchronizationManager;
public class ExtendedJpaTransactionManager extends JpaTransactionManager {
private final List<TransactionLifecycleCallback> lifecycleCallbacks = new CopyOnWriteArrayList<>();
public ExtendedJpaTransactionManager() {
setValidateExistingTransaction(true);
}
@Override
protected void doBegin(Object transaction, TransactionDefinition definition) {
super.doBegin(transaction, definition);
if (isReadOnlyConnection() || isReadOnlyTx(transaction)) {
EntityManager entityManager = getCurrentEntityManager();
if (entityManager != null) {
entityManager.setFlushMode(FlushModeType.COMMIT);
}
}
invokeLifecycleCallbacks(TransactionLifecycleCallback::afterBegin);
}
@Override
protected void doCommit(DefaultTransactionStatus status) {
if (isReadOnlyConnection() || isReadOnlyTx(status.getTransaction())) {
EntityManager entityManager = getCurrentEntityManager();
if (entityManager != null) {
entityManager.clear();
}
}
super.doCommit(status);
invokeLifecycleCallbacks(TransactionLifecycleCallback::afterCommit);
}
public boolean isReadOnlyConnection() {
try (Connection connection = getDataSource().getConnection()) {
return connection.isReadOnly();
} catch (SQLException e) {
throw new IllegalStateException(e);
}
}
@Override
protected void doCleanupAfterCompletion(Object transaction) {
super.doCleanupAfterCompletion(transaction);
invokeLifecycleCallbacks(TransactionLifecycleCallback::afterCompletion);
}
private boolean isReadOnlyTx(Object transaction) {
JdbcTransactionObjectSupport txObject = (JdbcTransactionObjectSupport) transaction;
return txObject.isReadOnly();
}
private EntityManager getCurrentEntityManager() {
EntityManagerHolder holder = (EntityManagerHolder) TransactionSynchronizationManager.getResource(obtainEntityManagerFactory());
if (holder != null) {
return holder.getEntityManager();
}
return null;
}
private void invokeLifecycleCallbacks(Consumer<TransactionLifecycleCallback> f) {
lifecycleCallbacks.forEach(f::accept);
}
public void setLifecycleCallbacks(List<TransactionLifecycleCallback> lifecycleCallbacks) {
this.lifecycleCallbacks.addAll(lifecycleCallbacks);
}
}
@@ -0,0 +1,28 @@
/**
* 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.persistence;
public interface TransactionLifecycleCallback {
default void afterBegin() {}
default void afterCommit() {}
default void afterCompletion() {}
}
@@ -0,0 +1,118 @@
/**
* 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.batch.service;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.when;
import jakarta.persistence.EntityManager;
import jakarta.ws.rs.core.UriInfo;
import java.util.List;
import org.apache.fineract.batch.command.CommandStrategy;
import org.apache.fineract.batch.command.CommandStrategyProvider;
import org.apache.fineract.batch.domain.BatchRequest;
import org.apache.fineract.batch.domain.BatchResponse;
import org.apache.fineract.infrastructure.core.filters.BatchRequestPreprocessor;
import org.apache.fineract.infrastructure.core.persistence.ExtendedJpaTransactionManager;
import org.apache.fineract.infrastructure.core.serialization.FromJsonHelper;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.transaction.support.DefaultTransactionStatus;
@ExtendWith(MockitoExtension.class)
class BatchApiServiceImplTest {
@Mock
private CommandStrategyProvider strategyProvider;
@Mock
private ExtendedJpaTransactionManager transactionManager;
@Mock
private EntityManager entityManager;
@Mock
private CommandStrategy commandStrategy;
@Mock
private UriInfo uriInfo;
private final ResolutionHelper resolutionHelper = Mockito.spy(new ResolutionHelper(new FromJsonHelper()));
private final List<BatchRequestPreprocessor> batchPreprocessors = Mockito.spy(List.of());
@InjectMocks
private BatchApiServiceImpl batchApiService;
private BatchRequest request;
private BatchResponse response;
@BeforeEach
void setUp() {
request = new BatchRequest();
request.setRequestId(1L);
request.setMethod("POST");
request.setRelativeUrl("/random_api");
response = new BatchResponse();
response.setRequestId(1L);
response.setStatusCode(200);
response.setBody("Success");
}
@AfterEach
void tearDown() {
Mockito.reset(resolutionHelper);
Mockito.reset(batchPreprocessors);
Mockito.reset(entityManager);
Mockito.reset(commandStrategy);
Mockito.reset(strategyProvider);
Mockito.reset(transactionManager);
}
@Test
void testHandleBatchRequestsWithEnclosingTransaction() {
List<BatchRequest> requestList = List.of(request);
when(strategyProvider.getCommandStrategy(any())).thenReturn(commandStrategy);
when(commandStrategy.execute(any(), any())).thenReturn(response);
// Regular transaction
when(transactionManager.getTransaction(any()))
.thenReturn(new DefaultTransactionStatus("txn_name", null, true, true, false, false, false, null));
List<BatchResponse> result = batchApiService.handleBatchRequestsWithEnclosingTransaction(requestList, uriInfo);
assertEquals(1, result.size());
assertEquals(200, result.get(0).getStatusCode());
assertTrue(result.get(0).getBody().contains("Success"));
Mockito.verify(entityManager, times(1)).flush();
}
@Test
void testHandleBatchRequestsWithEnclosingTransactionReadOnly() {
List<BatchRequest> requestList = List.of(request);
when(strategyProvider.getCommandStrategy(any())).thenReturn(commandStrategy);
when(commandStrategy.execute(any(), any())).thenReturn(response);
// Read-only transaction
when(transactionManager.getTransaction(any()))
.thenReturn(new DefaultTransactionStatus("txn_name", null, true, true, false, true, false, null));
List<BatchResponse> result = batchApiService.handleBatchRequestsWithEnclosingTransaction(requestList, uriInfo);
assertEquals(1, result.size());
assertEquals(200, result.get(0).getStatusCode());
assertTrue(result.get(0).getBody().contains("Success"));
Mockito.verifyNoInteractions(entityManager);
}
}