From 929505e121237b4e6ceaf647d4c250fb6c1800bc Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 14 May 2018 17:18:19 -0400 Subject: [PATCH] INT-4466: Do not force release groups if no match JIRA: https://jira.spring.io/browse/INT-4466 When we schedule group for force complete in the `AbstractCorrelatingMessageHandler`, we don't track a group `timestamp` and its `lastModified` before the scheduled task. This way, in the cluster environment, we may schedule several tasks for different messages and the first started may release the group too early. Just because we extract a `MessageGroup` from the store already in the task per se. * Propagate the actual `timestamp` and `lastModified` from group before scheduling task. Compare these value with the actual group metadata in the `processForceRelease()` before performing real `forceRelease()`. This way we restore behavior before fixing memory leak, when we propagated full `MessageGroup` to the scheduled task * Implement `ResequencingMessageHandler.getComponentType()` for consistency * Remove unnecessary overhead with the `volatile` on many `AbstractCorrelatingMessageHandler` properties, which hardly ever can be changed at runtime **Cherry-pick to 5.0.x and 4.3.x** --- .../AbstractCorrelatingMessageHandler.java | 45 +++++++++++-------- .../ResequencingMessageHandler.java | 7 ++- 2 files changed, 32 insertions(+), 20 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 8ff9370bb4..7d199cb9f8 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -88,6 +88,7 @@ import org.springframework.util.CollectionUtils; * @author David Liu * @author Enrique Rodriguez * @author Meherzad Lahewala + * * @since 2.0 */ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler @@ -103,41 +104,41 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private MessageGroupProcessor outputProcessor; - private volatile MessageGroupStore messageStore; + private MessageGroupStore messageStore; - private volatile CorrelationStrategy correlationStrategy; + private CorrelationStrategy correlationStrategy; - private volatile ReleaseStrategy releaseStrategy; + private ReleaseStrategy releaseStrategy; - private volatile boolean releaseStrategySet; + private boolean releaseStrategySet; - private volatile MessageChannel discardChannel; + private MessageChannel discardChannel; - private volatile String discardChannelName; + private String discardChannelName; private boolean sendPartialResultOnExpiry = false; - private volatile boolean sequenceAware = false; + private boolean sequenceAware = false; - private volatile LockRegistry lockRegistry = new DefaultLockRegistry(); + private LockRegistry lockRegistry = new DefaultLockRegistry(); private boolean lockRegistrySet = false; - private volatile long minimumTimeoutForEmptyGroups; + private long minimumTimeoutForEmptyGroups; - private volatile boolean releasePartialSequences; + private boolean releasePartialSequences; - private volatile Expression groupTimeoutExpression; + private Expression groupTimeoutExpression; - private volatile List forceReleaseAdviceChain; + private List forceReleaseAdviceChain; private MessageGroupProcessor forceReleaseProcessor = new ForceReleaseMessageGroupProcessor(); private EvaluationContext evaluationContext; - private volatile ApplicationEventPublisher applicationEventPublisher; + private ApplicationEventPublisher applicationEventPublisher; - private volatile boolean expireGroupsUponTimeout = true; + private boolean expireGroupsUponTimeout = true; private volatile boolean running; @@ -436,7 +437,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (this.releaseStrategy.canRelease(messageGroup)) { Collection> completedMessages = null; try { - completedMessages = this.completeGroup(message, correlationKey, messageGroup); + completedMessages = completeGroup(message, correlationKey, messageGroup); } finally { // Possible clean (implementation dependency) up @@ -520,10 +521,12 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (groupTimeout != null && groupTimeout >= 0) { if (groupTimeout > 0) { final Object groupId = messageGroup.getGroupId(); + final long timestamp = messageGroup.getTimestamp(); + final long lastModified = messageGroup.getLastModified(); ScheduledFuture scheduledFuture = getTaskScheduler() .schedule(() -> { try { - processForceRelease(groupId); + processForceRelease(groupId, timestamp, lastModified); } catch (MessageDeliveryException e) { if (AbstractCorrelatingMessageHandler.this.logger.isWarnEnabled()) { @@ -550,9 +553,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP scheduleGroupToForceComplete(messageGroup); } - private void processForceRelease(Object groupId) { + private void processForceRelease(Object groupId, long timestamp, long lastModified) { MessageGroup messageGroup = this.messageStore.getMessageGroup(groupId); - this.forceReleaseProcessor.processMessageGroup(messageGroup); + if (messageGroup.getTimestamp() == timestamp && messageGroup.getLastModified() == lastModified) { + this.forceReleaseProcessor.processMessageGroup(messageGroup); + } } private void discardMessage(Message message) { @@ -616,9 +621,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } long lastModifiedNow = groupNow.getLastModified(); int groupSize = groupNow.size(); + if ((!groupNow.isComplete() || groupSize == 0) && group.getLastModified() == lastModifiedNow && group.getTimestamp() == groupNow.getTimestamp()) { + if (groupSize > 0) { if (this.releaseStrategy.canRelease(groupNow)) { completeGroup(correlationKey, groupNow); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java index c60ffebb86..cf5994b9a6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -66,6 +66,11 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle super.setExpireGroupsUponTimeout(expireGroupsUponTimeout); } + @Override + public String getComponentType() { + return "resequencer"; + } + @Override protected boolean shouldCopyRequestHeaders() { return false;