FINERACT-2304: Create Configuration assembler and move resilience4j props into fineract namespace implement proper retry for SynchronousCommandProcessingService and BatchApiServiceImpl
This commit is contained in:
+29
-18
@@ -25,6 +25,7 @@ import com.google.gson.Gson;
|
||||
import com.jayway.jsonpath.JsonPathException;
|
||||
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
|
||||
import io.github.resilience4j.core.functions.Either;
|
||||
import io.github.resilience4j.retry.Retry;
|
||||
import jakarta.persistence.EntityManager;
|
||||
import jakarta.persistence.PersistenceContext;
|
||||
import jakarta.ws.rs.core.UriInfo;
|
||||
@@ -51,17 +52,18 @@ import org.apache.fineract.batch.domain.Header;
|
||||
import org.apache.fineract.batch.exception.BatchReferenceInvalidException;
|
||||
import org.apache.fineract.batch.exception.ErrorInfo;
|
||||
import org.apache.fineract.batch.service.ResolutionHelper.BatchRequestNode;
|
||||
import org.apache.fineract.commands.configuration.RetryConfigurationAssembler;
|
||||
import org.apache.fineract.infrastructure.core.domain.BatchRequestContextHolder;
|
||||
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.springframework.dao.ConcurrencyFailureException;
|
||||
import org.springframework.dao.NonTransientDataAccessException;
|
||||
import org.springframework.lang.NonNull;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.TransactionException;
|
||||
import org.springframework.transaction.TransactionExecution;
|
||||
import org.springframework.transaction.TransactionSystemException;
|
||||
@@ -90,6 +92,8 @@ public class BatchApiServiceImpl implements BatchApiService {
|
||||
|
||||
private final List<BatchRequestPreprocessor> batchPreprocessors;
|
||||
|
||||
private final RetryConfigurationAssembler retryConfigurationAssembler;
|
||||
|
||||
private EntityManager entityManager;
|
||||
|
||||
/**
|
||||
@@ -138,28 +142,35 @@ public class BatchApiServiceImpl implements BatchApiService {
|
||||
*/
|
||||
private List<BatchResponse> callInTransaction(Consumer<TransactionTemplate> transactionConfigurator,
|
||||
Supplier<List<BatchResponse>> request) {
|
||||
// Retry logic for enclosingTransaction=true and when the isolation level is REPEATABLE_READ or stricter we need
|
||||
// to restart the transaction as well!
|
||||
|
||||
Retry retry = retryConfigurationAssembler.getRetryConfigurationForBatchApiWithEnclosingTransaction();
|
||||
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);
|
||||
try {
|
||||
Supplier<List<BatchResponse>> batchSupplier = () -> {
|
||||
responseList.clear();
|
||||
try {
|
||||
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
|
||||
transactionTemplate.setIsolationLevel(TransactionDefinition.ISOLATION_REPEATABLE_READ);
|
||||
transactionConfigurator.accept(transactionTemplate);
|
||||
return transactionTemplate.execute(status -> {
|
||||
BatchRequestContextHolder.setEnclosingTransaction(status);
|
||||
responseList.addAll(request.get());
|
||||
return responseList;
|
||||
} catch (BatchExecutionException ex) {
|
||||
log.error("Exception during the batch request processing", ex);
|
||||
responseList.add(buildErrorResponse(ex.getCause(), ex.getRequest()));
|
||||
return responseList;
|
||||
} finally {
|
||||
BatchRequestContextHolder.resetTransaction();
|
||||
}
|
||||
});
|
||||
});
|
||||
} finally {
|
||||
BatchRequestContextHolder.resetTransaction();
|
||||
}
|
||||
};
|
||||
Supplier<List<BatchResponse>> retryingBatch = Retry.decorateSupplier(retry, batchSupplier);
|
||||
try {
|
||||
return retryingBatch.get();
|
||||
} catch (TransactionException | NonTransientDataAccessException ex) {
|
||||
return buildErrorResponses(ex, responseList);
|
||||
} catch (BatchExecutionException ex) {
|
||||
log.error("Exception during the batch request processing", ex);
|
||||
responseList.add(buildErrorResponse(ex.getCause(), ex.getRequest()));
|
||||
return responseList;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+94
@@ -0,0 +1,94 @@
|
||||
/**
|
||||
* 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.commands.configuration;
|
||||
|
||||
import io.github.resilience4j.core.IntervalFunction;
|
||||
import io.github.resilience4j.retry.Retry;
|
||||
import io.github.resilience4j.retry.RetryConfig;
|
||||
import io.github.resilience4j.retry.RetryRegistry;
|
||||
import java.util.Arrays;
|
||||
import lombok.AllArgsConstructor;
|
||||
import org.apache.fineract.batch.service.BatchExecutionException;
|
||||
import org.apache.fineract.infrastructure.core.config.FineractProperties;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
@AllArgsConstructor
|
||||
@Service
|
||||
public class RetryConfigurationAssembler {
|
||||
|
||||
public static final String EXECUTE_COMMAND = "executeCommand";
|
||||
public static final String BATCH_RETRY = "batchRetry";
|
||||
private final RetryRegistry registry;
|
||||
private final FineractProperties fineractProperties;
|
||||
|
||||
private boolean isAssignableFrom(Object e, Class<? extends Throwable>[] exceptionList) {
|
||||
return Arrays.stream(exceptionList).anyMatch(re -> re.isAssignableFrom(e.getClass()));
|
||||
}
|
||||
|
||||
public Retry getRetryConfigurationForExecuteCommand() {
|
||||
Class<? extends Throwable>[] exceptionList = fineractProperties.getRetry().getInstances().getExecuteCommand().getRetryExceptions();
|
||||
RetryConfig.Builder configBuilder = buildCommonExecuteCommandConfiguration();
|
||||
|
||||
if (exceptionList != null) {
|
||||
configBuilder.retryOnException(e -> isAssignableFrom(e, exceptionList));
|
||||
}
|
||||
|
||||
RetryConfig config = configBuilder.build();
|
||||
return registry.retry(EXECUTE_COMMAND, config);
|
||||
}
|
||||
|
||||
public Retry getRetryConfigurationForBatchApiWithEnclosingTransaction() {
|
||||
Class<? extends Throwable>[] exceptionList = fineractProperties.getRetry().getInstances().getExecuteCommand().getRetryExceptions();
|
||||
RetryConfig.Builder configBuilder = buildCommonExecuteCommandConfiguration();
|
||||
|
||||
if (exceptionList != null) {
|
||||
configBuilder.retryOnException(ex -> {
|
||||
if (ex instanceof BatchExecutionException e) {
|
||||
return isAssignableFrom(e.getCause(), exceptionList);
|
||||
} else {
|
||||
return isAssignableFrom(ex, exceptionList);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
RetryConfig config = configBuilder.build();
|
||||
return registry.retry(BATCH_RETRY, config);
|
||||
}
|
||||
|
||||
private RetryConfig.Builder buildCommonExecuteCommandConfiguration() {
|
||||
var props = fineractProperties.getRetry().getInstances().getExecuteCommand();
|
||||
|
||||
RetryConfig.Builder configBuilder = RetryConfig.custom().maxAttempts(props.getMaxAttempts());
|
||||
|
||||
if (props.getWaitDuration() != null && props.getWaitDuration().toMillis() >= 0) {
|
||||
if (Boolean.TRUE.equals(props.getEnableExponentialBackoff())) {
|
||||
Double multiplier = props.getExponentialBackoffMultiplier();
|
||||
if (multiplier != null) {
|
||||
configBuilder.intervalFunction(IntervalFunction.ofExponentialBackoff(props.getWaitDuration(), multiplier));
|
||||
} else {
|
||||
configBuilder.intervalFunction(IntervalFunction.ofExponentialBackoff(props.getWaitDuration()));
|
||||
}
|
||||
} else {
|
||||
configBuilder.waitDuration(props.getWaitDuration());
|
||||
}
|
||||
}
|
||||
|
||||
return configBuilder;
|
||||
}
|
||||
}
|
||||
+80
-65
@@ -24,15 +24,16 @@ import static org.apache.http.HttpStatus.SC_OK;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
import com.google.gson.reflect.TypeToken;
|
||||
import io.github.resilience4j.retry.annotation.Retry;
|
||||
import java.lang.reflect.Type;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.function.Supplier;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.fineract.batch.exception.ErrorInfo;
|
||||
import org.apache.fineract.commands.configuration.RetryConfigurationAssembler;
|
||||
import org.apache.fineract.commands.domain.CommandProcessingResultType;
|
||||
import org.apache.fineract.commands.domain.CommandSource;
|
||||
import org.apache.fineract.commands.domain.CommandWrapper;
|
||||
@@ -76,81 +77,96 @@ public class SynchronousCommandProcessingService implements CommandProcessingSer
|
||||
private final CommandHandlerProvider commandHandlerProvider;
|
||||
private final IdempotencyKeyResolver idempotencyKeyResolver;
|
||||
private final CommandSourceService commandSourceService;
|
||||
private final RetryConfigurationAssembler retryConfigurationAssembler;
|
||||
|
||||
private final FineractRequestContextHolder fineractRequestContextHolder;
|
||||
private final Gson gson = GoogleGsonSerializerHelper.createSimpleGson();
|
||||
|
||||
private CommandProcessingResult retryWrapper(Supplier<CommandProcessingResult> supplier) {
|
||||
if (!BatchRequestContextHolder.isEnclosingTransaction()) {
|
||||
try {
|
||||
return retryConfigurationAssembler.getRetryConfigurationForExecuteCommand().executeSupplier(supplier);
|
||||
} catch (RuntimeException e) {
|
||||
fallbackExecuteCommand(e);
|
||||
}
|
||||
}
|
||||
return supplier.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
@Retry(name = "executeCommand", fallbackMethod = "fallbackExecuteCommand")
|
||||
public CommandProcessingResult executeCommand(final CommandWrapper wrapper, final JsonCommand command,
|
||||
final boolean isApprovedByChecker) {
|
||||
// Do not store the idempotency key because of the exception handling
|
||||
setIdempotencyKeyStoreFlag(false);
|
||||
return retryWrapper(() -> {
|
||||
// Do not store the idempotency key because of the exception handling
|
||||
setIdempotencyKeyStoreFlag(false);
|
||||
|
||||
Long commandId = (Long) fineractRequestContextHolder.getAttribute(COMMAND_SOURCE_ID, null);
|
||||
boolean isRetry = commandId != null;
|
||||
boolean isEnclosingTransaction = BatchRequestContextHolder.isEnclosingTransaction();
|
||||
Long commandId = (Long) fineractRequestContextHolder.getAttribute(COMMAND_SOURCE_ID, null);
|
||||
boolean isRetry = commandId != null;
|
||||
boolean isEnclosingTransaction = BatchRequestContextHolder.isEnclosingTransaction();
|
||||
|
||||
CommandSource commandSource = null;
|
||||
String idempotencyKey;
|
||||
if (isRetry) {
|
||||
commandSource = commandSourceService.getCommandSource(commandId);
|
||||
idempotencyKey = commandSource.getIdempotencyKey();
|
||||
} else if ((commandId = command.commandId()) != null) { // action on the command itself
|
||||
commandSource = commandSourceService.getCommandSource(commandId);
|
||||
idempotencyKey = commandSource.getIdempotencyKey();
|
||||
} else {
|
||||
idempotencyKey = idempotencyKeyResolver.resolve(wrapper);
|
||||
}
|
||||
exceptionWhenTheRequestAlreadyProcessed(wrapper, idempotencyKey, isRetry);
|
||||
|
||||
AppUser user = context.authenticatedUser(wrapper);
|
||||
if (commandSource == null) {
|
||||
if (isEnclosingTransaction) {
|
||||
commandSource = commandSourceService.getInitialCommandSource(wrapper, command, user, idempotencyKey);
|
||||
CommandSource commandSource = null;
|
||||
String idempotencyKey;
|
||||
if (isRetry) {
|
||||
commandSource = commandSourceService.getCommandSource(commandId);
|
||||
idempotencyKey = commandSource.getIdempotencyKey();
|
||||
} else if ((commandId = command.commandId()) != null) { // action on the command itself
|
||||
commandSource = commandSourceService.getCommandSource(commandId);
|
||||
idempotencyKey = commandSource.getIdempotencyKey();
|
||||
} else {
|
||||
commandSource = commandSourceService.saveInitialNewTransaction(wrapper, command, user, idempotencyKey);
|
||||
commandId = commandSource.getId();
|
||||
idempotencyKey = idempotencyKeyResolver.resolve(wrapper);
|
||||
}
|
||||
}
|
||||
if (commandId != null) {
|
||||
exceptionWhenTheRequestAlreadyProcessed(wrapper, idempotencyKey, isRetry);
|
||||
|
||||
AppUser user = context.authenticatedUser(wrapper);
|
||||
if (commandSource == null) {
|
||||
if (isEnclosingTransaction) {
|
||||
commandSource = commandSourceService.getInitialCommandSource(wrapper, command, user, idempotencyKey);
|
||||
} else {
|
||||
commandSource = commandSourceService.saveInitialNewTransaction(wrapper, command, user, idempotencyKey);
|
||||
commandId = commandSource.getId();
|
||||
}
|
||||
}
|
||||
if (commandId != null) {
|
||||
storeCommandIdInContext(commandSource); // Store command id as a request attribute
|
||||
}
|
||||
|
||||
setIdempotencyKeyStoreFlag(true);
|
||||
|
||||
final CommandProcessingResult result;
|
||||
try {
|
||||
result = commandSourceService.processCommand(findCommandHandler(wrapper), command, commandSource, user,
|
||||
isApprovedByChecker);
|
||||
} catch (Throwable t) { // NOSONAR
|
||||
RuntimeException mappable = ErrorHandler.getMappable(t);
|
||||
ErrorInfo errorInfo = commandSourceService.generateErrorInfo(mappable);
|
||||
Integer statusCode = errorInfo.getStatusCode();
|
||||
commandSource.setResultStatusCode(statusCode);
|
||||
commandSource.setResult(errorInfo.getMessage());
|
||||
if (statusCode != SC_OK) {
|
||||
commandSource.setStatus(ERROR);
|
||||
}
|
||||
if (!isEnclosingTransaction) { // TODO: temporary solution
|
||||
commandSource = commandSourceService.saveResultNewTransaction(commandSource);
|
||||
}
|
||||
// must not throw any exception; must persist in new transaction as the current transaction was already
|
||||
// marked as rollback
|
||||
publishHookErrorEvent(wrapper, command, errorInfo);
|
||||
throw mappable;
|
||||
}
|
||||
|
||||
commandSource.setResultStatusCode(SC_OK);
|
||||
commandSource.updateForAudit(result);
|
||||
commandSource.setResult(toApiResultJsonSerializer.serializeResult(result));
|
||||
commandSource.setStatus(PROCESSED);
|
||||
commandSource = commandSourceService.saveResultSameTransaction(commandSource);
|
||||
storeCommandIdInContext(commandSource); // Store command id as a request attribute
|
||||
}
|
||||
|
||||
setIdempotencyKeyStoreFlag(true);
|
||||
|
||||
final CommandProcessingResult result;
|
||||
try {
|
||||
result = commandSourceService.processCommand(findCommandHandler(wrapper), command, commandSource, user, isApprovedByChecker);
|
||||
} catch (Throwable t) { // NOSONAR
|
||||
RuntimeException mappable = ErrorHandler.getMappable(t);
|
||||
ErrorInfo errorInfo = commandSourceService.generateErrorInfo(mappable);
|
||||
Integer statusCode = errorInfo.getStatusCode();
|
||||
commandSource.setResultStatusCode(statusCode);
|
||||
commandSource.setResult(errorInfo.getMessage());
|
||||
if (statusCode != SC_OK) {
|
||||
commandSource.setStatus(ERROR);
|
||||
}
|
||||
if (!isEnclosingTransaction) { // TODO: temporary solution
|
||||
commandSource = commandSourceService.saveResultNewTransaction(commandSource);
|
||||
}
|
||||
// must not throw any exception; must persist in new transaction as the current transaction was already
|
||||
// marked as rollback
|
||||
publishHookErrorEvent(wrapper, command, errorInfo);
|
||||
throw mappable;
|
||||
}
|
||||
|
||||
commandSource.setResultStatusCode(SC_OK);
|
||||
commandSource.updateForAudit(result);
|
||||
commandSource.setResult(toApiResultJsonSerializer.serializeResult(result));
|
||||
commandSource.setStatus(PROCESSED);
|
||||
commandSource = commandSourceService.saveResultSameTransaction(commandSource);
|
||||
storeCommandIdInContext(commandSource); // Store command id as a request attribute
|
||||
|
||||
result.setRollbackTransaction(null);
|
||||
publishHookEvent(wrapper.entityName(), wrapper.actionName(), command, result); // TODO must be performed in a
|
||||
// new transaction
|
||||
return result;
|
||||
result.setRollbackTransaction(null);
|
||||
publishHookEvent(wrapper.entityName(), wrapper.actionName(), command, result); // TODO must be performed in
|
||||
// a
|
||||
// new transaction
|
||||
return result;
|
||||
});
|
||||
}
|
||||
|
||||
private void storeCommandIdInContext(CommandSource savedCommandSource) {
|
||||
@@ -188,8 +204,7 @@ public class SynchronousCommandProcessingService implements CommandProcessingSer
|
||||
fineractRequestContextHolder.setAttribute(IDEMPOTENCY_KEY_STORE_FLAG, flag);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public CommandProcessingResult fallbackExecuteCommand(Exception e) {
|
||||
public void fallbackExecuteCommand(Exception e) {
|
||||
throw ErrorHandler.getMappable(e);
|
||||
}
|
||||
|
||||
|
||||
+28
@@ -84,6 +84,8 @@ public class FineractProperties {
|
||||
|
||||
private FineractCache cache;
|
||||
|
||||
private RetryProperties retry;
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
public static class FineractTenantProperties {
|
||||
@@ -612,4 +614,30 @@ public class FineractProperties {
|
||||
private Duration ttl;
|
||||
private Integer maximumEntries;
|
||||
}
|
||||
|
||||
@Setter
|
||||
@Getter
|
||||
public static class RetryProperties {
|
||||
|
||||
private InstancesProperties instances;
|
||||
|
||||
@Setter
|
||||
@Getter
|
||||
public static class InstancesProperties {
|
||||
|
||||
private ExecuteCommandProperties executeCommand;
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
public static class ExecuteCommandProperties {
|
||||
|
||||
private Class<? extends Throwable>[] retryExceptions;
|
||||
private Integer maxAttempts;
|
||||
private Boolean enableExponentialBackoff;
|
||||
private Double exponentialBackoffMultiplier;
|
||||
private Duration waitDuration;
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+108
@@ -21,16 +21,28 @@ 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.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import io.github.resilience4j.retry.Retry;
|
||||
import io.github.resilience4j.retry.RetryConfig;
|
||||
import io.github.resilience4j.retry.RetryRegistry;
|
||||
import jakarta.persistence.EntityManager;
|
||||
import jakarta.ws.rs.core.UriInfo;
|
||||
import java.time.Duration;
|
||||
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.batch.exception.ErrorInfo;
|
||||
import org.apache.fineract.commands.configuration.RetryConfigurationAssembler;
|
||||
import org.apache.fineract.infrastructure.core.config.FineractProperties;
|
||||
import org.apache.fineract.infrastructure.core.exception.ErrorHandler;
|
||||
import org.apache.fineract.infrastructure.core.filters.BatchRequestPreprocessor;
|
||||
import org.apache.fineract.infrastructure.core.persistence.ExtendedJpaTransactionManager;
|
||||
import org.apache.fineract.infrastructure.core.serialization.FromJsonHelper;
|
||||
@@ -57,8 +69,21 @@ class BatchApiServiceImplTest {
|
||||
private CommandStrategy commandStrategy;
|
||||
@Mock
|
||||
private UriInfo uriInfo;
|
||||
@Mock
|
||||
private ErrorHandler errorHandler;
|
||||
|
||||
@Mock
|
||||
private RetryRegistry registry;
|
||||
|
||||
@Mock
|
||||
private FineractProperties fineractProperties;
|
||||
|
||||
@InjectMocks
|
||||
private RetryConfigurationAssembler retryConfigurationAssembler;
|
||||
|
||||
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;
|
||||
@@ -66,6 +91,8 @@ class BatchApiServiceImplTest {
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
batchApiService = new BatchApiServiceImpl(strategyProvider, resolutionHelper, transactionManager, errorHandler, List.of(),
|
||||
batchPreprocessors, retryConfigurationAssembler);
|
||||
batchApiService.setEntityManager(entityManager);
|
||||
request = new BatchRequest();
|
||||
request.setRequestId(1L);
|
||||
@@ -75,6 +102,16 @@ class BatchApiServiceImplTest {
|
||||
response.setRequestId(1L);
|
||||
response.setStatusCode(200);
|
||||
response.setBody("Success");
|
||||
FineractProperties.RetryProperties settings = new FineractProperties.RetryProperties();
|
||||
settings.setInstances(new FineractProperties.RetryProperties.InstancesProperties());
|
||||
settings.getInstances().setExecuteCommand(new FineractProperties.RetryProperties.InstancesProperties.ExecuteCommandProperties());
|
||||
settings.getInstances().getExecuteCommand().setMaxAttempts(3);
|
||||
settings.getInstances().getExecuteCommand().setWaitDuration(Duration.ofMillis(2));
|
||||
settings.getInstances().getExecuteCommand().setEnableExponentialBackoff(false);
|
||||
settings.getInstances().getExecuteCommand().setRetryExceptions(new Class[] { RetryException.class });
|
||||
when(fineractProperties.getRetry()).thenReturn(settings);
|
||||
when(registry.retry(anyString(), any(RetryConfig.class)))
|
||||
.thenAnswer(i -> Retry.of((String) i.getArgument(0), (RetryConfig) i.getArgument(1)));
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
@@ -87,6 +124,74 @@ class BatchApiServiceImplTest {
|
||||
Mockito.reset(transactionManager);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testHandleBatchRequestsWithEnclosingTransactionResult200WithRetryOnTransactionFailure() {
|
||||
|
||||
List<BatchRequest> requestList = List.of(request);
|
||||
when(strategyProvider.getCommandStrategy(any())).thenReturn(commandStrategy);
|
||||
when(commandStrategy.execute(any(), any())).thenReturn(response).thenReturn(response);
|
||||
// throw exception at transaction commit to verify If OptimisticLockException or similar exceptions are on the
|
||||
// retry list, they can perform a retry.
|
||||
// do nothing on 2nd hit to simulate success commit
|
||||
doThrow(new RetryException()).doNothing().when(transactionManager).commit(any());
|
||||
|
||||
// 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.getFirst().getStatusCode());
|
||||
assertTrue(result.getFirst().getBody().contains("Success"));
|
||||
|
||||
verify(transactionManager, times(2)).commit(any());
|
||||
verify(entityManager, times(2)).flush();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testHandleBatchRequestsWithEnclosingTransactionResult200WithRetry() {
|
||||
|
||||
ErrorInfo errorInfo = mock(ErrorInfo.class);
|
||||
when(errorInfo.getMessage()).thenReturn("Failed");
|
||||
when(errorInfo.getStatusCode()).thenReturn(500);
|
||||
when(errorHandler.handle(any())).thenReturn(errorInfo);
|
||||
|
||||
List<BatchRequest> requestList = List.of(request);
|
||||
when(strategyProvider.getCommandStrategy(any())).thenReturn(commandStrategy);
|
||||
when(commandStrategy.execute(any(), any())).thenThrow(new RetryException()).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.getFirst().getStatusCode());
|
||||
assertTrue(result.getFirst().getBody().contains("Success"));
|
||||
Mockito.verify(entityManager, times(2)).flush();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testHandleBatchRequestsWithEnclosingTransactionFailsWithRetry() {
|
||||
|
||||
List<BatchRequest> requestList = List.of(request);
|
||||
when(strategyProvider.getCommandStrategy(any())).thenReturn(commandStrategy);
|
||||
when(commandStrategy.execute(any(), any())).thenThrow(new RetryException()).thenThrow(new RetryException())
|
||||
.thenThrow(new RetryException());
|
||||
|
||||
ErrorInfo errorInfo = mock(ErrorInfo.class);
|
||||
when(errorInfo.getMessage()).thenReturn("Failed");
|
||||
when(errorInfo.getStatusCode()).thenReturn(500);
|
||||
when(errorHandler.handle(any())).thenReturn(errorInfo);
|
||||
|
||||
// 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(500, result.getFirst().getStatusCode());
|
||||
assertTrue(result.getFirst().getBody().contains("Failed"));
|
||||
Mockito.verify(entityManager, times(3)).flush();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testHandleBatchRequestsWithEnclosingTransaction() {
|
||||
List<BatchRequest> requestList = List.of(request);
|
||||
@@ -116,4 +221,7 @@ class BatchApiServiceImplTest {
|
||||
assertTrue(result.get(0).getBody().contains("Success"));
|
||||
Mockito.verifyNoInteractions(entityManager);
|
||||
}
|
||||
|
||||
private static final class RetryException extends RuntimeException {}
|
||||
|
||||
}
|
||||
|
||||
@@ -6,23 +6,28 @@ For a deeper understanding of resilience4j, refer to the https://resilience4j.re
|
||||
|===
|
||||
|Name |Env Variable |Default Value |Description
|
||||
|
||||
|resilience4j.retry.instances.executeCommand.max-attempts
|
||||
|fineract.retry.instances.executeCommand.max-attempts
|
||||
|FINERACT_COMMAND_PROCESSING_RETRY_MAX_ATTEMPTS
|
||||
|3
|
||||
|The number of attempts that resilience4j will attempt to execute a command after a failed execution. Refer to org. apache. fineract. commands. service. SynchronousCommandProcessingService#executeCommand for more details
|
||||
|
||||
|resilience4j.retry.instances.executeCommand.wait-duration
|
||||
|fineract.retry.instances.executeCommand.wait-duration
|
||||
|FINERACT_COMMAND_PROCESSING_RETRY_WAIT_DURATION
|
||||
|1s
|
||||
|The fixed time value that the retry instance will wait before the next attempt can be made to execute a command
|
||||
|
||||
|resilience4j.retry.instances.executeCommand.enable-exponential-backoff
|
||||
|fineract.retry.instances.executeCommand.enable-exponential-backoff
|
||||
|FINERACT_COMMAND_PROCESSING_RETRY_ENABLE_EXPONENTIAL_BACKOFF
|
||||
|true
|
||||
|If set to true, the wait-duration will increase exponentially between each retry to execute a command
|
||||
|
||||
|resilience4j.retry.instances.executeCommand.retryExceptions
|
||||
|fineract.retry.instances.executeCommand.exponential-backoff-multiplier
|
||||
|FINERACT_COMMAND_PROCESSING_RETRY_EXPONENTIAL_BACKOFF_MULTIPLIER
|
||||
|3
|
||||
|The multiplier for exponential backoff, this is useful only when enable-exponential-backoff is set to true
|
||||
|
||||
|fineract.retry.instances.executeCommand.retryExceptions
|
||||
|FINERACT_COMMAND_PROCESSING_RETRY_EXCEPTIONS
|
||||
|org.springframework.dao.ConcurrencyFailureException,org.eclipse.persistence.exceptions.OptimisticLockException,jakarta.persistence.OptimisticLockException,org.springframework.orm.jpa.JpaOptimisticLockingFailureException,org.apache.fineract.infrastructure.core.exception.IdempotentCommandProcessUnderProcessingException
|
||||
|This property specifies the list of exceptions that the execute command retry instance will retry on
|
||||
|
||||
|
||||
@@ -441,11 +441,11 @@ spring.batch.initialize-schema=NEVER
|
||||
# Disabling Spring Batch jobs on startup
|
||||
spring.batch.job.enabled=false
|
||||
|
||||
resilience4j.retry.instances.executeCommand.max-attempts=${FINERACT_COMMAND_PROCESSING_RETRY_MAX_ATTEMPTS:3}
|
||||
resilience4j.retry.instances.executeCommand.wait-duration=${FINERACT_COMMAND_PROCESSING_RETRY_WAIT_DURATION:1s}
|
||||
resilience4j.retry.instances.executeCommand.enable-exponential-backoff=${FINERACT_COMMAND_PROCESSING_RETRY_ENABLE_EXPONENTIAL_BACKOFF:true}
|
||||
resilience4j.retry.instances.executeCommand.exponential-backoff-multiplier=${FINERACT_COMMAND_PROCESSING_RETRY_EXPONENTIAL_BACKOFF_MULTIPLIER:2}
|
||||
resilience4j.retry.instances.executeCommand.retryExceptions=${FINERACT_COMMAND_PROCESSING_RETRY_EXCEPTIONS:org.springframework.dao.ConcurrencyFailureException,org.eclipse.persistence.exceptions.OptimisticLockException,jakarta.persistence.OptimisticLockException,org.springframework.orm.jpa.JpaOptimisticLockingFailureException,org.apache.fineract.infrastructure.core.exception.IdempotentCommandProcessUnderProcessingException}
|
||||
fineract.retry.instances.executeCommand.max-attempts=${FINERACT_COMMAND_PROCESSING_RETRY_MAX_ATTEMPTS:3}
|
||||
fineract.retry.instances.executeCommand.wait-duration=${FINERACT_COMMAND_PROCESSING_RETRY_WAIT_DURATION:1s}
|
||||
fineract.retry.instances.executeCommand.enable-exponential-backoff=${FINERACT_COMMAND_PROCESSING_RETRY_ENABLE_EXPONENTIAL_BACKOFF:true}
|
||||
fineract.retry.instances.executeCommand.exponential-backoff-multiplier=${FINERACT_COMMAND_PROCESSING_RETRY_EXPONENTIAL_BACKOFF_MULTIPLIER:2}
|
||||
fineract.retry.instances.executeCommand.retryExceptions=${FINERACT_COMMAND_PROCESSING_RETRY_EXCEPTIONS:org.springframework.dao.ConcurrencyFailureException,org.eclipse.persistence.exceptions.OptimisticLockException,jakarta.persistence.OptimisticLockException,org.springframework.orm.jpa.JpaOptimisticLockingFailureException,org.apache.fineract.infrastructure.core.exception.IdempotentCommandProcessUnderProcessingException}
|
||||
|
||||
resilience4j.retry.instances.processJobDetailForExecution.max-attempts=${FINERACT_PROCESS_JOB_DETAIL_RETRY_MAX_ATTEMPTS:3}
|
||||
resilience4j.retry.instances.processJobDetailForExecution.wait-duration=${FINERACT_PROCESS_JOB_DETAIL_RETRY_WAIT_DURATION:1s}
|
||||
|
||||
+9
-3
@@ -23,10 +23,12 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
|
||||
import io.cucumber.java8.En;
|
||||
import io.github.resilience4j.retry.Retry;
|
||||
import io.github.resilience4j.retry.RetryRegistry;
|
||||
import io.github.resilience4j.retry.event.RetryEvent;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import org.apache.fineract.commands.configuration.RetryConfigurationAssembler;
|
||||
import org.apache.fineract.commands.domain.CommandWrapper;
|
||||
import org.apache.fineract.commands.exception.RollbackTransactionNotApprovedException;
|
||||
import org.apache.fineract.infrastructure.core.api.JsonCommand;
|
||||
@@ -50,6 +52,9 @@ public class CommandServiceStepDefinitions implements En {
|
||||
@Autowired
|
||||
private RetryRegistry retryRegistry;
|
||||
|
||||
@Autowired
|
||||
private RetryConfigurationAssembler retryConfigurationAssembler;
|
||||
|
||||
private PortfolioCommandSourceWritePlatformService commandSourceWritePlatformService;
|
||||
|
||||
private DummyCommand command;
|
||||
@@ -67,8 +72,9 @@ public class CommandServiceStepDefinitions implements En {
|
||||
Mockito.when(contextHolder.getAttribute(any(), any())).thenThrow(new CannotAcquireLockException("BLOW IT UP!!!"))
|
||||
.thenThrow(new ObjectOptimisticLockingFailureException("Dummy", new RuntimeException("BLOW IT UP!!!")))
|
||||
.thenThrow(new RollbackTransactionNotApprovedException(1L, null));
|
||||
|
||||
this.retryRegistry.retry("executeCommand").getEventPublisher().onRetry(event -> {
|
||||
Retry retry1 = retryConfigurationAssembler.getRetryConfigurationForExecuteCommand();
|
||||
assertNotNull(retry1);
|
||||
retry1.getEventPublisher().onRetry(event -> {
|
||||
log.warn("... retry event: {}", event);
|
||||
|
||||
counter.incrementAndGet();
|
||||
@@ -92,7 +98,7 @@ public class CommandServiceStepDefinitions implements En {
|
||||
assertEquals(2, retryEvent.getNumberOfRetryAttempts());
|
||||
});
|
||||
|
||||
Then("/^The command processing service execute function should be called 3 times$/", () -> {
|
||||
Then("/^The command processing service execute function should be called 2 times$/", () -> {
|
||||
assertEquals(2, counter.get());
|
||||
});
|
||||
}
|
||||
|
||||
+81
@@ -20,12 +20,21 @@ package org.apache.fineract.commands.service;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.mockito.Mockito.any;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.reset;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
|
||||
import io.github.resilience4j.retry.Retry;
|
||||
import io.github.resilience4j.retry.RetryConfig;
|
||||
import jakarta.servlet.http.HttpServletRequest;
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
import org.apache.fineract.batch.exception.ErrorInfo;
|
||||
import org.apache.fineract.commands.configuration.RetryConfigurationAssembler;
|
||||
import org.apache.fineract.commands.domain.CommandProcessingResultType;
|
||||
import org.apache.fineract.commands.domain.CommandSource;
|
||||
import org.apache.fineract.commands.domain.CommandWrapper;
|
||||
@@ -39,6 +48,7 @@ import org.apache.fineract.infrastructure.core.exception.PlatformApiDataValidati
|
||||
import org.apache.fineract.infrastructure.core.serialization.ToApiJsonSerializer;
|
||||
import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
|
||||
import org.apache.fineract.useradministration.domain.AppUser;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.InjectMocks;
|
||||
@@ -70,6 +80,9 @@ public class SynchronousCommandProcessingServiceTest {
|
||||
@Mock
|
||||
private CommandSourceService commandSourceService;
|
||||
|
||||
@Mock
|
||||
private RetryConfigurationAssembler retryConfigurationAssembler;
|
||||
|
||||
@Spy
|
||||
private FineractRequestContextHolder fineractRequestContextHolder;
|
||||
|
||||
@@ -83,6 +96,71 @@ public class SynchronousCommandProcessingServiceTest {
|
||||
public void setup() {
|
||||
MockitoAnnotations.openMocks(this);
|
||||
RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
|
||||
when(retryConfigurationAssembler.getRetryConfigurationForExecuteCommand())
|
||||
.thenReturn(Retry.of("IDENTIFIER", RetryConfig.custom().maxAttempts(3).failAfterMaxAttempts(true)
|
||||
.waitDuration(Duration.ofMillis(1)).retryOnException(e -> e instanceof RetryException).build()));
|
||||
ErrorInfo errorInfo = mock(ErrorInfo.class);
|
||||
when(errorInfo.getMessage()).thenReturn("Failed");
|
||||
when(errorInfo.getStatusCode()).thenReturn(500);
|
||||
when(commandSourceService.generateErrorInfo(any())).thenReturn(errorInfo);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
public void teardown() {
|
||||
reset(context);
|
||||
reset(applicationContext);
|
||||
reset(toApiJsonSerializer);
|
||||
reset(toApiResultJsonSerializer);
|
||||
reset(configurationDomainService);
|
||||
reset(commandHandlerProvider);
|
||||
reset(idempotencyKeyResolver);
|
||||
reset(commandSourceService);
|
||||
reset(retryConfigurationAssembler);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testExecuteCommandSuccessAfter2Fails() {
|
||||
CommandWrapper commandWrapper = Mockito.mock(CommandWrapper.class);
|
||||
when(commandWrapper.isDatatableResource()).thenReturn(false);
|
||||
when(commandWrapper.isNoteResource()).thenReturn(false);
|
||||
when(commandWrapper.isSurveyResource()).thenReturn(false);
|
||||
when(commandWrapper.isLoanDisburseDetailResource()).thenReturn(false);
|
||||
|
||||
long commandId = 1L;
|
||||
JsonCommand jsonCommand = Mockito.mock(JsonCommand.class);
|
||||
when(jsonCommand.commandId()).thenReturn(commandId);
|
||||
|
||||
NewCommandSourceHandler commandHandler = Mockito.mock(NewCommandSourceHandler.class);
|
||||
CommandProcessingResult commandProcessingResult = Mockito.mock(CommandProcessingResult.class);
|
||||
when(commandProcessingResult.isRollbackTransaction()).thenReturn(false);
|
||||
when(commandHandler.processCommand(jsonCommand)).thenReturn(commandProcessingResult);
|
||||
|
||||
when(commandHandlerProvider.getHandler(Mockito.any(), Mockito.any())).thenReturn(commandHandler);
|
||||
|
||||
when(configurationDomainService.isMakerCheckerEnabledForTask(Mockito.any())).thenReturn(false);
|
||||
String idk = "idk";
|
||||
when(idempotencyKeyResolver.resolve(commandWrapper)).thenReturn(idk);
|
||||
CommandSource commandSource = Mockito.mock(CommandSource.class);
|
||||
when(commandSource.getId()).thenReturn(commandId);
|
||||
when(commandSourceService.findCommandSource(commandWrapper, idk)).thenReturn(null);
|
||||
when(commandSourceService.getCommandSource(commandId)).thenReturn(commandSource);
|
||||
|
||||
AppUser appUser = Mockito.mock(AppUser.class);
|
||||
when(commandSourceService.saveInitialNewTransaction(commandWrapper, jsonCommand, appUser, idk)).thenReturn(commandSource);
|
||||
when(commandSourceService.saveResultSameTransaction(commandSource)).thenReturn(commandSource);
|
||||
when(commandSource.getStatus()).thenReturn(CommandProcessingResultType.PROCESSED.getValue());
|
||||
when(context.authenticatedUser(Mockito.any(CommandWrapper.class))).thenReturn(appUser);
|
||||
|
||||
when(commandSourceService.processCommand(commandHandler, jsonCommand, commandSource, appUser, false))
|
||||
.thenThrow(new RetryException()).thenThrow(new RetryException()).thenReturn(commandProcessingResult);
|
||||
|
||||
CommandProcessingResult actualCommandProcessingResult = underTest.executeCommand(commandWrapper, jsonCommand, false);
|
||||
|
||||
assertEquals(CommandProcessingResultType.PROCESSED.getValue(), commandSource.getStatus());
|
||||
assertEquals(commandProcessingResult, actualCommandProcessingResult);
|
||||
// verify 2x throw before success
|
||||
verify(commandSourceService, times(2)).generateErrorInfo(any());
|
||||
verify(commandSourceService).saveResultSameTransaction(commandSource);
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -186,4 +264,7 @@ public class SynchronousCommandProcessingServiceTest {
|
||||
underTest.publishHookEvent(entityName, actionName, command, Object.class);
|
||||
});
|
||||
}
|
||||
|
||||
private static final class RetryException extends RuntimeException {}
|
||||
|
||||
}
|
||||
|
||||
@@ -259,11 +259,11 @@ spring.batch.initialize-schema=NEVER
|
||||
# Disabling Spring Batch jobs on startup
|
||||
spring.batch.job.enabled=false
|
||||
|
||||
resilience4j.retry.instances.executeCommand.max-attempts=3
|
||||
resilience4j.retry.instances.executeCommand.wait-duration=1s
|
||||
resilience4j.retry.instances.executeCommand.enable-exponential-backoff=true
|
||||
resilience4j.retry.instances.executeCommand.exponential-backoff-multiplier=2
|
||||
resilience4j.retry.instances.executeCommand.retryExceptions=org.springframework.dao.CannotAcquireLockException,org.springframework.orm.ObjectOptimisticLockingFailureException
|
||||
fineract.retry.instances.executeCommand.max-attempts=3
|
||||
fineract.retry.instances.executeCommand.wait-duration=1s
|
||||
fineract.retry.instances.executeCommand.enable-exponential-backoff=true
|
||||
fineract.retry.instances.executeCommand.exponential-backoff-multiplier=2
|
||||
fineract.retry.instances.executeCommand.retryExceptions=org.springframework.dao.CannotAcquireLockException,org.springframework.orm.ObjectOptimisticLockingFailureException
|
||||
|
||||
resilience4j.retry.instances.processJobDetailForExecution.max-attempts=3
|
||||
resilience4j.retry.instances.processJobDetailForExecution.wait-duration=1s
|
||||
|
||||
@@ -19,9 +19,10 @@
|
||||
|
||||
Feature: Commands Service
|
||||
|
||||
@soma
|
||||
Scenario: Verify that command source write service are working with fallback function
|
||||
Given A command source write service
|
||||
When The user executes the command via a command write service with exceptions
|
||||
Then The command processing service should fallback as expected
|
||||
Then The command processing service execute function should be called 3 times
|
||||
Then The command processing service execute function should be called 2 times
|
||||
|
||||
|
||||
Reference in New Issue
Block a user