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 8413b3314d..13643b6143 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 @@ -46,7 +46,6 @@ import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.DiscardingMessageHandler; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; -import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; @@ -88,7 +87,7 @@ import org.springframework.util.CollectionUtils; public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler implements DiscardingMessageHandler, DisposableBean, ApplicationEventPublisherAware { - private static final Log logger = LogFactory.getLog(AbstractCorrelatingMessageHandler.class); + protected final Log logger = LogFactory.getLog(getClass()); private final Comparator> sequenceNumberComparator = new SequenceNumberComparator(); @@ -102,6 +101,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private volatile ReleaseStrategy releaseStrategy; + private volatile boolean releaseStrategySet; + private volatile MessageChannel discardChannel; private volatile String discardChannelName; @@ -140,7 +141,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.correlationStrategy = (correlationStrategy == null ? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID) : correlationStrategy); - this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy; + this.releaseStrategy = releaseStrategy == null ? new SimpleSequenceSizeReleaseStrategy() : releaseStrategy; this.sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy; } @@ -161,12 +162,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP public final void setMessageStore(MessageGroupStore store) { this.messageStore = store; - store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { - @Override - public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { - AbstractCorrelatingMessageHandler.this.forceReleaseProcessor.processMessageGroup(group); - } - }); + store.registerMessageGroupExpiryCallback( + (messageGroupStore, group) -> + AbstractCorrelatingMessageHandler.this.forceReleaseProcessor.processMessageGroup(group)); } public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) { @@ -178,6 +176,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP Assert.notNull(releaseStrategy); this.releaseStrategy = releaseStrategy; this.sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy; + this.releaseStrategySet = true; } public void setGroupTimeoutExpression(Expression groupTimeoutExpression) { @@ -228,14 +227,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (this.releasePartialSequences) { Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy, "Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName() + - "] cannot release partial sequences. Use the default SequenceSizeReleaseStrategy instead."); - ((SequenceSizeReleaseStrategy) this.releaseStrategy).setReleasePartialSequences(this.releasePartialSequences); + "] cannot release partial sequences. Use a SequenceSizeReleaseStrategy instead."); + ((SequenceSizeReleaseStrategy) this.releaseStrategy) + .setReleasePartialSequences(this.releasePartialSequences); } if (this.evaluationContext == null) { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); } + if (this.sequenceAware) { + this.logger.warn("Using a SequenceSizeReleaseStrategy with large groups may not perform well, consider " + + "using a SimpleSequenceSizeReleaseStrategy"); + } + /* * Disallow any further changes to the lock registry * (checked in the setter). @@ -287,11 +292,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } /** - * Set {@code releasePartialSequences} on an underlying - * {@link SequenceSizeReleaseStrategy}. + * Set {@code releasePartialSequences} on an underlying default + * {@link SequenceSizeReleaseStrategy}. Ignored for other release strategies. * @param releasePartialSequences true to allow release. */ public void setReleasePartialSequences(boolean releasePartialSequences) { + if (!this.releaseStrategySet && releasePartialSequences) { + setReleaseStrategy(new SequenceSizeReleaseStrategy()); + } this.releasePartialSequences = releasePartialSequences; } @@ -384,8 +392,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP Object correlationKey = this.correlationStrategy.getCorrelationKey(message); Assert.state(correlationKey != null, "Null correlation not allowed. Maybe the CorrelationStrategy is failing?"); - if (logger.isDebugEnabled()) { - logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message); + if (this.logger.isDebugEnabled()) { + this.logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message); } UUID groupIdUuid = UUIDConverter.getUUID(correlationKey); @@ -396,8 +404,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP ScheduledFuture scheduledFuture = this.expireGroupScheduledFutures.remove(groupIdUuid); if (scheduledFuture != null) { boolean canceled = scheduledFuture.cancel(true); - if (canceled && logger.isDebugEnabled()) { - logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ " + if (canceled && this.logger.isDebugEnabled()) { + this.logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ " + correlationKey + "]."); } } @@ -407,8 +415,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } if (!messageGroup.isComplete() && messageGroup.canAdd(message)) { - if (logger.isTraceEnabled()) { - logger.trace("Adding message to group [ " + messageGroup + "]"); + if (this.logger.isTraceEnabled()) { + this.logger.trace("Adding message to group [ " + messageGroup + "]"); } messageGroup = this.store(correlationKey, message); @@ -446,26 +454,21 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (groupTimeout > 0) { final Object groupId = messageGroup.getGroupId(); ScheduledFuture scheduledFuture = getTaskScheduler() - .schedule(new Runnable() { - - @Override - public void run() { - try { - processForceRelease(groupId); - } - catch (MessageDeliveryException e) { - if (logger.isDebugEnabled()) { - logger.debug("The MessageGroup [ " + groupId + - "] is rescheduled by the reason: " + e.getMessage()); - } - scheduleGroupToForceComplete(groupId); - } + .schedule(() -> { + try { + processForceRelease(groupId); + } + catch (MessageDeliveryException e) { + if (AbstractCorrelatingMessageHandler.this.logger.isDebugEnabled()) { + AbstractCorrelatingMessageHandler.this.logger.debug("The MessageGroup [ " + + groupId + "] is rescheduled by the reason: " + e.getMessage()); + } + scheduleGroupToForceComplete(groupId); } - }, new Date(System.currentTimeMillis() + groupTimeout)); - if (logger.isDebugEnabled()) { - logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'."); + if (this.logger.isDebugEnabled()) { + this.logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'."); } this.expireGroupScheduledFutures.put(UUIDConverter.getUUID(groupId), scheduledFuture); } @@ -520,8 +523,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.expireGroupScheduledFutures.remove(UUIDConverter.getUUID(correlationKey)); if (scheduledFuture != null) { boolean canceled = scheduledFuture.cancel(false); - if (canceled && logger.isDebugEnabled()) { - logger.debug("Cancel 'forceComplete' scheduling for MessageGroup [ " + group + "]."); + if (canceled && this.logger.isDebugEnabled()) { + this.logger.debug("Cancel 'forceComplete' scheduling for MessageGroup [ " + group + "]."); } } MessageGroup groupNow = group; @@ -568,23 +571,23 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP * setting minimumTimeoutForEmptyGroups. */ removeGroup = lastModifiedNow <= (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups); - if (removeGroup && logger.isDebugEnabled()) { - logger.debug("Removing empty group: " + correlationKey); + if (removeGroup && this.logger.isDebugEnabled()) { + this.logger.debug("Removing empty group: " + correlationKey); } } } else { removeGroup = false; - if (logger.isDebugEnabled()) { - logger.debug("Group expiry candidate (" + correlationKey + + if (this.logger.isDebugEnabled()) { + this.logger.debug("Group expiry candidate (" + correlationKey + ") has changed - it may be reconsidered for a future expiration"); } } } catch (MessageDeliveryException e) { removeGroup = false; - if (logger.isDebugEnabled()) { - logger.debug("Group expiry candidate (" + correlationKey + + if (this.logger.isDebugEnabled()) { + this.logger.debug("Group expiry candidate (" + correlationKey + ") has been affected by MessageDeliveryException - " + "it may be reconsidered for a future expiration one more time"); } @@ -603,7 +606,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } catch (InterruptedException ie) { Thread.currentThread().interrupt(); - logger.debug("Thread was interrupted while trying to obtain lock"); + this.logger.debug("Thread was interrupted while trying to obtain lock"); } } @@ -622,19 +625,19 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } protected void expireGroup(Object correlationKey, MessageGroup group) { - if (logger.isInfoEnabled()) { - logger.info("Expiring MessageGroup with correlationKey[" + correlationKey + "]"); + if (this.logger.isInfoEnabled()) { + this.logger.info("Expiring MessageGroup with correlationKey[" + correlationKey + "]"); } if (this.sendPartialResultOnExpiry) { - if (logger.isDebugEnabled()) { - logger.debug("Prematurely releasing partially complete group with key [" + if (this.logger.isDebugEnabled()) { + this.logger.debug("Prematurely releasing partially complete group with key [" + correlationKey + "] to: " + getOutputChannel()); } completeGroup(correlationKey, group); } else { - if (logger.isDebugEnabled()) { - logger.debug("Discarding messages of partially complete group with key [" + if (this.logger.isDebugEnabled()) { + this.logger.debug("Discarding messages of partially complete group with key [" + correlationKey + "] to: " + (this.discardChannelName != null ? this.discardChannelName : this.discardChannel)); } @@ -658,8 +661,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP @SuppressWarnings("unchecked") protected Collection> completeGroup(Message message, Object correlationKey, MessageGroup group) { - if (logger.isDebugEnabled()) { - logger.debug("Completing group with correlationKey [" + correlationKey + "]"); + if (this.logger.isDebugEnabled()) { + this.logger.debug("Completing group with correlationKey [" + correlationKey + "]"); } Object result = this.outputProcessor.processMessageGroup(group); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java index 8f7668ef01..6e2a8aa8e0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java @@ -28,8 +28,9 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.messaging.Message; /** - * An implementation of {@link ReleaseStrategy} that simply compares the current size of the message list to the - * expected 'sequenceSize'. + * An implementation of {@link ReleaseStrategy} that simply compares the current size of + * the message list to the expected 'sequenceSize'. Supports release of partial sequences. + * Correlating message handlers prevent the addition of duplicate sequences to the group. * * @author Mark Fisher * @author Marius Bogoevici @@ -47,10 +48,19 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { private volatile boolean releasePartialSequences; + /** + * Construct an instance that does not support releasing partial sequences. + */ public SequenceSizeReleaseStrategy() { this(false); } + /** + * Construct an instance that supports releasing partial sequences if + * releasePartialSequences is true. This can be an expensive operation on large + * groups. + * @param releasePartialSequences true to allow the release of partial sequences. + */ public SequenceSizeReleaseStrategy(boolean releasePartialSequences) { this.releasePartialSequences = releasePartialSequences; } @@ -59,6 +69,7 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { * Flag that determines if partial sequences are allowed. If true then as soon as * enough messages arrive that can be ordered they will be released, provided they * all have sequence numbers greater than those already released. + * This can be an expensive operation for large groups. * @param releasePartialSequences true when partial sequences should be released. */ public void setReleasePartialSequences(boolean releasePartialSequences) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleSequenceSizeReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleSequenceSizeReleaseStrategy.java new file mode 100644 index 0000000000..511c5bc2d3 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleSequenceSizeReleaseStrategy.java @@ -0,0 +1,37 @@ +/* + * Copyright 2016 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aggregator; + +import org.springframework.integration.store.MessageGroup; + +/** + * An implementation of {@link ReleaseStrategy} that simply compares the current size of + * the message list to the expected 'sequenceSize'. It does not support releasing partial + * sequences. Correlating message handlers using this strategy do not check for duplicate + * sequence numbers. + * @author Gary Russell + * @since 4.3.4 + * + */ +public class SimpleSequenceSizeReleaseStrategy implements ReleaseStrategy { + + @Override + public boolean canRelease(MessageGroup group) { + return group.getSequenceSize() == group.size(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index cf5b8b3e5f..09fa110a0a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -145,6 +145,7 @@ public interface MessageGroupStore extends BasicMessageGroupStore { /** * Invoked when a MessageGroupStore expires a group. */ + @FunctionalInterface interface MessageGroupCallback { void execute(MessageGroupStore messageGroupStore, MessageGroup group); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 790cdb2699..5a61e7ab34 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -415,6 +415,7 @@ public class AggregatorTests { Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); Message message4 = createMessage(7, "ABC", 3, 3, replyChannel, null); + this.aggregator.setReleaseStrategy(new SequenceSizeReleaseStrategy()); this.aggregator.handleMessage(message1); this.aggregator.handleMessage(message3); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index 646df9d75d..35c1a2715a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -284,6 +284,7 @@ public class ResequencerTests { Message message2 = createMessage("456", "ABC", 5, 1, null); this.resequencer.setSendPartialResultOnExpiry(false); this.resequencer.setDiscardChannel(discardChannel); + this.resequencer.setReleasePartialSequences(true); // force SequenceSizeReleaseStrategy this.resequencer.handleMessage(message1); this.resequencer.handleMessage(message2); // this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index 771ce25cd2..769d277fc9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -33,7 +33,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy; import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy; -import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy; +import org.springframework.integration.aggregator.SimpleSequenceSizeReleaseStrategy; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.test.util.TestUtils; @@ -52,7 +52,7 @@ public class AggregatorAnnotationTests { new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithDefaultAnnotation"; MessageHandler aggregator = this.getAggregator(context, endpointName); - assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SequenceSizeReleaseStrategy); + assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SimpleSequenceSizeReleaseStrategy); assertNull(getPropertyValue(aggregator, "outputChannel")); assertTrue(getPropertyValue(aggregator, "discardChannel") instanceof NullChannel); assertEquals(-1L, getPropertyValue(aggregator, "messagingTemplate.sendTimeout")); @@ -65,7 +65,7 @@ public class AggregatorAnnotationTests { new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithCustomizedAnnotation"; MessageHandler aggregator = this.getAggregator(context, endpointName); - assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SequenceSizeReleaseStrategy); + assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SimpleSequenceSizeReleaseStrategy); assertEquals("outputChannel", getPropertyValue(aggregator, "outputChannelName")); assertEquals("discardChannel", getPropertyValue(aggregator, "discardChannelName")); assertEquals(98765432L, getPropertyValue(aggregator, "messagingTemplate.sendTimeout")); diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 94842307ec..6ff8e7dae8 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -35,7 +35,7 @@ The Aggregation API consists of a number of classes: * The interface `MessageGroupProcessor`, and its subclasses:`MethodInvokingAggregatingMessageGroupProcessor` and `ExpressionEvaluatingMessageGroupProcessor` -* The `ReleaseStrategy` interface and its default implementation `SequenceSizeReleaseStrategy` +* The `ReleaseStrategy` interface and its default implementation `SimpleSequenceSizeReleaseStrategy` * The `CorrelationStrategy` interface and its default implementation `HeaderAttributeCorrelationStrategy` @@ -85,7 +85,7 @@ public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, Messag ... this.correlationStrategy = correlationStrategy == null ? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID) : correlationStrategy; - this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy; + this.releaseStrategy = releaseStrategy == null ? new SimpleSequenceSizeReleaseStrategy() : releaseStrategy; ... } ---- @@ -214,10 +214,13 @@ This can eventually cause out of memory conditions. To avoid such situations, you should consider configuring a `MessageGroupStoreReaper` to remove the group metadata; the expiry parameters should be set to expire groups after it is not expected that late messages will arrive. For information about configuring a reaper, see <>. -Spring Integration provides an out-of-the box implementation for `ReleaseStrategy`, the `SequenceSizeReleaseStrategy`. +Spring Integration provides an out-of-the box implementation for `ReleaseStrategy`, the `SimpleSequenceSizeReleaseStrategy`. This implementation consults the `SEQUENCE_NUMBER` and `SEQUENCE_SIZE` headers of each arriving message to decide when a message group is complete and ready to be aggregated. As shown above, it is also the default strategy. +NOTE: Before _version 5.0_, the default release strategy was `SequenceSizeReleaseStrategy` which does not perform well with large groups. +With that strategy, duplicate sequence numbers are detected and rejected; this operation can be expensive. + ===== CorrelationStrategy The `CorrelationStrategy` interface is defined as follows: @@ -263,7 +266,7 @@ Below you can see an example of an aggregator. ---- - + auto-startup="true" <2> input-channel="inputChannel" <3> output-channel="outputChannel" <4> @@ -680,7 +683,7 @@ Must be specified if this class will be used as an aggregator. <2> An annotation indicating that this method shall be used as the release strategy of an aggregator. -If not present on any method, the aggregator will use the SequenceSizeReleaseStrategy. +If not present on any method, the aggregator will use the `SimpleSequenceSizeReleaseStrategy`. diff --git a/src/reference/asciidoc/resequencer.adoc b/src/reference/asciidoc/resequencer.adoc index c77368c6ea..32ec905123 100644 --- a/src/reference/asciidoc/resequencer.adoc +++ b/src/reference/asciidoc/resequencer.adoc @@ -13,6 +13,9 @@ It simply releases them in the order of their `SEQUENCE_NUMBER` header values. With respect to that, the user might opt to release all messages at once (after the whole sequence, according to the `SEQUENCE_SIZE`, has been released), or as soon as a valid sequence is available. +IMPORTANT: The resequencer is intended to resequence relatively short sequences of messages with small gaps. +If you have a large number of disjoint sequences with many gaps, you may experience performance issues. + ==== Configuring a Resequencer Configuring a resequencer requires only including the appropriate element in XML. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 40b3b8eb56..97e08db496 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -65,6 +65,7 @@ See <> for more information. The AMQP outbound endpoints now support setting a delay expression for when using the RabbitMQ Delayed Message Exchange plugin. See <> for more information. + The inbound endpoints now support the Spring AMQP `DirectMessageListenerContainer`. See <> for more information. @@ -72,3 +73,8 @@ See <> for more information. The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`. See <> for more information. + +==== Aggregator Performance Changes + +Aggregators now use a `SimpleSequenceSizeReleaseStrategy` by default, which is more efficient, especially with large groups. +See <> for more information.