INT-4452: Expire immediately when group timeout<0

JIRA: https://jira.spring.io/browse/INT-4452

This is pretty typical in practice to get a `groupTimeout` to be
evaluated to a negative value: some business decisions, sensitive
calculations and so on.

* Treat any non-positive `groupTimeout` as a signal to expire group
immediately without scheduling.
Only `null` is considered as a signal do nothing for the current message
* Polishing for some tests for better performance - saves some execution
time

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -100,7 +100,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private final Map<UUID, ScheduledFuture<?>> expireGroupScheduledFutures = new ConcurrentHashMap<>();
 
-	private final Set<Object> groupIds =  ConcurrentHashMap.newKeySet();
+	private final Set<Object> groupIds = ConcurrentHashMap.newKeySet();
 
 	private MessageGroupProcessor outputProcessor;
 
@@ -144,15 +144,23 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
 			CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
+
 		Assert.notNull(processor, "'processor' must not be null");
 		Assert.notNull(store, "'store' must not be null");
 
 		setMessageStore(store);
 		this.outputProcessor = processor;
-		this.correlationStrategy = (correlationStrategy == null
-				? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID)
-				: correlationStrategy);
-		this.releaseStrategy = releaseStrategy == null ? new SimpleSequenceSizeReleaseStrategy() : releaseStrategy;
+
+		this.correlationStrategy =
+				correlationStrategy == null
+						? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID)
+						: correlationStrategy;
+
+		this.releaseStrategy =
+				releaseStrategy == null
+						? new SimpleSequenceSizeReleaseStrategy()
+						: releaseStrategy;
+
 		this.releaseStrategySet = releaseStrategy != null;
 		this.sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy;
 	}
@@ -195,7 +203,7 @@ public void setGroupTimeoutExpression(Expression groupTimeoutExpression) {
 	}
 
 	public void setForceReleaseAdviceChain(List<Advice> forceReleaseAdviceChain) {
-		Assert.notNull(forceReleaseAdviceChain, "forceReleaseAdviceChain must not be null");
+		Assert.notNull(forceReleaseAdviceChain, "'forceReleaseAdviceChain' must not be null");
 		this.forceReleaseAdviceChain = forceReleaseAdviceChain;
 	}
 
@@ -242,7 +250,7 @@ protected void onInit() throws Exception {
 		}
 
 		if (this.releasePartialSequences) {
-			Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy,
+			Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy, () ->
 					"Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName() +
 							"] cannot release partial sequences. Use a SequenceSizeReleaseStrategy instead.");
 			((SequenceSizeReleaseStrategy) this.releaseStrategy)
@@ -519,7 +527,7 @@ private void scheduleGroupToForceComplete(MessageGroup messageGroup) {
 		 * When 'groupTimeout' is evaluated to 'null' we do nothing.
 		 * The 'MessageGroupStoreReaper' can be used to 'forceComplete' message groups.
 		 */
-		if (groupTimeout != null && groupTimeout >= 0) {
+		if (groupTimeout != null) {
 			if (groupTimeout > 0) {
 				final Object groupId = messageGroup.getGroupId();
 				final long timestamp = messageGroup.getTimestamp();

spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 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.
@@ -47,6 +47,7 @@
  * @author Mark Fisher
  * @author Marius Bogoevici
  * @author Iwein Fuld
+ * @author Artem Bilan
  */
 public class ConcurrentAggregatorTests {
 
@@ -90,8 +91,7 @@ public void testCompleteGroupWithinTimeout() throws InterruptedException {
 	@Test
 	@Ignore
 	// dropped backwards compatibility for duplicate ID's
-	public void testCompleteGroupWithinTimeoutWithSameId()
-			throws InterruptedException {
+	public void testCompleteGroupWithinTimeoutWithSameId() {
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel,
 				"ID#1");
@@ -111,8 +111,8 @@ public void testCompleteGroupWithinTimeoutWithSameId()
 	}
 
 	@Test
-	public void testShouldNotSendPartialResultOnTimeoutByDefault()
-			throws InterruptedException {
+	public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException {
+
 		QueueChannel discardChannel = new QueueChannel();
 		this.aggregator.setDiscardChannel(discardChannel);
 		QueueChannel replyChannel = new QueueChannel();
@@ -126,17 +126,16 @@ public void testShouldNotSendPartialResultOnTimeoutByDefault()
 
 		assertEquals("Task should have completed within timeout", 0, latch
 				.getCount());
-		Message<?> reply = replyChannel.receive(1000);
+		Message<?> reply = replyChannel.receive(10);
 		assertNull("No message should have been sent normally", reply);
 		this.store.expireMessageGroups(-10000);
-		Message<?> discardedMessage = discardChannel.receive(1000);
+		Message<?> discardedMessage = discardChannel.receive(10000);
 		assertNotNull("A message should have been discarded", discardedMessage);
 		assertEquals(message, discardedMessage);
 	}
 
 	@Test
-	public void testShouldSendPartialResultOnTimeoutTrue()
-			throws InterruptedException {
+	public void testShouldSendPartialResultOnTimeoutTrue() throws InterruptedException {
 		this.aggregator.setSendPartialResultOnExpiry(true);
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
@@ -252,16 +251,14 @@ public void testTrackedCorrelationIdsCapacityPassesLimit() {
 	}
 
 	@Test(expected = MessageHandlingException.class)
-	public void testExceptionThrownIfNoCorrelationId()
-			throws InterruptedException {
+	public void testExceptionThrownIfNoCorrelationId() {
 		Message<?> message = createMessage(3, null, 2, 1, new QueueChannel(),
 				null);
 		this.aggregator.handleMessage(message);
 	}
 
 	@Test
-	public void testAdditionalMessageAfterCompletion()
-			throws InterruptedException {
+	public void testAdditionalMessageAfterCompletion() throws InterruptedException {
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
 		Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
@@ -291,6 +288,7 @@ public void testAdditionalMessageAfterCompletion()
 	private static Message<?> createMessage(Object payload,
 			Object correlationId, int sequenceSize, int sequenceNumber,
 			MessageChannel replyChannel, String predefinedId) {
+
 		MessageBuilder<Object> builder = MessageBuilder.withPayload(payload)
 				.setCorrelationId(correlationId).setSequenceSize(sequenceSize)
 				.setSequenceNumber(sequenceNumber)

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml

@@ -42,7 +42,7 @@
 
 	<aggregator input-channel="groupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"
 				send-partial-result-on-expiry="true"
-				group-timeout-expression="size() ge 2 ? 100 : -1"
+				group-timeout-expression="size() ge 2 ? 100 : null"
 			    release-strategy-expression="messages[0].headers.sequenceNumber == messages[0].headers.sequenceSize"/>
 
 	<aggregator input-channel="zeroGroupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 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.
@@ -45,8 +45,7 @@
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 
 /**
  * @author Iwein Fuld
@@ -55,8 +54,7 @@
  * @author Artem Bilan
  * @author Gary Russell
  */
-@RunWith(SpringJUnit4ClassRunner.class)
-@ContextConfiguration
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class AggregatorIntegrationTests {
 
@@ -91,29 +89,29 @@ public class AggregatorIntegrationTests {
 	private QueueChannel errors;
 
 	@Test
-	public void testVanillaAggregation() throws Exception {
+	public void testVanillaAggregation() {
 		for (int i = 0; i < 5; i++) {
 			Map<String, Object> headers = stubHeaders(i, 5, 1);
-			input.send(new GenericMessage<Integer>(i, headers));
+			input.send(new GenericMessage<>(i, headers));
 		}
 		Message<?> receive = output.receive(10000);
 		assertNotNull(receive);
 		assertEquals(1 + 2 + 3 + 4, receive.getPayload());
 	}
 
 	@Test
-	public void testNonExpiringAggregator() throws Exception {
+	public void testNonExpiringAggregator() {
 		for (int i = 0; i < 5; i++) {
 			Map<String, Object> headers = stubHeaders(i, 5, 1);
-			nonExpiringAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+			nonExpiringAggregatorInput.send(new GenericMessage<>(i, headers));
 		}
 		assertNotNull(output.receive(0));
 
 		assertNull(discard.receive(0));
 
 		for (int i = 5; i < 10; i++) {
 			Map<String, Object> headers = stubHeaders(i, 5, 1);
-			nonExpiringAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+			nonExpiringAggregatorInput.send(new GenericMessage<>(i, headers));
 		}
 		assertNull(output.receive(0));
 
@@ -125,18 +123,18 @@ public void testNonExpiringAggregator() throws Exception {
 	}
 
 	@Test
-	public void testExpiringAggregator() throws Exception {
+	public void testExpiringAggregator() {
 		for (int i = 0; i < 5; i++) {
 			Map<String, Object> headers = stubHeaders(i, 5, 1);
-			expiringAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+			expiringAggregatorInput.send(new GenericMessage<>(i, headers));
 		}
 		assertNotNull(output.receive(0));
 
 		assertNull(discard.receive(0));
 
 		for (int i = 5; i < 10; i++) {
 			Map<String, Object> headers = stubHeaders(i, 5, 1);
-			expiringAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+			expiringAggregatorInput.send(new GenericMessage<>(i, headers));
 		}
 		assertNotNull(output.receive(0));
 
@@ -148,7 +146,7 @@ public void testExpiringAggregator() throws Exception {
 	public void testGroupTimeoutScheduling() throws Exception {
 		for (int i = 0; i < 5; i++) {
 			Map<String, Object> headers = stubHeaders(i, 5, 1);
-			this.groupTimeoutAggregatorInput.send(new GenericMessage<Integer>(i, headers));
+			this.groupTimeoutAggregatorInput.send(new GenericMessage<>(i, headers));
 
 			//Wait until 'group-timeout' does its stuff.
 			MessageGroupStore mgs = TestUtils.getPropertyValue(this.context.getBean("gta.handler"), "messageStore",
@@ -166,11 +164,11 @@ public void testGroupTimeoutScheduling() throws Exception {
 	@Test
 	public void testGroupTimeoutReschedulingOnMessageDeliveryException() throws Exception {
 		for (int i = 0; i < 5; i++) {
-			this.output.send(new GenericMessage<String>("fake message"));
+			this.output.send(new GenericMessage<>("fake message"));
 		}
 
 		Map<String, Object> headers = stubHeaders(1, 2, 1);
-		this.groupTimeoutAggregatorInput.send(new GenericMessage<Integer>(1, headers));
+		this.groupTimeoutAggregatorInput.send(new GenericMessage<>(1, headers));
 
 		//Wait until 'group-timeout' does its stuff.
 		MessageGroupStore mgs = TestUtils.getPropertyValue(this.context.getBean("gta.handler"), "messageStore",
@@ -190,56 +188,56 @@ public void testGroupTimeoutReschedulingOnMessageDeliveryException() throws Exce
 	}
 
 	@Test
-	public void testGroupTimeoutExpressionScheduling() throws Exception {
-		// Since group-timeout-expression="size() >= 2 ? 100 : -1". The first message won't be scheduled to 'forceComplete'
-		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(1, stubHeaders(1, 6, 1)));
+	public void testGroupTimeoutExpressionScheduling() {
+		// Since group-timeout-expression="size() >= 2 ? 100 : null". The first message won't be scheduled to 'forceComplete'
+		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<>(1, stubHeaders(1, 6, 1)));
 		assertNull(this.output.receive(0));
 		assertNull(this.discard.receive(0));
 
 		// As far as 'group.size() >= 2' it will be scheduled to 'forceComplete'
-		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(2, stubHeaders(2, 6, 1)));
+		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<>(2, stubHeaders(2, 6, 1)));
 		assertNull(this.output.receive(0));
 		Message<?> receive = this.output.receive(10000);
 		assertNotNull(receive);
 		assertEquals(2, ((Collection<?>) receive.getPayload()).size());
 		assertNull(this.discard.receive(0));
 
 		// The same with these three messages
-		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(3, stubHeaders(3, 6, 1)));
+		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<>(3, stubHeaders(3, 6, 1)));
 		assertNull(this.output.receive(0));
 		assertNull(this.discard.receive(0));
 
-		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(4, stubHeaders(4, 6, 1)));
+		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<>(4, stubHeaders(4, 6, 1)));
 		assertNull(this.output.receive(0));
 		assertNull(this.discard.receive(0));
 
-		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(5, stubHeaders(5, 6, 1)));
+		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<>(5, stubHeaders(5, 6, 1)));
 		assertNull(this.output.receive(0));
 		receive = this.output.receive(10000);
 		assertNotNull(receive);
 		assertEquals(3, ((Collection<?>) receive.getPayload()).size());
 		assertNull(this.discard.receive(0));
 
 		// The last message in the sequence - normal release by provided 'ReleaseStrategy'
-		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(6, stubHeaders(6, 6, 1)));
+		this.groupTimeoutExpressionAggregatorInput.send(new GenericMessage<>(6, stubHeaders(6, 6, 1)));
 		receive = this.output.receive(10000);
 		assertNotNull(receive);
 		assertEquals(1, ((Collection<?>) receive.getPayload()).size());
 		assertNull(this.discard.receive(0));
 	}
 
 	@Test
-	public void testZeroGroupTimeoutExpressionScheduling() throws Exception {
+	public void testZeroGroupTimeoutExpressionScheduling() {
 		try {
 			this.output.purge(null);
 			this.errors.purge(null);
-			GenericMessage<String> message = new GenericMessage<String>("foo");
+			GenericMessage<String> message = new GenericMessage<>("foo");
 			this.output.send(message);
 			this.output.send(message);
 			this.output.send(message);
 			this.output.send(message);
 			this.output.send(message);
-			this.zeroGroupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(1, stubHeaders(1, 2, 1)));
+			this.zeroGroupTimeoutExpressionAggregatorInput.send(new GenericMessage<>(1, stubHeaders(1, 2, 1)));
 			ErrorMessage em = (ErrorMessage) this.errors.receive(10000);
 			assertNotNull(em);
 			assertThat(em.getPayload().getMessage().toLowerCase(),
@@ -263,7 +261,7 @@ public Integer sum(List<Integer> numbers) {
 	}
 
 	private Map<String, Object> stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) {
-		Map<String, Object> headers = new HashMap<String, Object>();
+		Map<String, Object> headers = new HashMap<>();
 		headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber);
 		headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize);
 		headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId);