GH-8770: Add PostgresSubsChannel.errorHandler (#8777)

* GH-8770: Add `PostgresSubsChannel.errorHandler`

Fixes #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()`

Author: artembilan

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java

@@ -20,7 +20,6 @@
 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.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 void setRetryTemplate(RetryTemplate retryTemplate) {
 		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 @@ protected boolean doSend(Message<?> message, long timeout) {
 	@Override
 	public void notifyUpdate() {
 		this.executor.execute(() -> {
+			Optional<?> dispatchedMessage;
+			do {
+				dispatchedMessage = pollAndDispatchMessage();
+			} while (dispatchedMessage.isPresent());
+		});
+	}
+
+	private Optional<?> pollAndDispatchMessage() {
+		try {
+			return doPollAndDispatchMessage();
+		}
+		catch (Exception ex) {
 			try {
-				Optional<Message<?>> dispatchedMessage;
-				do {
-					dispatchedMessage = askForMessage();
-				} while (dispatchedMessage.isPresent());
+				this.errorHandler.handleError(ex);
 			}
-			catch (Exception ex) {
+			catch (Exception ex1) {
 				LOGGER.error(ex, "Exception during message dispatch");
 			}
-		});
+			return FALLBACK_STUB;
+		}
 	}
 
-	private Optional<Message<?>> askForMessage() {
+	private Optional<?> doPollAndDispatchMessage() {
 		if (this.hasHandlers) {
 			if (this.transactionTemplate != null) {
 				return this.retryTemplate.execute(context ->

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java

@@ -22,6 +22,7 @@
 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.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 @@ void testMessagesDispatchedInTransaction() throws InterruptedException {
 		assertThat(messageStore.pollMessageFromGroup(groupId).getPayload()).isEqualTo("2");
 	}
 
+	@ParameterizedTest
+	@ValueSource(booleans = {true, false})
+	void errorHandlerIsCalled(boolean transactionsEnabled) throws InterruptedException {
+		if (transactionsEnabled) {
+			postgresSubscribableChannel.setTransactionManager(transactionManager);
+		}
+
+		AtomicReference<Throwable> 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 @@ DataSourceInitializer dataSourceInitializer(DataSource dataSource) {
 			ResourceDatabasePopulator databasePopulator =
 					new ResourceDatabasePopulator(new ByteArrayResource(INTEGRATION_DB_SCRIPTS.getBytes()));
 			databasePopulator.setSeparator(ScriptUtils.EOF_STATEMENT_SEPARATOR);
-			dataSourceInitializer.setDatabasePopulator(
-					databasePopulator);
+			dataSourceInitializer.setDatabasePopulator(databasePopulator);
 			return dataSourceInitializer;
 		}