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

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;
@@ -196,6 +198,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<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 {
@@ -253,8 +283,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;
 		}