From 9962ee49a2e0804ed11c3fe3b66e1e766900e5a4 Mon Sep 17 00:00:00 2001 From: NaccOll Date: Sat, 23 Nov 2024 13:58:23 -0500 Subject: [PATCH] GH-5123: Add `LockRegistry` to `AbstractMessageGroupStore` Fixes: https://github.com/spring-projects/spring-integration/issues/5123 When `RedisMessageStore`, for example, adds and removes messages, it operates on two keys separately, which may cause problems in multi-threading due to non-atomic operations. Although using Redis to delay messages is not a good idea, the abnormal loss of messages in the logs alerted me when the number of requests was not large. By comparing the logs, the problem that the message group representing the metadata is not consistent with the actual message. A simple solution is to add lock like in the `SimpleMessageStore`, which is also the approach taken in this pull request. * Add `LockRegistry` to `AbstractMessageGroupStore` * Normalize access levels and method name about the lock of `MessageGroupStore` * Add document about the lock of `AbstractMessageGroupStore` --- .../store/AbstractKeyValueMessageStore.java | 21 +- .../store/AbstractMessageGroupStore.java | 115 +++++++++- .../integration/store/SimpleMessageStore.java | 210 ++++++------------ ...bstractCorrelatingMessageHandlerTests.java | 5 +- .../integration/store/MessageStoreTests.java | 31 ++- .../store/SimpleMessageStoreTests.java | 12 +- .../jdbc/store/JdbcMessageStore.java | 19 +- ...stractConfigurableMongoDbMessageStore.java | 15 +- .../ConfigurableMongoDbMessageStore.java | 17 +- .../store/MongoDbChannelMessageStore.java | 5 +- .../mongodb/store/MongoDbMessageStore.java | 19 +- .../store/RedisMessageGroupStoreTests.java | 11 +- .../modules/ROOT/pages/message-store.adoc | 10 + .../antora/modules/ROOT/pages/whats-new.adoc | 8 +- 14 files changed, 284 insertions(+), 214 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index b74be6ca43..608e05e07d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -38,13 +38,12 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan * @author Ngoc Nhan + * @author Youbin Wu * * @since 2.1 */ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupStore implements MessageStore { - private static final String GROUP_ID_MUST_NOT_BE_NULL = "'groupId' must not be null"; - protected static final String MESSAGE_KEY_PREFIX = "MESSAGE_"; protected static final String MESSAGE_GROUP_KEY_PREFIX = "GROUP_OF_MESSAGES_"; @@ -206,7 +205,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'messages' must not be null"); @@ -240,7 +239,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'messages' must not be null"); @@ -283,7 +282,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messageId, "'messageId' must not be null"); Object mgm = doRetrieve(this.groupPrefix + groupId); @@ -305,7 +304,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); MessageGroupMetadata metadata = getGroupMetadata(groupId); if (metadata != null) { @@ -319,7 +318,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS * Remove the MessageGroup with the provided group ID. */ @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Object mgm = doRemove(this.groupPrefix + groupId); if (mgm != null) { @@ -337,7 +336,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { MessageGroupMetadata metadata = getGroupMetadata(groupId); if (metadata != null) { metadata.setCondition(condition); @@ -346,7 +345,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); MessageGroupMetadata metadata = getGroupMetadata(groupId); if (metadata == null) { @@ -359,7 +358,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { MessageGroupMetadata groupMetadata = getGroupMetadata(groupId); if (groupMetadata != null) { UUID firstId = groupMetadata.firstId(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index 8496a51bf1..89c8bdb05f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 the original author or authors. + * Copyright 2002-2025 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,16 +19,22 @@ package org.springframework.integration.store; import java.util.Arrays; import java.util.Collection; import java.util.LinkedHashSet; +import java.util.UUID; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.integration.support.locks.DefaultLockRegistry; +import org.springframework.integration.support.locks.LockRegistry; +import org.springframework.integration.util.CheckedCallable; +import org.springframework.integration.util.CheckedRunnable; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; +import org.springframework.util.Assert; /** * @author Dave Syer @@ -36,6 +42,7 @@ import org.springframework.messaging.Message; * @author Gary Russell * @author Artem Bilan * @author Christian Tzolov + * @author Youbin Wu * * @since 2.0 */ @@ -43,6 +50,10 @@ import org.springframework.messaging.Message; public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageGroupStore implements MessageGroupStore, Iterable { + protected static final String INTERRUPTED_WHILE_OBTAINING_LOCK = "Interrupted while obtaining lock"; + + protected static final String GROUP_ID_MUST_NOT_BE_NULL = "'groupId' must not be null"; + protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR final private final Lock lock = new ReentrantLock(); @@ -56,6 +67,8 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG private boolean timeoutOnIdle; + private LockRegistry lockRegistry = new DefaultLockRegistry(); + protected AbstractMessageGroupStore() { } @@ -109,6 +122,20 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG this.lazyLoadMessageGroups = lazyLoadMessageGroups; } + /** + * Specify the type of the {@link LockRegistry} to ensure atomic operations + * @param lockRegistry lockRegistryType + * @since 6.5 + */ + public final void setLockRegistry(LockRegistry lockRegistry) { + Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); + this.lockRegistry = lockRegistry; + } + + protected LockRegistry getLockRegistry() { + return this.lockRegistry; + } + @Override public void registerMessageGroupExpiryCallback(MessageGroupCallback callback) { if (callback instanceof UniqueExpiryCallback) { @@ -195,12 +222,98 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG removeMessagesFromGroup(key, Arrays.asList(messages)); } + @Override + public void removeMessagesFromGroup(Object key, Collection> messages) { + Assert.notNull(key, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(key, () -> doRemoveMessagesFromGroup(key, messages)); + } + + protected abstract void doRemoveMessagesFromGroup(Object key, Collection> messages); + + @Override + public void addMessagesToGroup(Object groupId, Message... messages) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doAddMessagesToGroup(groupId, messages)); + } + + protected abstract void doAddMessagesToGroup(Object groupId, Message... messages); + @Override public MessageGroup addMessageToGroup(Object groupId, Message message) { addMessagesToGroup(groupId, message); return getMessageGroup(groupId); } + @Override + public void removeMessageGroup(Object groupId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doRemoveMessageGroup(groupId)); + } + + protected abstract void doRemoveMessageGroup(Object groupId); + + @Override + public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + return executeLocked(groupId, () -> doRemoveMessageFromGroupById(groupId, messageId)); + } + + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { + throw new UnsupportedOperationException("Not supported for this store"); + } + + @Override + public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doSetLastReleasedSequenceNumberForGroup(groupId, sequenceNumber)); + } + + protected abstract void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber); + + @Override + public void completeGroup(Object groupId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doCompleteGroup(groupId)); + } + + protected abstract void doCompleteGroup(Object groupId); + + @Override + public void setGroupCondition(Object groupId, String condition) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + executeLocked(groupId, () -> doSetGroupCondition(groupId, condition)); + } + + protected abstract void doSetGroupCondition(Object groupId, String condition); + + @Override + public Message pollMessageFromGroup(Object groupId) { + Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); + return executeLocked(groupId, () -> doPollMessageFromGroup(groupId)); + } + + protected abstract Message doPollMessageFromGroup(Object groupId); + + protected T executeLocked(Object groupId, CheckedCallable runnable) { + try { + return this.lockRegistry.executeLocked(groupId, runnable); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); + } + } + + protected void executeLocked(Object groupId, CheckedRunnable runnable) { + try { + this.lockRegistry.executeLocked(groupId, runnable); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); + } + } + private void expire(MessageGroup group) { RuntimeException exception = null; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 5f6cd4be70..230b33cb48 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -45,6 +45,7 @@ import org.springframework.util.CollectionUtils; * @author Gary Russell * @author Ryan Barker * @author Artem Bilan + * @author Youbin Wu * * @since 2.0 */ @@ -55,8 +56,6 @@ public class SimpleMessageStore extends AbstractMessageGroupStore private static final String UPPER_BOUND_MUST_NOT_BE_NULL = "'upperBound' must not be null."; - private static final String INTERRUPTED_WHILE_OBTAINING_LOCK = "Interrupted while obtaining lock"; - private final ConcurrentMap> idToMessage = new ConcurrentHashMap<>(); private final ConcurrentMap groupIdToMessageGroup = new ConcurrentHashMap<>(); @@ -71,12 +70,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore private final long upperBoundTimeout; - private LockRegistry lockRegistry; - private boolean copyOnGet = false; - private volatile boolean isUsed; - /** * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given * capacity is less than 1. The capacities are applied independently to messages stored via @@ -133,8 +128,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore this.individualUpperBound = new UpperBound(individualCapacity); this.individualCapacity = individualCapacity; this.groupCapacity = groupCapacity; - this.lockRegistry = lockRegistry; this.upperBoundTimeout = upperBoundTimeout; + setLockRegistry(lockRegistry); } /** @@ -162,12 +157,6 @@ public class SimpleMessageStore extends AbstractMessageGroupStore this.copyOnGet = copyOnGet; } - public void setLockRegistry(LockRegistry lockRegistry) { - Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); - Assert.isTrue(!(this.isUsed), "Cannot change the lock registry after the store has been used"); - this.lockRegistry = lockRegistry; - } - @Override public void setLazyLoadMessageGroups(boolean lazyLoadMessageGroups) { throw new UnsupportedOperationException("The lazy-load isn't supported for in-memory 'SimpleMessageStore'"); @@ -181,7 +170,6 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override public Message addMessage(Message message) { - this.isUsed = true; if (!this.individualUpperBound.tryAcquire(this.upperBoundTimeout)) { throw new MessagingException(getClass().getSimpleName() + " was out of capacity (" @@ -246,7 +234,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override protected MessageGroup copy(MessageGroup group) { Object groupId = group.getGroupId(); - Lock lock = this.lockRegistry.obtain(groupId); + Lock lock = getLockRegistry().obtain(groupId); try { lock.lockInterruptibly(); try { @@ -261,9 +249,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore lock.unlock(); } } - catch (InterruptedException e) { + catch (InterruptedException ex) { Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); } } @@ -272,7 +260,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messages, "'messages' must not be null"); - Lock lock = this.lockRegistry.obtain(groupId); + Lock lock = getLockRegistry().obtain(groupId); try { lock.lockInterruptibly(); boolean unlocked = false; @@ -314,71 +302,50 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } } } - catch (InterruptedException e) { + catch (InterruptedException ex) { Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); } } - private MessagingException outOfCapacityException(Object groupId) { - return new MessagingException(getClass().getSimpleName() + + @Override + protected void doAddMessagesToGroup(Object groupId, Message... messages) { + // No implementation: the addMessagesToGroup() fully uses locking algorithm. + } + + private IllegalStateException outOfCapacityException(Object groupId) { + return new IllegalStateException(getClass().getSimpleName() + " was out of capacity (" + this.groupCapacity + ") for group '" + groupId + "', try constructing it with a larger number."); } @Override - public void removeMessageGroup(Object groupId) { - Lock lock = this.lockRegistry.obtain(groupId); - try { - lock.lockInterruptibly(); - try { - MessageGroup messageGroup = this.groupIdToMessageGroup.remove(groupId); - if (messageGroup != null) { - UpperBound upperBound = this.groupToUpperBound.remove(groupId); - Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); - upperBound.release(this.groupCapacity); - } - } - finally { - lock.unlock(); - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + protected void doRemoveMessageGroup(Object groupId) { + MessageGroup messageGroup = this.groupIdToMessageGroup.remove(groupId); + if (messageGroup != null) { + UpperBound upperBound = this.groupToUpperBound.remove(groupId); + Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); + upperBound.release(this.groupCapacity); } } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { - Lock lock = this.lockRegistry.obtain(groupId); - try { - lock.lockInterruptibly(); - try { - MessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, - () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + - "can not be located while attempting to remove Message(s) from the MessageGroup"); - UpperBound upperBound = this.groupToUpperBound.get(groupId); - Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); - boolean modified = false; - for (Message messageToRemove : messages) { - if (group.remove(messageToRemove)) { - upperBound.release(); - modified = true; - } - } - if (modified) { - group.setLastModified(System.currentTimeMillis()); - } - } - finally { - lock.unlock(); + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to remove Message(s) from the MessageGroup"); + UpperBound upperBound = this.groupToUpperBound.get(groupId); + Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); + boolean modified = false; + for (Message messageToRemove : messages) { + if (group.remove(messageToRemove)) { + upperBound.release(); + modified = true; } } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + if (modified) { + group.setLastModified(System.currentTimeMillis()); } } @@ -397,35 +364,22 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { - Lock lock = this.lockRegistry.obtain(groupId); - try { - lock.lockInterruptibly(); - try { - MessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, - () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + - "can not be located while attempting to remove Message from the MessageGroup"); - UpperBound upperBound = this.groupToUpperBound.get(groupId); - Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); - for (Message message : group.getMessages()) { - if (messageId.equals(message.getHeaders().getId())) { - group.remove(message); - upperBound.release(); - group.setLastModified(System.currentTimeMillis()); - return true; - } - } - return false; - } - finally { - lock.unlock(); + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to remove Message from the MessageGroup"); + UpperBound upperBound = this.groupToUpperBound.get(groupId); + Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL); + for (Message message : group.getMessages()) { + if (messageId.equals(message.getHeaders().getId())) { + group.remove(message); + upperBound.release(); + group.setLastModified(System.currentTimeMillis()); + return true; } } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); - } + return false; } @Override @@ -434,7 +388,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { MessageGroup group = this.groupIdToMessageGroup.get(groupId); if (group != null) { group.setCondition(condition); @@ -442,53 +396,27 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { - Lock lock = this.lockRegistry.obtain(groupId); - try { - lock.lockInterruptibly(); - try { - MessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, - () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + - "can not be located while attempting to set 'lastReleasedSequenceNumber'"); - group.setLastReleasedMessageSequenceNumber(sequenceNumber); - group.setLastModified(System.currentTimeMillis()); - } - finally { - lock.unlock(); - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); - } + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to set 'lastReleasedSequenceNumber'"); + group.setLastReleasedMessageSequenceNumber(sequenceNumber); + group.setLastModified(System.currentTimeMillis()); } @Override - public void completeGroup(Object groupId) { - Lock lock = this.lockRegistry.obtain(groupId); - try { - lock.lockInterruptibly(); - try { - MessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, - () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + - "can not be located while attempting to complete the MessageGroup"); - group.complete(); - group.setLastModified(System.currentTimeMillis()); - } - finally { - lock.unlock(); - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); - } + protected void doCompleteGroup(Object groupId) { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, + () -> MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " + + "can not be located while attempting to complete the MessageGroup"); + group.complete(); + group.setLastModified(System.currentTimeMillis()); } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { Collection> messageList = getMessageGroup(groupId).getMessages(); Message message = null; if (!CollectionUtils.isEmpty(messageList)) { @@ -521,7 +449,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } public void clearMessageGroup(Object groupId) { - Lock lock = this.lockRegistry.obtain(groupId); + Lock lock = getLockRegistry().obtain(groupId); try { lock.lockInterruptibly(); try { @@ -539,9 +467,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore lock.unlock(); } } - catch (InterruptedException e) { + catch (InterruptedException ex) { Thread.currentThread().interrupt(); - throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e); + throw new IllegalStateException(INTERRUPTED_WHILE_OBTAINING_LOCK, ex); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index 757ed23a32..1c472142f1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2025 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. @@ -56,6 +56,7 @@ import static org.mockito.Mockito.verify; * @author Gary Russell * @author Artem Bilan * @author Meherzad Lahewala + * @author Youbin Wu * * @since 2.2 * @@ -350,7 +351,7 @@ public class AbstractCorrelatingMessageHandlerTests { SimpleMessageStore messageStore = new SimpleMessageStore() { @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { throw new RuntimeException("intentional"); } }; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java index 988de55451..d12df63758 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -22,9 +22,8 @@ import java.util.Collections; import java.util.Iterator; import java.util.List; -import org.junit.Test; +import org.junit.jupiter.api.Test; -import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.util.ReflectionTestUtils; @@ -35,22 +34,22 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Dave Syer * @author Gary Russell * @author Artem Bilan + * @author Youbin Wu */ public class MessageStoreTests { @Test - public void shouldRegisterCallbacks() throws Exception { + public void shouldRegisterCallbacks() { TestMessageStore store = new TestMessageStore(); - store.setExpiryCallbacks(Collections.singletonList((messageGroupStore, group) -> { + store.setExpiryCallbacks(Collections.singletonList((messageGroupStore, group) -> { })); assertThat(((Collection) ReflectionTestUtils.getField(store, "expiryCallbacks")).size()).isEqualTo(1); } @Test - public void shouldExpireMessageGroup() throws Exception { - + public void shouldExpireMessageGroup() { TestMessageStore store = new TestMessageStore(); - final List list = new ArrayList(); + final List list = new ArrayList<>(); store.registerMessageGroupExpiryCallback((messageGroupStore, group) -> { list.add(group.getOne().getPayload().toString()); messageGroupStore.removeMessageGroup(group.getGroupId()); @@ -63,13 +62,13 @@ public class MessageStoreTests { } @Test - public void testGroupCount() throws Exception { + public void testGroupCount() { TestMessageStore store = new TestMessageStore(); assertThat(store.getMessageGroupCount()).isEqualTo(1); } @Test - public void testGroupSizes() throws Exception { + public void testGroupSizes() { TestMessageStore store = new TestMessageStore(); assertThat(store.getMessageCountForAllMessageGroups()).isEqualTo(1); } @@ -91,7 +90,7 @@ public class MessageStoreTests { } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { throw new UnsupportedOperationException(); } @@ -101,30 +100,30 @@ public class MessageStoreTests { } @Override - public void removeMessagesFromGroup(Object key, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object key, Collection> messages) { throw new UnsupportedOperationException(); } @Override - public void removeMessageGroup(Object correlationKey) { + protected void doRemoveMessageGroup(Object correlationKey) { if (correlationKey.equals(testMessages.getGroupId())) { removed = true; } } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { throw new UnsupportedOperationException(); } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { throw new UnsupportedOperationException(); } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { return null; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index 7e87f040ac..bb196a0af7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -34,6 +34,7 @@ import org.springframework.test.util.ReflectionTestUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; /** * @author Iwein Fuld @@ -45,7 +46,6 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; public class SimpleMessageStoreTests { @Test - @SuppressWarnings("unchecked") public void shouldRetainMessage() { SimpleMessageStore store = new SimpleMessageStore(); Message testMessage1 = MessageBuilder.withPayload("foo").build(); @@ -124,7 +124,7 @@ public class SimpleMessageStoreTests { Message testMessage1 = MessageBuilder.withPayload("foo").build(); Message testMessage2 = MessageBuilder.withPayload("bar").build(); store.addMessageToGroup("foo", testMessage1); - assertThatExceptionOfType(MessagingException.class) + assertThatIllegalStateException() .isThrownBy(() -> store.addMessageToGroup("foo", testMessage2)); } @@ -158,7 +158,7 @@ public class SimpleMessageStoreTests { SimpleMessageStore store2 = new SimpleMessageStore(1, 1, 1); store2.addMessageToGroup("foo", MessageBuilder.withPayload("foo").build()); - assertThatExceptionOfType(MessagingException.class) + assertThatIllegalStateException() .isThrownBy(() -> store2.addMessageToGroup("foo", MessageBuilder.withPayload("bar").build())); } @@ -178,13 +178,13 @@ public class SimpleMessageStoreTests { Message testMessage2 = MessageBuilder.withPayload("bar").build(); store.addMessageToGroup("foo", testMessage1); - assertThatExceptionOfType(MessagingException.class) + assertThatIllegalStateException() .isThrownBy(() -> store.addMessageToGroup("foo", testMessage2)) .withMessageContaining("was out of capacity (1) for group 'foo'"); store.removeMessagesFromGroup("foo", testMessage2); - assertThatExceptionOfType(MessagingException.class) + assertThatIllegalStateException() .isThrownBy(() -> store.addMessageToGroup("foo", testMessage2)) .withMessageContaining("was out of capacity (1) for group 'foo'"); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java index db47f54235..73d0f99760 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -80,6 +80,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Artem Bilan * @author Ngoc Nhan + * @author Youbin Wu * * @since 2.0 */ @@ -472,7 +473,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { String groupKey = getKey(groupId); MessageGroupMetadata groupMetadata = getGroupMetadata(groupKey); @@ -576,7 +577,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messages, "'messages' must not be null"); @@ -621,7 +622,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { String groupKey = getKey(groupId); String messageKey = getKey(messageId); int messageToGroupRemoved = @@ -634,7 +635,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { String groupKey = getKey(groupId); this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGES_FROM_GROUP), @@ -653,7 +654,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { final String groupKey = getKey(groupId); if (logger.isDebugEnabled()) { @@ -664,7 +665,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { Assert.notNull(groupId, "'groupId' must not be null"); String groupKey = getKey(groupId); Timestamp updatedDate = new Timestamp(System.currentTimeMillis()); @@ -675,7 +676,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, "'groupId' must not be null"); String groupKey = getKey(groupId); @@ -687,7 +688,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { String key = getKey(groupId); Message polledMessage = doPollForMessage(key); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java index d7cbd4259f..3e23749141 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2024 the original author or authors. + * Copyright 2014-2025 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. @@ -61,6 +61,7 @@ import org.springframework.util.Assert; * * @author Artem Bilan * @author Adama Sorho + * @author Youbin Wu * * @since 4.0 */ @@ -201,7 +202,7 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { this.mongoTemplate.remove(groupIdQuery(groupId), this.collectionName); } @@ -250,17 +251,17 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } @Override - public void removeMessagesFromGroup(Object key, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object key, Collection> messages) { throw NOT_IMPLEMENTED; } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { throw NOT_IMPLEMENTED; } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { throw NOT_IMPLEMENTED; } @@ -270,7 +271,7 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { throw NOT_IMPLEMENTED; } @@ -280,7 +281,7 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { throw NOT_IMPLEMENTED; } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java index 6176a2ccaf..04e9d860c8 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2024 the original author or authors. + * Copyright 2013-2025 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. @@ -54,6 +54,7 @@ import org.springframework.util.Assert; * @author Artem Bilan * @author Gary Russell * @author Ngoc Nhan + * @author Youbin Wu * * @since 3.0 */ @@ -147,7 +148,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'message' must not be null"); @@ -183,7 +184,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'messageToRemove' must not be null"); @@ -215,7 +216,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messageId, "'messageId' must not be null"); Query query = @@ -234,7 +235,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public Message pollMessageFromGroup(final Object groupId) { + protected Message doPollMessageFromGroup(final Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Sort sort = Sort.by(MessageDocumentFields.LAST_MODIFIED_TIME, MessageDocumentFields.SEQUENCE); @@ -249,17 +250,17 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.LAST_RELEASED_SEQUENCE, sequenceNumber)); } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { updateGroup(groupId, lastModifiedUpdate().set("condition", condition)); } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.COMPLETE, true)); } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java index 76ee7da77a..b9c0ed38b5 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2023 the original author or authors. + * Copyright 2014-2025 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. @@ -44,6 +44,7 @@ import org.springframework.util.Assert; * * @author Artem Bilan * @author Adama Sorho + * @author Youbin Wu * * @since 4.0 */ @@ -132,7 +133,7 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa } @Override - public Message pollMessageFromGroup(Object groupId) { + protected Message doPollMessageFromGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); Sort sort = Sort.by(MessageDocumentFields.LAST_MODIFIED_TIME, MessageDocumentFields.SEQUENCE); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java index 617d7e4c11..62baaf27bb 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 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. @@ -97,6 +97,7 @@ import org.springframework.util.StringUtils; * @author Jodie StJohn * @author Gary Russell * @author Artem Bilan + * @author Youbin Wu * * @since 2.1 */ @@ -294,7 +295,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public void addMessagesToGroup(Object groupId, Message... messages) { + protected void doAddMessagesToGroup(Object groupId, Message... messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'message' must not be null"); Query query = whereGroupIdOrder(groupId); @@ -328,7 +329,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public void removeMessagesFromGroup(Object groupId, Collection> messages) { + protected void doRemoveMessagesFromGroup(Object groupId, Collection> messages) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messages, "'messageToRemove' must not be null"); @@ -367,7 +368,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public boolean removeMessageFromGroupById(Object groupId, UUID messageId) { + protected boolean doRemoveMessageFromGroupById(Object groupId, UUID messageId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Assert.notNull(messageId, "'messageId' must not be null"); return this.template.remove(whereMessageIdIsAndGroupIdIs(messageId, groupId), this.collectionName) @@ -375,7 +376,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public void removeMessageGroup(Object groupId) { + protected void doRemoveMessageGroup(Object groupId) { this.template.remove(whereGroupIdIs(groupId), this.collectionName); } @@ -395,7 +396,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public Message pollMessageFromGroup(final Object groupId) { + protected Message doPollMessageFromGroup(final Object groupId) { Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL); Query query = whereGroupIdIs(groupId).with(Sort.by(GROUP_UPDATE_TIMESTAMP_KEY, SEQUENCE)); MessageWrapper messageWrapper = this.template.findAndRemove(query, MessageWrapper.class, this.collectionName); @@ -415,17 +416,17 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override - public void setGroupCondition(Object groupId, String condition) { + protected void doSetGroupCondition(Object groupId, String condition) { updateGroup(groupId, lastModifiedUpdate().set("_condition", condition)); } @Override - public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + protected void doSetLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { updateGroup(groupId, lastModifiedUpdate().set(LAST_RELEASED_SEQUENCE_NUMBER, sequenceNumber)); } @Override - public void completeGroup(Object groupId) { + protected void doCompleteGroup(Object groupId) { this.updateGroup(groupId, lastModifiedUpdate().set(GROUP_COMPLETE_KEY, true)); } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java index 68bd2d112c..d54cf724c0 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2024 the original author or authors. + * Copyright 2007-2025 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. @@ -493,6 +493,15 @@ class RedisMessageGroupStoreTests implements RedisContainerTest { assertThat(store.messageGroupSize("2")).isEqualTo(1); } + @Test + public void testMessageGroupCondition() { + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").build(); + store.addMessagesToGroup(groupId, message); + store.setGroupCondition(groupId, "testCondition"); + assertThat(store.getMessageGroup(groupId).getCondition()).isEqualTo("testCondition"); + } + private record Foo(String foo) { } diff --git a/src/reference/antora/modules/ROOT/pages/message-store.adoc b/src/reference/antora/modules/ROOT/pages/message-store.adoc index 3688e452bd..8dfbfab10e 100644 --- a/src/reference/antora/modules/ROOT/pages/message-store.adoc +++ b/src/reference/antora/modules/ROOT/pages/message-store.adoc @@ -158,3 +158,13 @@ It also allows the end marker to arrive at the aggregator before all the other r In addition, for configuration convenience, a `GroupConditionProvider` contract has been introduced. The `AbstractCorrelatingMessageHandler` checks if the provided `ReleaseStrategy` implements this interface and extracts a `conditionSupplier` for group condition evaluation logic. + +[[use-lock-registry]] +== Use `LockRegistry` + +Starting with version 6.5, the `AbstractMessageGroupStore` abstraction operates a metadata of message group with a lock. +This lock acquires the groupId and generated by `LockRegister`. +Its purpose is to operate on the atomicity of messages and message groups. +In multiple threads, adding or removing messages or updating metadata at the same time, some implementations may have message group errors if the lock is missing. +By default, the `DefaultLockRegistry` is used, any `LockRegister` can be injected via `AbstractMessageGroupStore.setLockRegistry()`, usually an implementation for the same persistent store. +See more xref:distributed-locks.adoc[Distributed Locks] for more information. \ No newline at end of file diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 2726f6db8f..5216a7a6ab 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -25,4 +25,10 @@ See xref:control-bus.adoc[Control Bus] for more information. The `AbstractCorrelatingMessageHandler` does not throw an `IllegalArgumentException` for the collection of payloads as a result of the `MessageGroupProcessor`. Instead, such a collection is wrapped into a single reply message. -See xref:aggregator.adoc[Aggregator] for more information. \ No newline at end of file +See xref:aggregator.adoc[Aggregator] for more information. + +[[x6.4-message-store-with-locks]] +== The `LockRegistry` in the `MessageStore` + +The `AbstractMessageGroupStore` now can be configured with a `LockRegistry` to perform series of persistent operation atomically. +See xref:message-store.adoc#use-lock-registry[Use LockRegistry] for more information. \ No newline at end of file