spring-projectsGH-3785: Close stream for persistent collection

Fixes spring-projects#3785

* Fix `CollectionArgumentResolver` and `PayloadsArgumentResolver` to
close the `Stream` of message after its usage
* Rework `AbstractKeyValueMessageStore.removeMessagesFromGroup()`
to iterate input collection of messages not its stream to avoid
the mentioned problem

**Cherry-pick to `5.5.x`**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/handler/support/CollectionArgumentResolver.java

@@ -20,6 +20,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.convert.TypeDescriptor;
@@ -71,17 +72,16 @@ public Object resolveArgument(MethodParameter parameter, Message<?> message) {
 		if (this.canProcessMessageList) {
 			Assert.state(value instanceof Collection,
 					"This Argument Resolver only supports messages with a payload of Collection<Message<?>>, "
-					+ "payload is: " + value.getClass());
+							+ "payload is: " + value.getClass());
 
 			Collection<Message<?>> messages = (Collection<Message<?>>) value;
 
-			if (Message.class.isAssignableFrom(parameter.nested().getNestedParameterType())) {
-				value = messages;
-			}
-			else {
-				value = messages.stream()
-						.map(Message::getPayload)
-						.collect(Collectors.toList());
+			if (!Message.class.isAssignableFrom(parameter.nested().getNestedParameterType())) {
+				try (Stream<Message<?>> messageStream = messages.stream()) {
+					value = messageStream
+							.map(Message::getPayload)
+							.collect(Collectors.toList());
+				}
 			}
 		}
 

spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadsArgumentResolver.java

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.convert.TypeDescriptor;
@@ -78,15 +79,17 @@ public Object resolveArgument(MethodParameter parameter, Message<?> message) {
 			return evaluateExpression(expression, messages, parameter.getParameterType());
 		}
 		else {
-			List<?> payloads = messages.stream()
-					.map(Message::getPayload)
-					.collect(Collectors.toList());
+			try (Stream<Message<?>> messageStream = messages.stream()) {
+				List<?> payloads = messageStream
+						.map(Message::getPayload)
+						.collect(Collectors.toList());
+				return getEvaluationContext()
+						.getTypeConverter()
+						.convertValue(payloads,
+								TypeDescriptor.forObject(payloads),
+								TypeDescriptor.valueOf(parameter.getParameterType()));
+			}
 
-			return getEvaluationContext()
-					.getTypeConverter()
-					.convertValue(payloads,
-							TypeDescriptor.forObject(payloads),
-							TypeDescriptor.valueOf(parameter.getParameterType()));
 		}
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java

@@ -244,17 +244,17 @@ public void removeMessagesFromGroup(Object groupId, Collection<Message<?>> messa
 			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
 			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
 
-			List<UUID> ids =
-					messages.stream()
-							.map(messageToRemove -> messageToRemove.getHeaders().getId())
-							.collect(Collectors.toList());
+			List<UUID> ids = new ArrayList<>();
+			for (Message<?> messageToRemove : messages) {
+				ids.add(messageToRemove.getHeaders().getId());
+			}
 
 			messageGroupMetadata.removeAll(ids);
 
-			List<Object> messageIds =
-					ids.stream()
-							.map(id -> this.messagePrefix + id)
-							.collect(Collectors.toList());
+			List<Object> messageIds = new ArrayList<>();
+			for (UUID id : ids) {
+				messageIds.add(this.messagePrefix + id);
+			}
 
 			doRemoveAll(messageIds);