From 91e806afc519c97f4790723a26079b899128761e Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 25 Oct 2023 13:26:24 -0400 Subject: [PATCH] GH-8770: Add `PostgresSubsChannel.errorHandler` (#8777) * GH-8770: Add `PostgresSubsChannel.errorHandler` Fixes https://github.com/spring-projects/spring-integration/issues/8770 The problem with the `PostgresSubscribableChannel.notifyUpdate()` is that the try-catch block is outside the loop, so the loop will die on an exception, leaving further messages unprocessed. * Add ``PostgresSubscribableChannel.errorHandler` option to be invoked after a `RetryTemplate` and for every failed message. * The `askForMessage()` new logic is to catch an exception on a message and call `errorHandler` returning a `FALLBACK_STUB` to continue an outer loop in the `notifyUpdate()` **Cherry-pick to `6.1.x` & `6.0.x`** * * Rename private `PostgresSubscribableChannel.askForMessage()` method to more specific `pollAndDispatchMessage()` --- .../channel/PostgresSubscribableChannel.java | 51 ++++++++++++++----- ...resChannelMessageTableSubscriberTests.java | 33 +++++++++++- 2 files changed, 70 insertions(+), 14 deletions(-) diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java index 4d8d079b67..7536f2b050 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java @@ -20,7 +20,6 @@ import java.util.Optional; import java.util.concurrent.Executor; import org.springframework.core.log.LogAccessor; -import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.channel.AbstractSubscribableChannel; import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.dispatcher.UnicastingDispatcher; @@ -31,6 +30,8 @@ import org.springframework.retry.support.RetryTemplate; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; +import org.springframework.util.ErrorHandler; +import org.springframework.util.ReflectionUtils; /** * An {@link AbstractSubscribableChannel} for receiving push notifications for @@ -53,6 +54,8 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel private static final LogAccessor LOGGER = new LogAccessor(PostgresSubscribableChannel.class); + private static final Optional FALLBACK_STUB = Optional.of(new Object()); + private final JdbcChannelMessageStore jdbcChannelMessageStore; private final Object groupId; @@ -65,7 +68,9 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel private RetryTemplate retryTemplate = RetryTemplate.builder().maxAttempts(1).build(); - private Executor executor = new SimpleAsyncTaskExecutor(); + private ErrorHandler errorHandler = ReflectionUtils::rethrowRuntimeException; + + private Executor executor; private volatile boolean hasHandlers; @@ -77,6 +82,7 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel */ public PostgresSubscribableChannel(JdbcChannelMessageStore jdbcChannelMessageStore, Object groupId, PostgresChannelMessageTableSubscriber messageTableSubscriber) { + Assert.notNull(jdbcChannelMessageStore, "A jdbcChannelMessageStore must be provided."); Assert.notNull(groupId, "A groupId must be set."); Assert.notNull(messageTableSubscriber, "A messageTableSubscriber must be set."); @@ -117,6 +123,17 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel this.retryTemplate = retryTemplate; } + /** + * Set a {@link ErrorHandler} for messages which cannot be dispatched by this channel. + * Used as a recovery callback after {@link RetryTemplate} execution throws an exception. + * @param errorHandler the {@link ErrorHandler} to use. + * @since 6.0.9 + */ + public void setErrorHandler(ErrorHandler errorHandler) { + Assert.notNull(errorHandler, "'errorHandler' must not be null."); + this.errorHandler = errorHandler; + } + @Override public boolean subscribe(MessageHandler handler) { boolean subscribed = super.subscribe(handler); @@ -152,19 +169,29 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel @Override public void notifyUpdate() { this.executor.execute(() -> { - try { - Optional> dispatchedMessage; - do { - dispatchedMessage = askForMessage(); - } while (dispatchedMessage.isPresent()); - } - catch (Exception ex) { - LOGGER.error(ex, "Exception during message dispatch"); - } + Optional dispatchedMessage; + do { + dispatchedMessage = pollAndDispatchMessage(); + } while (dispatchedMessage.isPresent()); }); } - private Optional> askForMessage() { + private Optional pollAndDispatchMessage() { + try { + return doPollAndDispatchMessage(); + } + catch (Exception ex) { + try { + this.errorHandler.handleError(ex); + } + catch (Exception ex1) { + LOGGER.error(ex, "Exception during message dispatch"); + } + return FALLBACK_STUB; + } + } + + private Optional doPollAndDispatchMessage() { if (this.hasHandlers) { if (this.transactionTemplate != null) { return this.retryTemplate.execute(context -> diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java index 005d4330e8..fb4348a167 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java @@ -22,6 +22,7 @@ import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import javax.sql.DataSource; @@ -46,6 +47,7 @@ import org.springframework.jdbc.datasource.init.DataSourceInitializer; import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator; import org.springframework.jdbc.datasource.init.ScriptUtils; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; import org.springframework.retry.support.RetryTemplate; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -210,6 +212,34 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta assertThat(messageStore.pollMessageFromGroup(groupId).getPayload()).isEqualTo("2"); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void errorHandlerIsCalled(boolean transactionsEnabled) throws InterruptedException { + if (transactionsEnabled) { + postgresSubscribableChannel.setTransactionManager(transactionManager); + } + + AtomicReference exceptionReference = new AtomicReference<>(); + CountDownLatch errorHandlerLatch = new CountDownLatch(1); + postgresSubscribableChannel.setErrorHandler(ex -> { + exceptionReference.set(ex); + errorHandlerLatch.countDown(); + }); + + postgresChannelMessageTableSubscriber.start(); + + postgresSubscribableChannel.subscribe(message -> { + throw new RuntimeException("An error has occurred"); + }); + + messageStore.addMessageToGroup(groupId, new GenericMessage<>("1")); + + assertThat(errorHandlerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(exceptionReference.get()) + .isInstanceOf(MessagingException.class) + .hasStackTraceContaining("An error has occurred"); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) void testRetryOnErrorDuringDispatch(boolean transactionsEnabled) throws InterruptedException { @@ -267,8 +297,7 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta ResourceDatabasePopulator databasePopulator = new ResourceDatabasePopulator(new ByteArrayResource(INTEGRATION_DB_SCRIPTS.getBytes())); databasePopulator.setSeparator(ScriptUtils.EOF_STATEMENT_SEPARATOR); - dataSourceInitializer.setDatabasePopulator( - databasePopulator); + dataSourceInitializer.setDatabasePopulator(databasePopulator); return dataSourceInitializer; }