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 422b01333a..bcb9004b20 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
@@ -67,6 +67,8 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
 
 	private Executor executor = new SimpleAsyncTaskExecutor();
 
+	private volatile boolean hasHandlers;
+
 	/**
 	 * Create a subscribable channel for a Postgres database.
 	 * @param jdbcChannelMessageStore The message store to use for the relevant region.
@@ -120,6 +122,7 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
 		boolean subscribed = super.subscribe(handler);
 		if (this.dispatcher.getHandlerCount() == 1) {
 			this.messageTableSubscriber.subscribe(this);
+			this.hasHandlers = true;
 			notifyUpdate();
 		}
 		return subscribed;
@@ -130,6 +133,7 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
 		boolean unsubscribed = super.unsubscribe(handle);
 		if (this.dispatcher.getHandlerCount() == 0) {
 			this.messageTableSubscriber.unsubscribe(this);
+			this.hasHandlers = false;
 		}
 		return unsubscribed;
 	}
@@ -151,18 +155,7 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
 			try {
 				Optional<Message<?>> dispatchedMessage;
 				do {
-					if (this.transactionTemplate != null) {
-						dispatchedMessage =
-								this.retryTemplate.execute(context ->
-										this.transactionTemplate.execute(status ->
-												pollMessage()
-														.map(this::dispatch)));
-					}
-					else {
-						dispatchedMessage =
-								pollMessage()
-										.map(message -> this.retryTemplate.execute(context -> dispatch(message)));
-					}
+					dispatchedMessage = askForMessage();
 				} while (dispatchedMessage.isPresent());
 			}
 			catch (Exception ex) {
@@ -171,6 +164,20 @@ public class PostgresSubscribableChannel extends AbstractSubscribableChannel
 		});
 	}
 
+	private Optional<Message<?>> askForMessage() {
+		if (this.hasHandlers) {
+			if (this.transactionTemplate != null) {
+				return this.retryTemplate.execute(context ->
+						this.transactionTemplate.execute(status -> pollMessage().map(this::dispatch)));
+			}
+			else {
+				return pollMessage()
+						.map(message -> this.retryTemplate.execute(context -> dispatch(message)));
+			}
+		}
+		return Optional.empty();
+	}
+
 	private Optional<Message<?>> pollMessage() {
 		return Optional.ofNullable(this.jdbcChannelMessageStore.pollMessageFromGroup(this.groupId));
 	}
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 bbe9ab567d..20f2b21c04 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
@@ -45,6 +45,7 @@ import org.springframework.jdbc.datasource.DataSourceTransactionManager;
 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.support.GenericMessage;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.test.annotation.DirtiesContext;
@@ -169,20 +170,26 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta
 		postgresSubscribableChannel.setTransactionManager(transactionManager);
 
 		postgresChannelMessageTableSubscriber.start();
-		postgresSubscribableChannel.subscribe(message -> {
+		MessageHandler messageHandler =
+				message -> {
 			try {
 				throw new RuntimeException("An error has occurred");
 			}
 			finally {
 				latch.countDown();
 			}
-		});
+		};
+		postgresSubscribableChannel.subscribe(messageHandler);
 
 		messageStore.addMessageToGroup(groupId, new GenericMessage<>("1"));
 		messageStore.addMessageToGroup(groupId, new GenericMessage<>("2"));
 
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 
+		// Stop subscriber to unlock records from TX for the next verification
+		postgresChannelMessageTableSubscriber.stop();
+		postgresSubscribableChannel.unsubscribe(messageHandler);
+
 		assertThat(messageStore.messageGroupSize(groupId)).isEqualTo(2);
 		assertThat(messageStore.pollMessageFromGroup(groupId).getPayload()).isEqualTo("1");
 		assertThat(messageStore.pollMessageFromGroup(groupId).getPayload()).isEqualTo("2");