Fix PostgresSubscribableChannel race condition

The `PostgresSubscribableChannel` uses a task executor
for dispatching messages.
Even if we stop `PostgresChannelMessageTableSubscriber` and
unsubscribe from the channel, the task might be ongoing.

* Use explicit `ThreadPoolTaskExecutor` in the test
to shout it down and wait for tasks to be completed before
verifying DB status
* Optimize `PostgresSubscribableChannel` to mark TX for rollback
when we got a message from DB, but no handlers subscribed

Author: artembilan

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

@@ -176,7 +176,16 @@ private Optional<Message<?>> askForMessage() {
 		if (this.hasHandlers) {
 			if (this.transactionTemplate != null) {
 				return this.retryTemplate.execute(context ->
-						this.transactionTemplate.execute(status -> pollMessage().map(this::dispatch)));
+						this.transactionTemplate.execute(status ->
+								pollMessage()
+										.filter(message -> {
+											if (!this.hasHandlers) {
+												status.setRollbackOnly();
+												return false;
+											}
+											return true;
+										})
+										.map(this::dispatch)));
 			}
 			else {
 				return pollMessage()

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

@@ -48,6 +48,7 @@
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.retry.support.RetryTemplate;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.transaction.PlatformTransactionManager;
@@ -95,6 +96,8 @@ CREATE FUNCTION INT_CHANNEL_MESSAGE_NOTIFY_FCT()
 
 	private PostgresSubscribableChannel postgresSubscribableChannel;
 
+	private ThreadPoolTaskExecutor taskExecutor;
+
 	private String groupId;
 
 	@BeforeEach
@@ -107,17 +110,26 @@ void setUp(TestInfo testInfo) {
 										POSTGRES_CONTAINER.getPassword())
 								.unwrap(PgConnection.class));
 
+
+		this.taskExecutor = new ThreadPoolTaskExecutor();
+		this.taskExecutor.setCorePoolSize(10);
+		this.taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
+		this.taskExecutor.setAwaitTerminationSeconds(10);
+		this.taskExecutor.afterPropertiesSet();
+
 		this.groupId = testInfo.getDisplayName();
 
 		this.postgresSubscribableChannel =
 				new PostgresSubscribableChannel(messageStore, groupId, postgresChannelMessageTableSubscriber);
 		this.postgresSubscribableChannel.setBeanName("testPostgresChannel");
+		this.postgresSubscribableChannel.setDispatcherExecutor(this.taskExecutor);
 		this.postgresSubscribableChannel.afterPropertiesSet();
 	}
 
 	@AfterEach
 	void tearDown() {
 		this.postgresChannelMessageTableSubscriber.stop();
+		this.taskExecutor.shutdown();
 	}
 
 
@@ -160,13 +172,13 @@ void testMessagesDispatchedInTransaction() throws InterruptedException {
 		postgresChannelMessageTableSubscriber.start();
 		MessageHandler messageHandler =
 				message -> {
-			try {
-				throw new RuntimeException("An error has occurred");
-			}
-			finally {
-				latch.countDown();
-			}
-		};
+					try {
+						throw new RuntimeException("An error has occurred");
+					}
+					finally {
+						latch.countDown();
+					}
+				};
 		postgresSubscribableChannel.subscribe(messageHandler);
 
 		messageStore.addMessageToGroup(groupId, new GenericMessage<>("1"));
@@ -177,6 +189,7 @@ void testMessagesDispatchedInTransaction() throws InterruptedException {
 		// Stop subscriber to unlock records from TX for the next verification
 		postgresChannelMessageTableSubscriber.stop();
 		postgresSubscribableChannel.unsubscribe(messageHandler);
+		this.taskExecutor.shutdown();
 
 		assertThat(messageStore.messageGroupSize(groupId)).isEqualTo(2);
 		assertThat(messageStore.pollMessageFromGroup(groupId).getPayload()).isEqualTo("1");