From ad61cef7fc8839afe9a7c574eaa789987d63f25f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 14 Jan 2025 13:03:29 -0500 Subject: [PATCH] GH-9754: Add `discardIndividuallyOnExpiry` to aggregator Fixes: https://github.com/spring-projects/spring-integration/issues/9754 Right now a correlation handler can discard messages in the expired group one by one. In some scenarios it would be useful to have single message in discard for the whole group. * Expose `discardIndividuallyOnExpiry` for the `AbstractCorrelatingMessageHandler`, and `AggregatorFactoryBean`, and respective `CorrelationHandlerSpec` for DSL. This new option takes action only if a `discardChannel` is provided, and `sendPartialResultOnExpiry` is not set to `true`. When `discardIndividuallyOnExpiry` is false, the messages in the expired group are packed into a list for payload of a discarding single message. * Test and document the new feature --- .../AbstractCorrelatingMessageHandler.java | 31 +++++++++++++++++-- .../config/AggregatorFactoryBean.java | 17 ++++++++-- .../dsl/CorrelationHandlerSpec.java | 15 ++++++++- ...bstractCorrelatingMessageHandlerTests.java | 28 +++++++++++++++++ .../antora/modules/ROOT/pages/aggregator.adoc | 12 ++++--- .../antora/modules/ROOT/pages/whats-new.adoc | 6 ++++ 6 files changed, 99 insertions(+), 10 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index fa13a3af6e..68433fecd2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -18,6 +18,7 @@ package org.springframework.integration.aggregator; import java.time.Duration; import java.time.Instant; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Comparator; @@ -63,6 +64,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.core.DestinationResolutionException; +import org.springframework.messaging.support.GenericMessage; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; @@ -80,7 +82,7 @@ import org.springframework.util.ObjectUtils; * {@link ReleaseStrategy}, and {@link MessageGroupProcessor} implementations as * you require. *

- * By default the {@link CorrelationStrategy} will be a + * By default, the {@link CorrelationStrategy} will be a * {@link HeaderAttributeCorrelationStrategy} and the {@link ReleaseStrategy} will be a * {@link SequenceSizeReleaseStrategy}. *

@@ -129,6 +131,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private boolean sendPartialResultOnExpiry; + private boolean discardIndividuallyOnExpiry = true; + private boolean sequenceAware; private LockRegistry lockRegistry = new DefaultLockRegistry(); @@ -262,6 +266,18 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.sendPartialResultOnExpiry = sendPartialResultOnExpiry; } + /** + * Set to {@code false} to send to discard channel a whole expired group as a single message. + * This option makes sense only if {@link #sendPartialResultOnExpiry} is set to {@code false} (default). + * And also if {@link #discardChannel} is injected. + * @param discardIndividuallyOnExpiry false to discard the whole group as one message. + * @since 6.5 + * @see #sendPartialResultOnExpiry + */ + public void setDiscardIndividuallyOnExpiry(boolean discardIndividuallyOnExpiry) { + this.discardIndividuallyOnExpiry = discardIndividuallyOnExpiry; + } + /** * By default, when a MessageGroupStoreReaper is configured to expire partial * groups, empty groups are also removed. Empty groups exist after a group @@ -876,8 +892,17 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (this.releaseLockBeforeSend) { lock.unlock(); } - group.getMessages() - .forEach(this::discardMessage); + MessageChannel messageChannel = getDiscardChannel(); + if (messageChannel != null) { + if (this.discardIndividuallyOnExpiry) { + group.getMessages() + .forEach(this::discardMessage); + } + else { + List> messagesInGroupToDiscard = new ArrayList<>(group.getMessages()); + discardMessage(new GenericMessage<>(messagesInGroupToDiscard)); + } + } } if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent( diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java index f1f2cb7e13..72ec5b296d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2024 the original author or authors. + * Copyright 2015-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. @@ -86,6 +86,8 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe private Boolean sendPartialResultOnExpiry; + private Boolean discardIndividuallyOnExpiry; + private Long minimumTimeoutForEmptyGroups; private Boolean expireGroupsUponTimeout; @@ -195,6 +197,16 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe this.groupConditionSupplier = groupConditionSupplier; } + /** + * Set to {@code false} to send to discard channel a whole expired group as a single message. + * @param discardIndividuallyOnExpiry false to discard the whole group as one message. + * @since 6.5 + * @see org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler#setDiscardIndividuallyOnExpiry(boolean) + */ + public void setDiscardIndividuallyOnExpiry(Boolean discardIndividuallyOnExpiry) { + this.discardIndividuallyOnExpiry = discardIndividuallyOnExpiry; + } + @Override protected AggregatingMessageHandler createHandler() { MessageGroupProcessor outputProcessor; @@ -242,7 +254,8 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe .acceptIfNotNull(this.expireDuration, (duration) -> aggregator.setExpireDuration(Duration.ofMillis(duration))) .acceptIfNotNull(this.groupConditionSupplier, aggregator::setGroupConditionSupplier) - .acceptIfNotNull(this.expireTimeout, aggregator::setExpireTimeout); + .acceptIfNotNull(this.expireTimeout, aggregator::setExpireTimeout) + .acceptIfNotNull(this.discardIndividuallyOnExpiry, aggregator::setDiscardIndividuallyOnExpiry); return aggregator; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java index 739e1e31c4..a531fba07b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2022 the original author or authors. + * Copyright 2016-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. @@ -85,6 +85,19 @@ public abstract class CorrelationHandlerSpec group); + handler.setReleaseStrategy(group -> false); + QueueChannel discardChannel = new QueueChannel(); + handler.setDiscardChannel(discardChannel); + handler.setExpireTimeout(1); + handler.setDiscardIndividuallyOnExpiry(false); + + Message message1 = MessageBuilder.withPayload("test1").setCorrelationId("test").build(); + Message message2 = MessageBuilder.withPayload("test2").setCorrelationId("test").build(); + + handler.handleMessageInternal(message1); + handler.handleMessageInternal(message2); + + // Slight delay to let the group be treated as expired. + Thread.sleep(100); + + handler.purgeOrphanedGroups(); + + Message receive = discardChannel.receive(10000); + assertThat(receive) + .extracting(Message::getPayload) + .asInstanceOf(InstanceOfAssertFactories.LIST) + .containsOnly(message1, message2); + } + } diff --git a/src/reference/antora/modules/ROOT/pages/aggregator.adoc b/src/reference/antora/modules/ROOT/pages/aggregator.adoc index 06f726907d..298d0df328 100644 --- a/src/reference/antora/modules/ROOT/pages/aggregator.adoc +++ b/src/reference/antora/modules/ROOT/pages/aggregator.adoc @@ -213,6 +213,10 @@ Any new messages for this group are sent to the discard channel (if defined). Setting `expire-groups-upon-completion` to `true` (the default is `false`) removes the entire group, and any new messages (with the same correlation ID as the removed group) form a new group. You can release partial sequences by using a `MessageGroupStoreReaper` together with `send-partial-result-on-expiry` being set to `true`. +Starting with version 6.5, the correlation handler can also be configured with a `discardIndividuallyOnExpiry` option to discard the whole group as a single message. +Essentially, the payload of this message is a list of messages from the expired group. +Works only if `sendPartialResultOnExpiry` is set to `false` (default) and `dicardChannel` is provided. + IMPORTANT: To facilitate discarding of late-arriving messages, the aggregator must maintain state about the group after it has been released. This can eventually cause out-of-memory conditions. To avoid such situations, you should consider configuring a `MessageGroupStoreReaper` to remove the group metadata. @@ -519,7 +523,7 @@ Empty groups can be removed later by using a `MessageGroupStoreReaper` in combin `expire-groups-upon-completion` relates to "`normal`" completion when the `ReleaseStrategy` releases the group. This defaults to `false`. -If a group is not completed normally but is released or discarded because of a timeout, the group is normally expired. +If a group is not complete normally but is released or discarded because of a timeout, the group is normally expired. Since version 4.1, you can control this behavior by using `expire-groups-upon-timeout`. It defaults to `true` for backwards compatibility. @@ -531,12 +535,12 @@ Timed-out groups are either discarded or a partial release occurs (based on `sen Since version 5.0, empty groups are also scheduled for removal after `empty-group-min-timeout`. If `expireGroupsUponCompletion == false` and `minimumTimeoutForEmptyGroups > 0`, the task to remove the group is scheduled when normal or partial sequences release happens. -Starting with version 5.4, the aggregator (and resequencer) can be configured to expire orphaned groups (groups in a persistent message store that might not otherwise be released). +Starting with version 5.4, the aggregator (and resequencer) can be configured to expire orphaned groups (those in a persistent message store that might not otherwise be released). The `expireTimeout` (if greater than `0`) indicates that groups older than this value in the store should be purged. The `purgeOrphanedGroups()` method is called on start up and, together with the provided `expireDuration`, periodically within a scheduled task. This method is also can be called externally at any time. The expiration logic is fully delegated to the `forceComplete(MessageGroup)` functionality according to the provided expiration options mentioned above. -Such a periodic purge functionality is useful when a message store is needed to be cleaned up from those old groups which are not going to be released any more with regular message arrival logic. +Such a periodic purge functionality is useful when a message store is needed to be cleaned up from those old groups which are not going to be released anymore with regular message arrival logic. In most cases this happens after an application restart, when using a persistent message group store. The functionality is similar to the `MessageGroupStoreReaper` with a scheduled task, but provides a convenient way to deal with old groups within specific components, when using group timeout instead of a reaper. The `MessageGroupStore` must be provided exclusively for the current correlation endpoint. @@ -695,7 +699,7 @@ Otherwise, it is discarded. There is a difference between `groupTimeout` behavior and `MessageGroupStoreReaper` (see xref:aggregator.adoc#aggregator-xml[Configuring an Aggregator with XML]). The reaper initiates forced completion for all `MessageGroup` s in the `MessageGroupStore` periodically. The `groupTimeout` does it for each `MessageGroup` individually if a new message does not arrive during the `groupTimeout`. -Also, the reaper can be used to remove empty groups (empty groups are retained in order to discard late messages if `expire-groups-upon-completion` is false). +Also, the reaper can be used to remove empty groups (those retained in order to discard late messages if `expire-groups-upon-completion` is false). Starting with version 5.5, the `groupTimeoutExpression` can be evaluated to a `java.util.Date` instance. This can be useful in cases like determining a scheduled task moment based on the group creation time (`MessageGroup.getTimestamp()`) instead of a current message arrival as it is calculated when `groupTimeoutExpression` is evaluated to `long`: diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 2f44836489..f73d778a40 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -27,6 +27,12 @@ The `AbstractCorrelatingMessageHandler` does not throw an `IllegalArgumentExcept Instead, such a collection is wrapped into a single reply message. See xref:aggregator.adoc[Aggregator] for more information. +[[x6.4-correlation-changes]] +== The `discardIndividuallyOnExpiry` Option For Correlation Handlers + +The aggregator and resequencer can now discard the whole expired group as a single message via setting `discardIndividuallyOnExpiry` to `false`. +See xref:aggregator.adoc#releasestrategy[ReleaseStrategy] for more information. + [[x6.4-message-store-with-locks]] == The `LockRegistry` in the `MessageStore`