diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/AbstractChannelMessageStoreQueryProvider.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/AbstractChannelMessageStoreQueryProvider.java index a78ce461b0..a6f80445ea 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/AbstractChannelMessageStoreQueryProvider.java +++ b/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 { diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MySqlChannelMessageStoreQueryProvider.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MySqlChannelMessageStoreQueryProvider.java index 13a5d2ac2b..90766ae124 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MySqlChannelMessageStoreQueryProvider.java +++ b/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"; } } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/PostgresChannelMessageStoreQueryProvider.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/PostgresChannelMessageStoreQueryProvider.java index da7c375efe..f119a3bfec 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/PostgresChannelMessageStoreQueryProvider.java +++ b/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"; } } 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 f29a8bf422..bd7571afcc 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 @@ -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");