Some cleanup for PostgresSubscribableChannel

The `PostgresChannelMessageTableSubscriberTests.testUnsubscribeHandlerDuringDispatch()` is not stable

* Fix the test to use less stress on unsubscription.
It is still hard to reach consensus with a fast reader which is able to take a lock again and again in its loop.
* Use `fair` for `ReentrantReadWriteLock` in the `PostgresSubscribableChannel`.
However, it still doesn't show entropy in the test execution
* Fix typos in the `PostgresChannelMessageTableSubscriber`
* Add logging for the `inStoreCount` & `receivedCount` in the `testUnsubscribeHandlerDuringDispatch()`

**Auto-cherry-pick to `6.5.x` & `6.4.x`**

Author: artembilan

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

@@ -168,9 +168,7 @@ public void start() {
 			this.latch = new CountDownLatch(1);
 
 			CountDownLatch startingLatch = new CountDownLatch(1);
-			this.future = this.taskExecutor.submit(() -> {
-				doStart(startingLatch);
-			});
+			this.future = this.taskExecutor.submit(() -> doStart(startingLatch));
 
 			try {
 				if (!startingLatch.await(5, TimeUnit.SECONDS)) {
@@ -242,7 +240,7 @@ private void doStart(CountDownLatch startingLatch) {
 				}
 				catch (Exception e) {
 					// The getNotifications method does not throw a meaningful message on interruption.
-					// Therefore, we do not log an error, unless it occurred while active.
+					// Therefore, we do not log an error unless it occurred while active.
 					if (isActive()) {
 						LOGGER.error(e, "Failed to poll notifications from Postgres database");
 					}
@@ -252,7 +250,6 @@ private void doStart(CountDownLatch startingLatch) {
 		finally {
 			this.latch.countDown();
 		}
-
 	}
 
 	private boolean isActive() {
@@ -285,6 +282,7 @@ public void stop() {
 				}
 			}
 			catch (InterruptedException ignored) {
+				Thread.currentThread().interrupt();
 			}
 		}
 		finally {
@@ -311,8 +309,8 @@ public interface Subscription {
 		/**
 		 * Indicate that a message was added to the represented region and
 		 * group id. Note that this method might also be invoked if there are
-		 * no new messages to read, for example if another subscription already
-		 * read those messages or if a new messages might have arrived during
+		 * no new messages to read, for example, if another subscription already
+		 * read those messages or if a new message might have arrived during
 		 * a temporary connection loss.
 		 */
 		void notifyUpdate();

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

@@ -73,7 +73,7 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
 
 	private final UnicastingDispatcher dispatcher = new UnicastingDispatcher();
 
-	private final ReentrantReadWriteLock hasHandlersLock = new ReentrantReadWriteLock();
+	private final ReentrantReadWriteLock hasHandlersLock = new ReentrantReadWriteLock(true);
 
 	private @Nullable TransactionTemplate transactionTemplate;
 

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

@@ -32,6 +32,8 @@
 import javax.sql.DataSource;
 
 import org.apache.commons.dbcp2.BasicDataSource;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -40,6 +42,7 @@
 import org.junit.jupiter.params.provider.ValueSource;
 import org.postgresql.jdbc.PgConnection;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -52,7 +55,6 @@
 import org.springframework.integration.jdbc.channel.PostgresSubscribableChannel;
 import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
 import org.springframework.integration.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider;
-import org.springframework.integration.test.support.TestApplicationContextAware;
 import org.springframework.jdbc.datasource.DataSourceTransactionManager;
 import org.springframework.jdbc.datasource.init.DataSourceInitializer;
 import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
@@ -79,7 +81,9 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
-public class PostgresChannelMessageTableSubscriberTests implements PostgresContainerTest, TestApplicationContextAware {
+public class PostgresChannelMessageTableSubscriberTests implements PostgresContainerTest {
+
+	private static final Log LOGGER = LogFactory.getLog(PostgresChannelMessageTableSubscriberTests.class);
 
 	private static final String INTEGRATION_DB_SCRIPTS = """
 			CREATE FUNCTION INT_CHANNEL_MESSAGE_NOTIFY_FCT()
@@ -99,6 +103,9 @@ CREATE FUNCTION INT_CHANNEL_MESSAGE_NOTIFY_FCT()
 			^^^ END OF SCRIPT ^^^
 			""";
 
+	@Autowired
+	private BeanFactory beanFactory;
+
 	@Autowired
 	private JdbcChannelMessageStore messageStore;
 
@@ -124,17 +131,16 @@ void setUp(TestInfo testInfo) {
 
 		this.taskExecutor = new ThreadPoolTaskExecutor();
 		this.taskExecutor.setCorePoolSize(10);
-		this.taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
-		this.taskExecutor.setAwaitTerminationSeconds(10);
 		this.taskExecutor.afterPropertiesSet();
 
 		this.groupId = testInfo.getDisplayName();
+		messageStore.removeMessageGroup(this.groupId);
 
 		this.postgresSubscribableChannel =
 				new PostgresSubscribableChannel(messageStore, groupId, postgresChannelMessageTableSubscriber);
 		this.postgresSubscribableChannel.setBeanName("testPostgresChannel");
 		this.postgresSubscribableChannel.setDispatcherExecutor(this.taskExecutor);
-		this.postgresSubscribableChannel.setBeanFactory(TEST_INTEGRATION_CONTEXT);
+		this.postgresSubscribableChannel.setBeanFactory(this.beanFactory);
 		this.postgresSubscribableChannel.afterPropertiesSet();
 	}
 
@@ -314,45 +320,33 @@ public void testRenewConnection() throws Exception {
 
 	@Test
 	public void testUnsubscribeHandlerDuringDispatch() throws InterruptedException {
-		int numberOfMessages = 200;
+		int numberOfMessages = 2000;
 		CountDownLatch messagesUntilUnsubscribe = new CountDownLatch(10);
 		AtomicInteger receivedMessages = new AtomicInteger();
 		AtomicReference<Throwable> receivedException = new AtomicReference<>();
 
 		postgresChannelMessageTableSubscriber.start();
 		postgresSubscribableChannel.setErrorHandler(receivedException::set);
 
-		MessageHandler messageHandlerOne = message -> {
-			receivedMessages.getAndIncrement();
-			messagesUntilUnsubscribe.countDown();
-		};
-		postgresSubscribableChannel.subscribe(messageHandlerOne);
-
-		MessageHandler messageHandlerTwo = message -> {
+		MessageHandler messageHandler = message -> {
 			receivedMessages.getAndIncrement();
 			messagesUntilUnsubscribe.countDown();
 		};
-		postgresSubscribableChannel.subscribe(messageHandlerTwo);
-
-		IntStream.range(0, numberOfMessages).forEach(i -> {
-			taskExecutor.execute(() ->
-				messageStore.addMessageToGroup(groupId, new GenericMessage<>(i)));
-		});
+		postgresSubscribableChannel.subscribe(messageHandler);
 
-		taskExecutor.execute(postgresSubscribableChannel::notifyUpdate);
-		taskExecutor.execute(postgresSubscribableChannel::notifyUpdate);
-		taskExecutor.execute(postgresSubscribableChannel::notifyUpdate);
-		taskExecutor.execute(postgresSubscribableChannel::notifyUpdate);
-		taskExecutor.execute(postgresSubscribableChannel::notifyUpdate);
+		IntStream.range(0, numberOfMessages)
+				.forEach(i -> messageStore.addMessageToGroup(groupId, new GenericMessage<>(i)));
 
 		assertThat(messagesUntilUnsubscribe.await(10, TimeUnit.SECONDS)).isTrue();
-		postgresSubscribableChannel.unsubscribe(messageHandlerOne);
-		postgresSubscribableChannel.unsubscribe(messageHandlerTwo);
+		postgresSubscribableChannel.unsubscribe(messageHandler);
 
-		int undeliveredMessages = messageStore.messageGroupSize(groupId);
-		int processedMessages = undeliveredMessages + receivedMessages.get();
-		assertThat(processedMessages).isEqualTo(numberOfMessages);
 		assertThat(receivedException).hasNullValue();
+
+		int inStoreCount = messageStore.messageGroupSize(groupId);
+		LOGGER.warn("inStoreCount: " + inStoreCount);
+		int receivedCount = receivedMessages.get();
+		LOGGER.warn("receivedCount: " + receivedCount);
+		assertThat(inStoreCount + receivedCount).isEqualTo(numberOfMessages);
 	}
 
 	@Configuration