GH-3797: Improve batch processing in the framework (#3820)

* GH-3797: Improve batch processing in the framework

Fixes https://github.com/spring-projects/spring-integration/issues/3797

* Handle `Message` items of the `Iterable` payload properly in the `JdbcMessageHandler`.
Otherwise, they've been wrapped into an extra `Message`
* Produce a single message with a `Collection<Message<?>>` payload in the `AggregatingMessageHandler`
when the `getOutputProcessor()` is not an instance of `SimpleMessageGroupProcessor`
* Mention these changes in docs
* Point to the error handling sample from docs

* * Fix language in docs

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -57,14 +57,14 @@ import org.springframework.util.LinkedCaseInsensitiveMap;
  * properties can be referred to by name in the query string, e.g.
  *
  * <pre class="code">
- * INSERT INTO FOOS (MESSAGE_ID, PAYLOAD) VALUES (:headers[id], :payload)
+ * INSERT INTO ITEMS (MESSAGE_ID, PAYLOAD) VALUES (:headers[id], :payload)
  * </pre>
  *
  * <p>
  * When a message payload is an instance of {@link Iterable}, a
  * {@link NamedParameterJdbcOperations#batchUpdate(String, SqlParameterSource[])} is performed, where each
  * {@link SqlParameterSource} instance is based on items wrapped into an internal {@link Message} implementation with
- * headers from the request message.
+ * headers from the request message. The item is wrapped only if it is not a {@link Message} already.
  * <p>
  * When a {@link #preparedStatementSetter} is configured, it is applied for each item in the appropriate
  * {@link JdbcOperations#batchUpdate(String, BatchPreparedStatementSetter)} function.
@@ -219,19 +219,7 @@ public class JdbcMessageHandler extends AbstractMessageHandler {
 			if (message.getPayload() instanceof Iterable) {
 				Stream<? extends Message<?>> messageStream =
 						StreamSupport.stream(((Iterable<?>) message.getPayload()).spliterator(), false)
-								.map(payload -> new Message<Object>() {
-
-									@Override
-									public Object getPayload() {
-										return payload;
-									}
-
-									@Override
-									public MessageHeaders getHeaders() {
-										return message.getHeaders();
-									}
-
-								});
+								.map(payload -> payloadToMessage(payload, message.getHeaders()));
 
 				int[] updates;
 
@@ -288,4 +276,24 @@ public class JdbcMessageHandler extends AbstractMessageHandler {
 		}
 	}
 
+	private static Message<?> payloadToMessage(Object payload, MessageHeaders messageHeaders) {
+		if (payload instanceof Message) {
+			return (Message<?>) payload;
+		}
+
+		return new Message<>() {
+
+			@Override
+			public Object getPayload() {
+				return payload;
+			}
+
+			@Override
+			public MessageHeaders getHeaders() {
+				return messageHeaders;
+			}
+
+		};
+	}
+
 }

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageHandlerIntegrationTests.java

@@ -17,11 +17,13 @@
 package org.springframework.integration.jdbc;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
 
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.IntStream;
 
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
@@ -29,6 +31,7 @@ import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
@@ -108,6 +111,38 @@ public class JdbcMessageHandlerIntegrationTests {
 		assertThat(foos.get(2).get("NAME")).isEqualTo("foo3");
 	}
 
+	@Test
+	public void testInsertBatchOfMessages() {
+		JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate,
+				"insert into foos (id, status, name) values (:id, 0, :payload)");
+		ExpressionEvaluatingSqlParameterSourceFactory sqlParameterSourceFactory =
+				new ExpressionEvaluatingSqlParameterSourceFactory();
+		sqlParameterSourceFactory.setParameterExpressions(Map.of("id", "headers.id", "payload", "payload"));
+		sqlParameterSourceFactory.setBeanFactory(mock(BeanFactory.class));
+		handler.setSqlParameterSourceFactory(sqlParameterSourceFactory);
+		handler.afterPropertiesSet();
+
+		List<GenericMessage<String>> payload =
+				IntStream.range(1, 4)
+						.mapToObj(i -> "Item" + i)
+						.map(GenericMessage::new)
+						.toList();
+
+		handler.handleMessage(new GenericMessage<>(payload));
+
+		List<Map<String, Object>> foos = jdbcTemplate.queryForList("SELECT * FROM FOOS ORDER BY NAME");
+
+		assertThat(foos.size()).isEqualTo(3);
+
+		assertThat(foos.get(0).get("NAME")).isEqualTo("Item1");
+		assertThat(foos.get(1).get("NAME")).isEqualTo("Item2");
+		assertThat(foos.get(2).get("NAME")).isEqualTo("Item3");
+
+		assertThat(foos.get(0).get("ID"))
+				.isNotEqualTo(foos.get(0).get("NAME"))
+				.isEqualTo(payload.get(0).getHeaders().getId().toString());
+	}
+
 	@Test
 	public void testInsertWithMessagePreparedStatementSetter() {
 		JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate,