GH-8711: Add SKIP LOCKED for QueryProviders

Fixes https://github.com/spring-projects/spring-integration/issues/8711

* Add `SKIP LOCKED` into `MySqlChannelMessageStoreQueryProvider` and `PostgresChannelMessageStoreQueryProvider`
* Stop the `postgresChannelMessageTableSubscriber` after the latch waiting 10s has been passed in `testMessagesDispatchedInTransaction` test case

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ package org.springframework.integration.jdbc.store.channel;
 /**
  * @author Gunnar Hillert
  * @author Artem Bilan
+ *
  * @since 2.2
  */
 public abstract class AbstractChannelMessageStoreQueryProvider implements ChannelMessageStoreQueryProvider {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,8 @@ package org.springframework.integration.jdbc.store.channel;
 /**
  * @author Gunnar Hillert
  * @author Artem Bilan
+ * @author Adama Sorho
+ *
  * @since 2.2
  */
 public class MySqlChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
@@ -32,26 +34,26 @@ public class MySqlChannelMessageStoreQueryProvider extends AbstractChannelMessag
 	public String getPollFromGroupExcludeIdsQuery() {
 		return SELECT_COMMON
 				+ "and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) "
-				+ "order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
+				+ "order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE SKIP LOCKED";
 	}
 
 	@Override
 	public String getPollFromGroupQuery() {
 		return SELECT_COMMON +
-				"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
+				"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE SKIP LOCKED";
 	}
 
 	@Override
 	public String getPriorityPollFromGroupExcludeIdsQuery() {
 		return SELECT_COMMON +
 				"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
-				"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
+				"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE SKIP LOCKED";
 	}
 
 	@Override
 	public String getPriorityPollFromGroupQuery() {
 		return SELECT_COMMON +
-				"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
+				"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE SKIP LOCKED";
 	}
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,8 @@ package org.springframework.integration.jdbc.store.channel;
 /**
  * @author Gunnar Hillert
  * @author Artem Bilan
+ * @author Adama Sorho
+ *
  * @since 2.2
  */
 public class PostgresChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
@@ -32,26 +34,26 @@ public class PostgresChannelMessageStoreQueryProvider extends AbstractChannelMes
 	public String getPollFromGroupExcludeIdsQuery() {
 		return SELECT_COMMON
 				+ "and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) "
-				+ "order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
+				+ "order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE SKIP LOCKED";
 	}
 
 	@Override
 	public String getPollFromGroupQuery() {
 		return SELECT_COMMON +
-				"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
+				"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE SKIP LOCKED";
 	}
 
 	@Override
 	public String getPriorityPollFromGroupExcludeIdsQuery() {
 		return SELECT_COMMON +
 				"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
-				"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
+				"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE SKIP LOCKED";
 	}
 
 	@Override
 	public String getPriorityPollFromGroupQuery() {
 		return SELECT_COMMON +
-				"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
+				"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE SKIP LOCKED";
 	}
 
 }

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

@@ -57,6 +57,7 @@ import static org.assertj.core.api.Assertions.assertThat;
  * @author Rafael Winterhalter
  * @author Artem Bilan
  * @author Igor Lovich
+ * @author Adama Sorho
  *
  * @since 6.0
  */
@@ -185,6 +186,9 @@ public class PostgresChannelMessageTableSubscriberTests implements PostgresConta
 
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 
+		// Stop subscriber to unlock records from TX for the next verification
+		postgresChannelMessageTableSubscriber.stop();
+
 		assertThat(messageStore.messageGroupSize(groupId)).isEqualTo(2);
 		assertThat(messageStore.pollMessageFromGroup(groupId).getPayload()).isEqualTo("1");
 		assertThat(messageStore.pollMessageFromGroup(groupId).getPayload()).isEqualTo("2");