Add option for MutableMB to not generate headers

Since `MutableMessage` can be mutated, the `ID` and `TIMESTAMP` headers
can be added lately.
Or there might be some scenarios which don't care about those headers at
all.
Plus we get a performance gain when we skip their generation

Coerce the `null` cast to the proper target ctor

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

@@ -37,6 +37,7 @@ import org.springframework.util.ObjectUtils;
  * @author Artem Bilan
  * @author Stuart Williams
  * @author David Turanski
+ *
  * @since 4.0
  *
  */
@@ -49,16 +50,19 @@ public class MutableMessage<T> implements Message<T>, Serializable {
 	private final MutableMessageHeaders headers;
 
 	public MutableMessage(T payload) {
-		this(payload, null);
+		this(payload, (Map<String, Object>) null);
 	}
 
 	public MutableMessage(T payload, Map<String, Object> headers) {
-		Assert.notNull(payload, "payload must not be null");
-		this.payload = payload;
-
-		this.headers = new MutableMessageHeaders(headers);
+		this(payload, new MutableMessageHeaders(headers));
 	}
 
+	protected MutableMessage(T payload, MutableMessageHeaders headers) {
+		Assert.notNull(payload, "payload must not be null");
+		Assert.notNull(headers, "headers must not be null");
+		this.payload = payload;
+		this.headers = headers;
+	}
 
 	@Override
 	public MutableMessageHeaders getHeaders() {

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

@@ -23,6 +23,7 @@ import java.util.Map.Entry;
 
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
 import org.springframework.util.PatternMatchUtils;
 import org.springframework.util.StringUtils;
@@ -64,6 +65,39 @@ public final class MutableMessageBuilder<T> extends AbstractIntegrationMessageBu
 		return this.headers;
 	}
 
+	/**
+	 * Create a builder for a new {@link Message} instance with the provided payload.
+	 * @param payload the payload for the new message
+	 * @param <T> The type of the payload.
+	 * @return A MutableMessageBuilder.
+	 */
+	public static <T> MutableMessageBuilder<T> withPayload(T payload) {
+		return withPayload(payload, true);
+	}
+
+	/**
+	 * Create a builder for a new {@link Message} instance with the provided payload.
+	 * The {@code generateHeaders} flag allows to disable {@link MessageHeaders#ID}
+	 * and {@link MessageHeaders#TIMESTAMP} headers generation.
+	 * @param payload the payload for the new message
+	 * @param generateHeaders whether generate {@link MessageHeaders#ID}
+	 * and {@link MessageHeaders#TIMESTAMP} headers
+	 * @param <T> The type of the payload.
+	 * @return A MutableMessageBuilder.
+	 * @since 5.0
+	 */
+	public static <T> MutableMessageBuilder<T> withPayload(T payload, boolean generateHeaders) {
+		MutableMessage<T> message;
+		if (generateHeaders) {
+			message = new MutableMessage<>(payload);
+		}
+		else {
+			message = new MutableMessage<>(payload, new MutableMessageHeaders(null, MessageHeaders.ID_VALUE_NONE, -1L));
+		}
+
+		return fromMessage(message);
+	}
+
 	/**
 	 * Create a builder for a new {@link Message} instance pre-populated with all of the headers copied from the
 	 * provided message. The payload of the provided Message will also be used as the payload for the new message.
@@ -76,16 +110,6 @@ public final class MutableMessageBuilder<T> extends AbstractIntegrationMessageBu
 		return new MutableMessageBuilder<T>(message);
 	}
 
-	/**
-	 * Create a builder for a new {@link Message} instance with the provided payload.
-	 * @param payload the payload for the new message
-	 * @param <T> The type of the payload.
-	 * @return A MessageBuilder.
-	 */
-	public static <T> MutableMessageBuilder<T> withPayload(T payload) {
-		return new MutableMessageBuilder<T>(new MutableMessage<T>(payload));
-	}
-
 	@Override
 	public AbstractIntegrationMessageBuilder<T> setHeader(String headerName, Object headerValue) {
 		Assert.notNull(headerName, "'headerName' must not be null");

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

@@ -19,6 +19,7 @@ package org.springframework.integration.support;
 import java.util.Map;
 import java.util.UUID;
 
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageHeaders;
 
 
@@ -36,8 +37,8 @@ public class MutableMessageHeaders extends MessageHeaders {
 
 	private static final long serialVersionUID = 3084692953798643018L;
 
-	public MutableMessageHeaders(Map<String, Object> headers) {
-		super(headers,
+	public MutableMessageHeaders(@Nullable Map<String, Object> headers) {
+		this(headers,
 				(headers != null ?
 						(UUID) headers.get(MessageHeaders.ID)
 						: null),
@@ -46,6 +47,10 @@ public class MutableMessageHeaders extends MessageHeaders {
 						: null));
 	}
 
+	protected MutableMessageHeaders(@Nullable Map<String, Object> headers, @Nullable UUID id, @Nullable Long timestamp) {
+		super(headers, id, timestamp);
+	}
+
 	@Override
 	protected Map<String, Object> getRawHeaders() {
 		return super.getRawHeaders();