From 571ed20905059c2c52f22e74e305f69320e244ad Mon Sep 17 00:00:00 2001 From: Adama Sorho Date: Wed, 30 Aug 2023 18:36:00 -0400 Subject: [PATCH] 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 --- .../AbstractChannelMessageStoreQueryProvider.java | 3 ++- .../MySqlChannelMessageStoreQueryProvider.java | 12 +++++++----- .../PostgresChannelMessageStoreQueryProvider.java | 12 +++++++----- .../PostgresChannelMessageTableSubscriberTests.java | 4 ++++ 4 files changed, 20 insertions(+), 11 deletions(-) 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");